From e20cd19e575a0f77aa293f9ae999bbb3b055b217 Mon Sep 17 00:00:00 2001 From: wunianxiao Date: Tue, 16 Jun 2026 22:53:19 +0800 Subject: [PATCH] =?UTF-8?q?chore:=20=E5=A2=9E=E5=BC=BA=E6=97=A5=E5=BF=97?= =?UTF-8?q?=E8=BF=BD=E8=B8=AA=E4=B8=8E=E4=BB=A3=E7=A0=81=E6=A0=BC=E5=BC=8F?= =?UTF-8?q?=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - daemon: 添加 LLM 回合开始/结束标记日志，工具执行结果追踪 - daemon: 发送消费者日志增强（显示完整文本、截断用户 ID） - conversation: 移除日志截断，输出完整工具调用参数和结果 - deepseek: 代码格式化，预留请求/响应摘要日志 - 删除 systemd/ias.service --- src/daemon.rs | 240 +++++++++++++++++++++++----------------- src/llm/conversation.rs | 4 +- src/llm/deepseek.rs | 61 ++++++---- systemd/ias.service | 17 --- 4 files changed, 181 insertions(+), 141 deletions(-) delete mode 100644 systemd/ias.service diff --git a/src/daemon.rs b/src/daemon.rs index 10c47a4..0131eed 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -109,9 +109,9 @@ pub async fn run(database: &Arc, memory_store: &Arc) { info!("Daemon 已启动"); // 4-6. 审批管理器、工具列表、模型配置 - let approval_manager = Arc::new(ApprovalManager::new( - Some(Arc::new(database.pool().clone())), - )); + let approval_manager = Arc::new(ApprovalManager::new(Some(Arc::new( + database.pool().clone(), + )))); let tools_list = build_tools_list(); let system_prompt = std::env::var("WEIXIN_LLM_SYSTEM_PROMPT").unwrap_or_else(|_| String::new()); let model = std::env::var("DEEPSEEK_MODEL").unwrap_or_else(|_| "deepseek-v4-flash".to_string()); @@ -158,28 +158,28 @@ pub async fn run(database: &Arc, memory_store: &Arc) { let sched_enqueue = enqueue_handle.clone(); tokio::spawn(async move { let scheduler = crate::scheduler::Scheduler::new(Some(Arc::new(sched_db))); - scheduler - .run(move |to: &str, name: &str, msg: &str| { - let eq = sched_enqueue.clone(); - let uid = to.to_string(); - let text = msg.to_string(); - let tn = name.to_string(); - tokio::spawn(async move { - eq.enqueue(PipelineMessage { - channel: ChannelId::wechat(uid), - correlation_id: Uuid::new_v4(), - kind: MessageKind::ScheduledTask { - text, - task_name: tn, - }, - }) - .await; - }); - Ok(()) - }) - .await; - }); - info!("定时任务调度器已启动"); + scheduler + .run(move |to: &str, name: &str, msg: &str| { + let eq = sched_enqueue.clone(); + let uid = to.to_string(); + let text = msg.to_string(); + let tn = name.to_string(); + tokio::spawn(async move { + eq.enqueue(PipelineMessage { + channel: ChannelId::wechat(uid), + correlation_id: Uuid::new_v4(), + kind: MessageKind::ScheduledTask { + text, + task_name: tn, + }, + }) + .await; + }); + Ok(()) + }) + .await; + }); + info!("定时任务调度器已启动"); // 12. QueueRunner let runner_handle = tokio::spawn(async move { @@ -349,8 +349,6 @@ async fn llm_consumer_loop( context_token, approved_tool, } => { - info!("LLM Consumer: user={} text={:.60}", user_id, text); - // 加载数据 ctx.memory_store.load_if_needed(&user_id).await; let history = load_history(&ctx.db, &user_id).await; @@ -399,45 +397,46 @@ async fn llm_consumer_loop( let session = conv.session(); let approved_tool = approved_tool.clone(); - let executor: ToolExecutor = Arc::new(move |name: &str, args_json: &str, tc_id: &str| { - let eq = eq.clone(); - let uid = uid.clone(); - let corr = corr; - let session = session.clone(); - let n = name.to_string(); - let args = args_json.to_string(); - let is_approved = approved_tool.as_ref() == Some(&n); - let tool_call_id = tc_id.to_string(); + let executor: ToolExecutor = + Arc::new(move |name: &str, args_json: &str, tc_id: &str| { + let eq = eq.clone(); + let uid = uid.clone(); + let corr = corr; + let session = session.clone(); + let n = name.to_string(); + let args = args_json.to_string(); + let is_approved = approved_tool.as_ref() == Some(&n); + let tool_call_id = tc_id.to_string(); - Box::pin(async move { - let ch = ChannelId::wechat(&uid); - let sid = { session.lock().await.session_id }; - if is_approved { - eq.enqueue(PipelineMessage::approved_tool_call( - ch, - corr, - sid, - &tool_call_id, - &n, - &args, - None, - )) - .await; - } else { - eq.enqueue(PipelineMessage::tool_call( - ch, - corr, - sid, - &tool_call_id, - &n, - &args, - None, - )) - .await; - } - Ok(format!("{}{}", ASYNC_MARKER, n)) - }) - }); + Box::pin(async move { + let ch = ChannelId::wechat(&uid); + let sid = { session.lock().await.session_id }; + if is_approved { + eq.enqueue(PipelineMessage::approved_tool_call( + ch, + corr, + sid, + &tool_call_id, + &n, + &args, + None, + )) + .await; + } else { + eq.enqueue(PipelineMessage::tool_call( + ch, + corr, + sid, + &tool_call_id, + &n, + &args, + None, + )) + .await; + } + Ok(format!("{}{}", ASYNC_MARKER, n)) + }) + }); conv.set_tool_executor(executor); // 缓存 Conversation（记录创建时间用于 TTL） @@ -465,11 +464,6 @@ async fn llm_consumer_loop( result, context_token, } => { - info!( - "LLM Consumer: 工具结果回喂 user={} tool={}", - user_id, tool_name - ); - // 从缓存恢复会话，注入工具结果并恢复 LLM 循环 let should_resume = { let mut map = sessions.lock().await; @@ -532,6 +526,8 @@ async fn run_llm_round( sessions: &Arc>>, enqueue: &EnqueueHandle, ) { + info!("══════════════════════════════════════════════════"); + info!("🔄 LLM 回合开始 user={} text={}", user_id, text); let result = { let map = sessions.lock().await; let conv = match map.get(&correlation_id) { @@ -575,6 +571,8 @@ async fn resume_llm_round( sessions: &Arc>>, enqueue: &EnqueueHandle, ) { + info!("══════════════════════════════════════════════════"); + info!("🔄 LLM 恢复 user={}", user_id); let result = { let map = sessions.lock().await; let conv = match map.get(&correlation_id) { @@ -636,6 +634,7 @@ async fn handle_chat_result( enqueue.enqueue(msg).await; } sessions.lock().await.remove(&correlation_id); + info!("✅ LLM 回合结束 user={}", user_id); } // ─── Tool Consumer ─── @@ -660,7 +659,10 @@ async fn tool_consumer_loop( context_token, approved, } => { - info!("Tool Consumer: user={} tool={}", user_id, tool_name); + info!( + "Tool Consumer: user={} tool={} args={}", + user_id, tool_name, arguments + ); // 高风险工具 → 审批（除非已被用户批准） if !approved && crate::tools::is_high_risk(&tool_name) { @@ -734,7 +736,9 @@ fn execute_tool<'a>( // 上下文工具 match name { "read_memories" => { - return ctx.memory_store.read_for(user_id, None).await; + let result = ctx.memory_store.read_for(user_id, None).await; + tracing::info!("🔧 execute_tool: read_memories → {}", result); + return result; } "write_memory" => { let params: serde_json::Value = serde_json::from_str(arguments).unwrap_or_default(); @@ -743,7 +747,9 @@ fn execute_tool<'a>( return "请提供 content 参数".to_string(); } ctx.memory_store.write_for(user_id, content).await; - return format!("已添加长期记忆: {}", content); + let result = format!("已添加长期记忆: {}", content); + tracing::info!("🔧 execute_tool: write_memory → {}", result); + return result; } "delete_memory" => { let params: serde_json::Value = serde_json::from_str(arguments).unwrap_or_default(); @@ -754,7 +760,9 @@ fn execute_tool<'a>( if memory_id <= 0 { return "请提供有效的 memory_id 参数".to_string(); } - return ctx.memory_store.delete_for(user_id, memory_id).await; + let result = ctx.memory_store.delete_for(user_id, memory_id).await; + tracing::info!("🔧 execute_tool: delete_memory({}) → {}", memory_id, result); + return result; } "update_memory" => { let params: serde_json::Value = serde_json::from_str(arguments).unwrap_or_default(); @@ -769,26 +777,35 @@ fn execute_tool<'a>( if content.is_empty() { return "请提供 content 参数".to_string(); } - return ctx + let result = ctx .memory_store .update_for(user_id, memory_id, content) .await; + tracing::info!("🔧 execute_tool: update_memory({}) → {}", memory_id, result); + return result; } "read_summaries" => { let sums = load_summaries(&ctx.db, user_id).await; - if sums.is_empty() { - return "暂无历史摘要".to_string(); - } - return sums - .iter() - .enumerate() - .map(|(i, s)| format!("{}. {}", i + 1, s)) - .collect::>() - .join("\n---\n"); + let result = if sums.is_empty() { + "暂无历史摘要".to_string() + } else { + sums.iter() + .enumerate() + .map(|(i, s)| format!("{}. {}", i + 1, s)) + .collect::>() + .join("\n---\n") + }; + tracing::info!("🔧 execute_tool: read_summaries → {}条", sums.len()); + return result; } "query_capabilities" => { let specs = crate::tools::specs(); - return crate::tools::build_capability_guide(&specs); + let result = crate::tools::build_capability_guide(&specs); + tracing::info!( + "🔧 execute_tool: query_capabilities → {}个工具", + specs.len() + ); + return result; } "call_capability" => { // 解包 {name, prompt} → 提取目标工具名和参数，递归派发 @@ -807,6 +824,11 @@ fn execute_tool<'a>( } let target_args = serde_json::to_string(&crate::tools::unpack_call_params(&cp)) .unwrap_or_default(); + tracing::info!( + "🔧 execute_tool: call_capability → {} args={}", + target_name, + target_args + ); // 递归派发 return execute_tool(&target_name, &target_args, ctx, user_id).await; } @@ -815,8 +837,19 @@ fn execute_tool<'a>( // 内置工具注册表 match crate::tools::execute(name, arguments).await { - Some(r) => r.output, - None => format!("未知工具: {}", name), + Some(r) => { + tracing::info!( + "🔧 execute_tool: {} → success={} output={}", + name, + r.success, + r.output + ); + r.output + } + None => { + tracing::warn!("🔧 execute_tool: 未知工具 {}", name); + format!("未知工具: {}", name) + } } }) } @@ -836,7 +869,12 @@ async fn send_consumer_loop( context_token, } => { let user_id = &msg.channel.user_id; - info!("Send Consumer: user={} text_len={}", user_id, text.len()); + info!( + "Send Consumer: user={} text_len={} text={}", + user_id, + text.len(), + text + ); match ctx .client @@ -844,18 +882,22 @@ async fn send_consumer_loop( .await { Ok(sent_id) => { - info!("回复成功 user={} msg_id={}", user_id, sent_id); + info!( + "回复成功 user={} msg_id={}", + user_id.chars().take(5).collect::(), + sent_id + ); let _ = crate::db::models::insert_chat_record( - ctx.db.pool(), - "outbound", - user_id, - &ctx.account_id, - &text, - "llm", - context_token.as_deref(), - &sent_id, - ) - .await; + ctx.db.pool(), + "outbound", + user_id, + &ctx.account_id, + &text, + "llm", + context_token.as_deref(), + &sent_id, + ) + .await; } Err(e) => error!("发送回复失败: {}", e), } diff --git a/src/llm/conversation.rs b/src/llm/conversation.rs index 9d62844..c64d764 100644 --- a/src/llm/conversation.rs +++ b/src/llm/conversation.rs @@ -313,7 +313,7 @@ impl Conversation { calls.len(), calls .iter() - .map(|c| format!("{}({:.80})", c.function.name, c.function.arguments)) + .map(|c| format!("{}({})", c.function.name, c.function.arguments)) .collect::>() .join(", ") ); @@ -344,7 +344,7 @@ impl Conversation { continue; } - tracing::info!("📦 {} → {:.150}", tc.function.name, result); + tracing::info!("📦 {} → {}", tc.function.name, result); self.session.lock().await.add_tool_result( &tc.id, &tc.function.name, diff --git a/src/llm/deepseek.rs b/src/llm/deepseek.rs index c1fcd22..eb14025 100644 --- a/src/llm/deepseek.rs +++ b/src/llm/deepseek.rs @@ -154,6 +154,13 @@ async fn stream_deepseek( return Err(format!("HTTP {}: {}", status, text)); } + // 记录请求摘要 + // tracing::info!( + // "🌐 LLM 请求: model={} messages_count={}", + // body.get("model").and_then(|v| v.as_str()).unwrap_or("?"), + // body.get("messages").and_then(|v| v.as_array()).map(|a| a.len()).unwrap_or(0), + // ); + let mut full_text = String::new(); let mut full_reasoning = String::new(); let mut usage: Option = None; @@ -216,33 +223,33 @@ async fn stream_deepseek( // 检查最后的 buffer if !buf.trim().is_empty() - && let Some(parsed) = parse_chat_chunk(buf.trim()) { - match parsed { - ParsedChunk::Text(t) => full_text.push_str(&t), - ParsedChunk::ToolCallDelta { - index, - id, - name, - arguments, - } => { - let entry = - tool_call_builders - .entry(index) - .or_insert((None, None, String::new())); - if let Some(call_id) = id { - entry.0 = Some(call_id); - } - if let Some(call_name) = name { - entry.1 = Some(call_name); - } - entry.2.push_str(&arguments); + && let Some(parsed) = parse_chat_chunk(buf.trim()) + { + match parsed { + ParsedChunk::Text(t) => full_text.push_str(&t), + ParsedChunk::ToolCallDelta { + index, + id, + name, + arguments, + } => { + let entry = tool_call_builders + .entry(index) + .or_insert((None, None, String::new())); + if let Some(call_id) = id { + entry.0 = Some(call_id); } - ParsedChunk::Usage(u) => { - usage = Some(u); + if let Some(call_name) = name { + entry.1 = Some(call_name); } - _ => {} + entry.2.push_str(&arguments); } + ParsedChunk::Usage(u) => { + usage = Some(u); + } + _ => {} } + } // 构建 tool_calls let tool_calls: Option> = if tool_call_builders.is_empty() { @@ -267,6 +274,14 @@ async fn stream_deepseek( ) }; + // 记录完整的响应输出 + // tracing::info!( + // "🌐 LLM 响应: text_len={} tool_calls={} usage={:?}", + // full_text.len(), + // tool_calls.as_ref().map(|tc| tc.len()).unwrap_or(0), + // usage + // ); + let _ = tx .send(StreamChunk::Done { text: full_text, diff --git a/systemd/ias.service b/systemd/ias.service deleted file mode 100644 index fc1019b..0000000 --- a/systemd/ias.service +++ /dev/null @@ -1,17 +0,0 @@ -[Unit] -Description=iAs - 微信 AI 智能助手 -After=network.target postgresql.service - -[Service] -Type=simple -User=xiao -WorkingDirectory=/home/xiao/project/iAs -EnvironmentFile=/home/xiao/project/iAs/.env -ExecStart=/home/xiao/project/iAs/target/release/iAs service -Restart=on-failure -RestartSec=10 -StandardOutput=null -StandardError=null - -[Install] -WantedBy=multi-user.target