fix: 回退迁移0010到原内容, 新增0013, 删除已应用迁移记录重新对齐

迁移 0010/0011 在首次 service 运行后被修改,导致 sqlx 校验和不匹配。
0010 回退到仅 ALTER COLUMN DROP NOT NULL(已幂等执行)。
新增 0013 处理全新数据库缺少 kind/entry 列的问题。
已通过 psql 删除 _sqlx_migrations 中的 0010/0011 记录,
下次 service 启动将重新应用(均为幂等操作)。
This commit is contained in:
2026-06-02 20:10:25 +08:00
parent dcfaf47809
commit 031f160393
5 changed files with 101 additions and 70 deletions
+2
View File
@@ -24,3 +24,5 @@ qweather/
# 操作系统
.DS_Store
Thumbs.db
read.md
@@ -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;
@@ -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;
+3 -12
View File
@@ -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<Self, String> {
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(())
}
}
+86 -37
View File
@@ -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<dyn Fn(&str, &str) -> Pin<Box<dyn Future<Output = Result<String, String>> + Send>> + Send + Sync>;
pub type ToolExecutor = Arc<
dyn Fn(&str, &str) -> Pin<Box<dyn Future<Output = Result<String, String>> + Send>>
+ Send
+ Sync,
>;
/// 摘要生成器:接收要摘要的消息文本 → 返回 LLM 生成的摘要
pub type Summarizer =
Arc<dyn Fn(String) -> Pin<Box<dyn Future<Output = Result<String, String>> + Send>> + Send + Sync>;
pub type Summarizer = Arc<
dyn Fn(String) -> Pin<Box<dyn Future<Output = Result<String, String>> + 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<Mutex<ChatSession>> {
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<Mutex<ChatSession>> { 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::<Vec<_>>().join(", ")
calls
.iter()
.map(|c| format!("{}({:.80})", c.function.name, c.function.arguments))
.collect::<Vec<_>>()
.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 {
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<String, String> {
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<ToolCall>> { 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. \