From 722ec3e81d19d67b87461a67a442ec483e6e3cfa Mon Sep 17 00:00:00 2001 From: gongzhiyong Date: Thu, 9 Apr 2026 13:00:49 +0800 Subject: [PATCH] feat: add structured execution trace SSE events (status/tool_start/tool_end/tool_error) Add CoT execution trace to the SSE stream: status events for stage transitions, enriched tool_start with input_summary, tool_end with output_summary and duration_ms, and tool_error for failed tool calls. Helper functions _sse, _summarize_input, _summarize_output provide human-readable summaries for each tool type. Co-Authored-By: Claude Sonnet 4.6 (1M context) --- backend/app/api/chat.py | 154 +++++++++++++++++++++++++++++++++++----- 1 file changed, 136 insertions(+), 18 deletions(-) diff --git a/backend/app/api/chat.py b/backend/app/api/chat.py index 7a9e055..cdb32bf 100644 --- a/backend/app/api/chat.py +++ b/backend/app/api/chat.py @@ -4,6 +4,7 @@ from __future__ import annotations import json import logging +import time import uuid from collections.abc import AsyncIterator @@ -79,8 +80,14 @@ async def _stream_response(request: ChatRequest) -> AsyncIterator[bytes]: full_content: list[str] = [] logger = logging.getLogger(__name__) + tool_start_ts: dict[str, int] = {} # track per-tool start timestamps + has_tool_activity = False # whether any tool has been called + final_status_emitted = False # whether "整理答案" status was emitted try: + # Emit initial status before graph starts + yield _sse({"type": "status", "stage": "分析问题", "message": "正在理解您的问题..."}) + async for event in graph.astream_events( input_data, config=config, @@ -93,29 +100,56 @@ async def _stream_response(request: ChatRequest) -> AsyncIterator[bytes]: if chunk and hasattr(chunk, "content") and chunk.content: # Only stream text content, skip tool call chunks if isinstance(chunk.content, str): + # Before first token after tool activity, emit "整理答案" status + if has_tool_activity and not final_status_emitted: + yield _sse({ + "type": "status", + "stage": "整理答案", + "message": "正在结合检索结果生成回复...", + }) + final_status_emitted = True full_content.append(chunk.content) - sse_data = json.dumps( - {"type": "token", "content": chunk.content}, - ensure_ascii=False, - ) - yield f"data: {sse_data}\n\n".encode("utf-8") + yield _sse({"type": "token", "content": chunk.content}) elif kind == "on_tool_start": - # Notify frontend that a tool is being called tool_name = event.get("name", "unknown") - sse_data = json.dumps( - {"type": "tool_start", "tool": tool_name}, - ensure_ascii=False, - ) - yield f"data: {sse_data}\n\n".encode("utf-8") + tool_input = event.get("data", {}).get("input", {}) + ts = int(time.time() * 1000) + tool_start_ts[tool_name] = ts + has_tool_activity = True + yield _sse({ + "type": "tool_start", + "tool": tool_name, + "title": _TOOL_TITLES.get(tool_name, tool_name), + "input_summary": _summarize_input(tool_name, tool_input), + "ts": ts, + }) elif kind == "on_tool_end": tool_name = event.get("name", "unknown") - sse_data = json.dumps( - {"type": "tool_end", "tool": tool_name}, - ensure_ascii=False, - ) - yield f"data: {sse_data}\n\n".encode("utf-8") + output = event.get("data", {}).get("output", "") + output_str = output if isinstance(output, str) else str(output) + ts = int(time.time() * 1000) + duration_ms = ts - tool_start_ts.pop(tool_name, ts) + if _is_tool_error(output_str): + yield _sse({ + "type": "tool_error", + "tool": tool_name, + "title": _TOOL_TITLES.get(tool_name, tool_name), + "error_summary": _extract_error_summary(output_str), + "duration_ms": duration_ms, + "ts": ts, + }) + else: + yield _sse({ + "type": "tool_end", + "tool": tool_name, + "title": _TOOL_TITLES.get(tool_name, tool_name), + "output_summary": _summarize_output(tool_name, output_str), + "status": "success", + "duration_ms": duration_ms, + "ts": ts, + }) except Exception as exc: logger.error("SSE stream error for conversation %s: %s", request.conversation_id, exc, exc_info=True) @@ -156,8 +190,92 @@ async def _stream_response(request: ChatRequest) -> AsyncIterator[bytes]: await _persist_ai_message(request.conversation_id, ai_content) # Always send done so the frontend closes the stream cleanly - done_data = json.dumps({"type": "done"}) - yield f"data: {done_data}\n\n".encode("utf-8") + yield _sse({"type": "done"}) + + +# ── Trace helpers ───────────────────────────────────────────────────────────── + +def _sse(data: dict) -> bytes: + return f"data: {json.dumps(data, ensure_ascii=False)}\n\n".encode("utf-8") + + +_TOOL_TITLES: dict[str, str] = { + "kb_search": "检索知识库", + "ticket_list": "查询工单列表", + "ticket_detail": "查询工单详情", + "web_search": "外部搜索", + "generate_document": "生成文档", + "sandbox_run": "执行沙盒代码", +} + +_ERROR_KEYWORDS = ( + "出错", "失败", "超时", "error", "failed", "timeout", + "not available", "no download link", "execution failed", +) + + +def _is_tool_error(output: str) -> bool: + lo = output.lower() + return any(kw in lo for kw in _ERROR_KEYWORDS) + + +def _extract_error_summary(output: str) -> str: + first_line = output.split("\n")[0].strip() + return first_line[:80] if first_line else "工具调用失败" + + +def _summarize_input(tool_name: str, inp: dict | str) -> str: + if isinstance(inp, str): + return inp[:60] + match tool_name: + case "kb_search": + return f"查询：{str(inp.get('query', ''))[:50]}" + case "ticket_list": + return f"第 {inp.get('page', 1)} 页，每页 {inp.get('page_size', 20)} 条" + case "ticket_detail": + return f"工单 ID：{inp.get('ticket_id', '')}" + case "web_search": + return f"搜索：{str(inp.get('query', ''))[:50]}" + case "generate_document": + return str(inp.get('prompt', ''))[:60] + case "sandbox_run": + lang = inp.get("language", "python") + lines = len(str(inp.get("code", "")).splitlines()) + return f"{lang} 代码（{lines} 行）" + case _: + return str(inp)[:60] + + +def _summarize_output(tool_name: str, output: str) -> str: + if not output or output.strip() in ("", "(no output)"): + return "无结果" + match tool_name: + case "kb_search": + count = output.count("---") + 1 if "---" in output else 1 + return f"命中 {count} 条知识库记录" + case "ticket_list": + import re as _re + m = _re.search(r"Found (\d+) tickets", output) + return f"返回 {m.group(1)} 条工单" if m else "工单列表已获取" + case "ticket_detail": + return "工单详情已获取" + case "web_search": + count = output.count("##") + return f"找到 {max(count, 1)} 条搜索结果" + case "generate_document": + if "Download:" in output: + if "[PPT]" in output: + return "PPT 已生成，可下载" + if "[Excel]" in output or "[Table]" in output: + return "表格已生成，可下载" + return "Word 文档已生成，可下载" + return "文档生成完成" + case "sandbox_run": + lines = len(output.splitlines()) + has_error = output.startswith("[Exit code:") + return f"执行完成，输出 {lines} 行{'（含错误）' if has_error else ''}" + case _: + return output[:60] @post("/api/chat/stream")