diff --git a/.gitignore b/.gitignore index 9bb529a..998d50c 100644 --- a/.gitignore +++ b/.gitignore @@ -24,3 +24,5 @@ qweather/ # 操作系统 .DS_Store Thumbs.db + +read.md diff --git a/migrations/20260601000010_fix_llm_usage_legacy_columns.sql b/migrations/20260601000010_fix_llm_usage_legacy_columns.sql index 961a84f..de9f930 100644 --- a/migrations/20260601000010_fix_llm_usage_legacy_columns.sql +++ b/migrations/20260601000010_fix_llm_usage_legacy_columns.sql @@ -1,18 +1,4 @@ --- 修复 llm_usage 表兼容性：旧库有 iPet 遗留的 kind/entry 列 (NOT NULL)，新库没有 --- kind/entry 保留为兼容字段，新代码写入默认值 ("llm_call" / null) --- 1) 添加列（如果不存在） -ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS kind TEXT; -ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS entry JSONB; --- 2) 只在列存在且 NOT NULL 时解除约束 -DO $$ -BEGIN - IF EXISTS (SELECT 1 FROM information_schema.columns - WHERE table_name = 'llm_usage' AND column_name = 'kind' AND is_nullable = 'NO') THEN - ALTER TABLE llm_usage ALTER COLUMN kind DROP NOT NULL; - END IF; - IF EXISTS (SELECT 1 FROM information_schema.columns - WHERE table_name = 'llm_usage' AND column_name = 'entry' AND is_nullable = 'NO') THEN - ALTER TABLE llm_usage ALTER COLUMN entry DROP NOT NULL; - END IF; -END; -$$; +-- 移除 llm_usage 表中旧 iPet 遗留列的 NOT NULL 约束 +-- kind/entry 是旧格式字段，新代码不使用它们 +ALTER TABLE llm_usage ALTER COLUMN kind DROP NOT NULL; +ALTER TABLE llm_usage ALTER COLUMN entry DROP NOT NULL; \ No newline at end of file diff --git a/migrations/20260601000013_fix_llm_usage_kind_entry_newdb.sql b/migrations/20260601000013_fix_llm_usage_kind_entry_newdb.sql new file mode 100644 index 0000000..276885b --- /dev/null +++ b/migrations/20260601000013_fix_llm_usage_kind_entry_newdb.sql @@ -0,0 +1,3 @@ +-- 确保 llm_usage 表在全新数据库上也有 kind/entry 列（迁移 0010 仅解除了旧库的 NOT NULL） +ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS kind TEXT; +ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS entry JSONB; diff --git a/src/db/mod.rs b/src/db/mod.rs index 991b36e..e915a5f 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -1,7 +1,7 @@ pub mod models; -use sqlx::postgres::PgPoolOptions; use sqlx::PgPool; +use sqlx::postgres::PgPoolOptions; use std::time::Duration; /// 数据库管理器 @@ -12,8 +12,8 @@ pub struct Database { impl Database { /// 从 DATABASE_URL 环境变量连接并运行迁移 pub async fn connect() -> Result { - let database_url = - std::env::var("DATABASE_URL").map_err(|_| "请设置 DATABASE_URL 环境变量".to_string())?; + let database_url = std::env::var("DATABASE_URL") + .map_err(|_| "请设置 DATABASE_URL 环境变量".to_string())?; let pool = PgPoolOptions::new() .max_connections(5) @@ -37,13 +37,4 @@ impl Database { pub fn pool(&self) -> &PgPool { &self.pool } - - /// 检查数据库是否可用 - pub async fn health_check(&self) -> Result<(), String> { - sqlx::query("SELECT 1") - .execute(&self.pool) - .await - .map_err(|e| format!("数据库健康检查失败: {}", e))?; - Ok(()) - } } diff --git a/src/llm/conversation.rs b/src/llm/conversation.rs index e5b746a..fa7efa8 100644 --- a/src/llm/conversation.rs +++ b/src/llm/conversation.rs @@ -1,18 +1,22 @@ use crate::context::builder; use crate::context::types::ChatSession; -use crate::llm::provider::{create_provider, BoxedProvider, StreamReceiver}; +use crate::llm::provider::{BoxedProvider, StreamReceiver, create_provider}; use crate::llm::types::{ConversationConfig, Message, StreamChunk, ToolCall, Usage}; use std::future::Future; use std::pin::Pin; use std::sync::Arc; use tokio::sync::Mutex; -pub type ToolExecutor = - Arc Pin> + Send>> + Send + Sync>; +pub type ToolExecutor = Arc< + dyn Fn(&str, &str) -> Pin> + Send>> + + Send + + Sync, +>; /// 摘要生成器：接收要摘要的消息文本 → 返回 LLM 生成的摘要 -pub type Summarizer = - Arc Pin> + Send>> + Send + Sync>; +pub type Summarizer = Arc< + dyn Fn(String) -> Pin> + Send>> + Send + Sync, +>; pub struct Conversation { config: ConversationConfig, @@ -32,21 +36,24 @@ impl Conversation { }) } - pub fn with_provider(config: ConversationConfig, provider: BoxedProvider) -> Self { - Self { - config, - provider, - session: Arc::new(Mutex::new(ChatSession::new())), - tool_executor: None, - } + pub fn set_tool_executor(&mut self, executor: ToolExecutor) { + self.tool_executor = Some(executor); + } + pub fn session(&self) -> Arc> { + Arc::clone(&self.session) + } + pub fn model(&self) -> &str { + &self.config.model + } + pub fn provider_name(&self) -> &str { + self.provider.name() + } + pub fn spec(&self) -> &ConversationConfig { + &self.config + } + pub fn summarizer(&self) -> Summarizer { + self.create_summarizer() } - - pub fn set_tool_executor(&mut self, executor: ToolExecutor) { self.tool_executor = Some(executor); } - pub fn session(&self) -> Arc> { Arc::clone(&self.session) } - pub fn model(&self) -> &str { &self.config.model } - pub fn provider_name(&self) -> &str { self.provider.name() } - pub fn spec(&self) -> &ConversationConfig { &self.config } - pub fn summarizer(&self) -> Summarizer { self.create_summarizer() } /// 创建 LLM 摘要生成器（用当前 provider 做非流式调用） pub fn create_summarizer(&self) -> Summarizer { @@ -54,15 +61,22 @@ impl Conversation { Arc::new(move |prompt: String| { let provider = provider.clone(); Box::pin(async move { - let msgs = vec![Message::system("你是一个对话摘要助手，用中文输出简洁摘要。"), Message::user(prompt)]; - let mut rx = provider.chat_stream(&ConversationConfig::default(), &msgs).await + let msgs = vec![ + Message::system("你是一个对话摘要助手，用中文输出简洁摘要。"), + Message::user(prompt), + ]; + let mut rx = provider + .chat_stream(&ConversationConfig::default(), &msgs) + .await .map_err(|e| format!("摘要请求失败: {}", e))?; let mut text = String::new(); while let Some(chunk) = rx.recv().await { match chunk { StreamChunk::Text(t) => text.push_str(&t), StreamChunk::Done { text: full, .. } => { - if !full.is_empty() { text = full; } + if !full.is_empty() { + text = full; + } break; } StreamChunk::Error(e) => return Err(e), @@ -90,7 +104,10 @@ impl Conversation { let rx = self.provider.chat_stream(&self.config, &messages).await?; Ok(ChatHandle { - rx: Some(rx), full_text: String::new(), tool_calls: None, usage: None, + rx: Some(rx), + full_text: String::new(), + tool_calls: None, + usage: None, session: Arc::clone(&self.session), }) } @@ -132,8 +149,15 @@ impl Conversation { while let Some(chunk) = rx.recv().await { match chunk { StreamChunk::Text(t) => full_text.push_str(&t), - StreamChunk::Done { text, tool_calls: tc, usage, .. } => { - if !text.is_empty() { full_text = text; } + StreamChunk::Done { + text, + tool_calls: tc, + usage, + .. + } => { + if !text.is_empty() { + full_text = text; + } tool_calls = tc; last_usage = usage; break; @@ -149,21 +173,34 @@ impl Conversation { tracing::info!( "🔧 LLM 请求 {} 个工具: {}", calls.len(), - calls.iter().map(|c| format!("{}({:.80})", c.function.name, c.function.arguments)).collect::>().join(", ") + calls + .iter() + .map(|c| format!("{}({:.80})", c.function.name, c.function.arguments)) + .collect::>() + .join(", ") ); - self.session.lock().await.add_assistant_tool_calls(calls.clone()); + self.session + .lock() + .await + .add_assistant_tool_calls(calls.clone()); for tc in &calls { let result = match &self.tool_executor { - Some(exec) => match exec(&tc.function.name, &tc.function.arguments).await { - Ok(r) => r, - Err(e) => format!("执行失败: {}", e), - }, + Some(exec) => { + match exec(&tc.function.name, &tc.function.arguments).await { + Ok(r) => r, + Err(e) => format!("执行失败: {}", e), + } + } None => "工具执行器未配置".to_string(), }; tracing::info!("📦 {} → {:.150}", tc.function.name, result); - self.session.lock().await.add_tool_result(&tc.id, &tc.function.name, &result); + self.session.lock().await.add_tool_result( + &tc.id, + &tc.function.name, + &result, + ); } continue; } @@ -181,7 +218,8 @@ impl Conversation { let estimated: u32 = recent.iter().map(|m| builder::estimate_tokens(m)).sum(); if estimated > s.token_budget { drop(s); - builder::trigger_overflow_summary(&self.session, Some(&self.summarizer())).await; + builder::trigger_overflow_summary(&self.session, Some(&self.summarizer())) + .await; tracing::info!("📦 上下文超预算，已触发溢出摘要"); } } @@ -206,8 +244,15 @@ impl ChatHandle { let chunk = self.rx.as_mut()?.recv().await; match &chunk { Some(StreamChunk::Text(t)) => self.full_text.push_str(t), - Some(StreamChunk::Done { text, tool_calls, usage, .. }) => { - if !text.is_empty() { self.full_text = text.clone(); } + Some(StreamChunk::Done { + text, + tool_calls, + usage, + .. + }) => { + if !text.is_empty() { + self.full_text = text.clone(); + } self.tool_calls = tool_calls.clone(); self.usage = usage.clone(); if !self.full_text.is_empty() { @@ -216,7 +261,9 @@ impl ChatHandle { } self.rx = None; } - Some(StreamChunk::Error(_)) => { self.rx = None; } + Some(StreamChunk::Error(_)) => { + self.rx = None; + } _ => {} } chunk @@ -225,14 +272,16 @@ impl ChatHandle { /// 消费所有流式块，返回完整文本 pub async fn consume(&mut self) -> Result { while let Some(chunk) = self.next_chunk().await { - if let StreamChunk::Error(e) = chunk { return Err(e); } + if let StreamChunk::Error(e) = chunk { + return Err(e); + } } Ok(self.full_text.clone()) } - pub fn current_text(&self) -> &str { &self.full_text } - pub fn get_tool_calls(&self) -> Option<&Vec> { self.tool_calls.as_ref() } - pub fn get_usage(&self) -> Option<&Usage> { self.usage.as_ref() } + pub fn get_usage(&self) -> Option<&Usage> { + self.usage.as_ref() + } } pub const DEFAULT_SYSTEM_PROMPT: &str = "You are a concise and helpful assistant replying in Chinese. \