# -*- coding: utf-8 -*- """ SSE 核心工具函数 —— 流式事件生成器、文本提取、响应清理等通用辅助 """ import json import re import time import traceback from typing import AsyncGenerator from langgraph.types import Command # 程序化 interrupt() 恢复时需启用 from tools.tool_names_cn import TOOL_NAME_CN as _TOOL_CN from db.chat_store import save_message, get_messages from tools.context_tracker import capture_context_usage # ── 工具名 → 中文描述映射 ── def _cn_tool_desc(tool_name: str) -> str: """将工具函数名映射为中文描述短语（用于 SSE message 字段）。 映射表统一维护在 tools/tool_names_cn.py 中，新增工具只需改那一处。 """ return _TOOL_CN.get(tool_name, tool_name) # ── 通用 SSE 流式事件生成器 ── async def sse_event_generator( agent, user_message: str, thread_id: str, session_id: str, config: dict, agent_cn_name: str = "智能助手", save_to_db: bool = True, original_user_message: str | None = None, interrupt_before: list | None = None, interrupt_after: list | None = None, ) -> AsyncGenerator[str, None]: """ 通用 SSE 流式事件生成器。 使用 stream_mode=["updates", "messages"] 获取完整执行图景， 前端可根据 event type 区分：thinking / executing / tool_call / token / done / error。 Args: save_to_db: 是否将消息保存到数据库。点选解读等非对话场景应设为 False。 original_user_message: 用户的原始消息（不含系统注入的前缀）。 若提供，则 DB 保存此原始消息；user_message 仍作为 Agent 的输入。 """ # 保存用户消息（仅对话场景写库），优先使用原始消息 if save_to_db: save_message(session_id, "user", original_user_message or user_message) # 用于收集完整回复 full_response = "" start_time = time.time() _token_usage = {"prompt": 0, "completion": 0} # 从流中捕获的实际 token 用量 try: # ── 发送 agent_start 事件，前端据此创建 Agent 卡片 ── yield f"data: {json.dumps({'type': 'agent_start', 'agent': agent_cn_name, 'cn_agent': agent_cn_name, 'message': f'「{agent_cn_name}」开始处理...'}, ensure_ascii=False)}\n\n" # 构建消息列表（包含历史消息 + 当前消息） if save_to_db: messages = [{"role": "user", "content": msg["content"]} for msg in get_messages(session_id, limit=50)] else: messages = [] messages.append({"role": "user", "content": user_message}) # 使用 agent.astream() 异步流式调用，多模式获取完整图景 async for chunk in agent.astream( {"messages": messages}, stream_mode=["updates", "messages"], config=config, interrupt_before=interrupt_before, interrupt_after=interrupt_after, version="v2", ): # 判断事件来源：主代理 vs 子代理 is_subagent = any(s.startswith("tools:") for s in chunk.get("ns", [])) source = "subagent" if is_subagent else "main" # ── updates 模式：步骤级事件（thinking / executing）── if chunk["type"] == "updates": for node_name in chunk["data"]: if node_name == "model": # 代理正在思考/推理 yield f"data: {json.dumps({'type': 'thinking', 'source': source, 'node': node_name, 'cn_agent': agent_cn_name, 'message': '正在分析您的问题...'}, ensure_ascii=False)}\n\n" elif node_name == "tools": # 代理正在执行工具调用，提取工具名称 tools_data = chunk["data"].get(node_name, {}) tool_names = [] for msg in tools_data.get("messages", []): if hasattr(msg, "name"): tool_names.append(msg.name) elif isinstance(msg, dict) and msg.get("name"): tool_names.append(msg["name"]) # 生成中文描述 cn_names = [_cn_tool_desc(t) for t in tool_names] if tool_names else ["工具调用"] msg_text = f"正在{'、'.join(cn_names)}..." yield f"data: {json.dumps({'type': 'executing', 'source': source, 'cn_agent': agent_cn_name, 'tools': tool_names, 'cn_tools': cn_names, 'message': msg_text}, ensure_ascii=False)}\n\n" # ── messages 模式：token 级事件（token / tool_call / tool_result / updated_todo_list）── elif chunk["type"] == "messages": token_data = chunk["data"] if isinstance(token_data, (list, tuple)) and len(token_data) >= 1: msg_obj, _metadata = token_data[0], token_data[1] if len(token_data) > 1 else {} else: msg_obj = token_data # 检测工具调用（tool_call_chunks 在流式传输中逐步到达） if hasattr(msg_obj, "tool_call_chunks") and msg_obj.tool_call_chunks: for tc in msg_obj.tool_call_chunks: if tc.get("name"): cn = _cn_tool_desc(tc["name"]) yield f"data: {json.dumps({'type': 'tool_call', 'source': source, 'cn_agent': agent_cn_name, 'tool': tc['name'], 'cn_tool': cn, 'message': f'调用工具：{cn}'}, ensure_ascii=False)}\n\n" # 检测工具结果 is_tool_msg = hasattr(msg_obj, "type") and msg_obj.type == "tool" if is_tool_msg: tool_name = getattr(msg_obj, "name", "unknown") if tool_name == "write_todos": # write_todos 特殊处理：提取 JSON 并发送 updated_todo_list 事件 tool_content = _extract_text_content(msg_obj) or "" todo_list = _parse_todo_list(tool_content) if todo_list: yield f"data: {json.dumps({'type': 'updated_todo_list', 'cn_agent': agent_cn_name, 'source': source, 'todos': todo_list, 'message': '任务进度已更新'}, ensure_ascii=False)}\n\n" else: # 解析失败时也至少发一个 tool_result 事件，并打印诊断日志 print(f"[write_todos] 解析失败，原始内容: {tool_content[:300]}") yield f"data: {json.dumps({'type': 'tool_result', 'source': source, 'cn_agent': agent_cn_name, 'tool': tool_name, 'cn_tool': _cn_tool_desc(tool_name), 'message': '任务进度已更新'}, ensure_ascii=False)}\n\n" else: cn = _cn_tool_desc(tool_name) yield f"data: {json.dumps({'type': 'tool_result', 'source': source, 'cn_agent': agent_cn_name, 'tool': tool_name, 'cn_tool': cn, 'message': f'{cn} 完成'}, ensure_ascii=False)}\n\n" # 提取并流式输出文本内容（工具结果不当作 token 输出，仅输出 AI 生成的文本） if not is_tool_msg: # 先检查推理/思考内容（DeepSeek 等模型的 reasoning_content） reasoning = _extract_reasoning_content(msg_obj) if reasoning: if _DEBUG_REASONING: print(f"[reasoning] ✅ YIELD reasoning event ({len(reasoning)} chars)") yield f"data: {json.dumps({'type': 'reasoning', 'source': source, 'content': reasoning}, ensure_ascii=False)}\n\n" # 再提取常规文本内容 content = _extract_text_content(msg_obj) if content and isinstance(content, str): full_response += content yield f"data: {json.dumps({'type': 'token', 'source': source, 'content': content}, ensure_ascii=False)}\n\n" # ── 提取 token 用量元数据（最后一个消息块通常携带 usage）── _capture_token_meta(msg_obj, _token_usage) # 保存助手回复（仅对话场景写库） if save_to_db and full_response.strip(): cleaned = _clean_response(full_response) save_message(session_id, "assistant", cleaned) print(full_response) # ── 更新上下文用量 ── if save_to_db and _token_usage["prompt"] > 0: await capture_context_usage( session_id=session_id, prompt_tokens=_token_usage["prompt"], completion_tokens=_token_usage["completion"], current_message=user_message, ) # 发送 agent_done + done 事件 duration_ms = int((time.time() - start_time) * 1000) yield f"data: {json.dumps({'type': 'agent_done', 'agent': agent_cn_name, 'cn_agent': agent_cn_name, 'duration_ms': duration_ms, 'message': f'「{agent_cn_name}」完成（{duration_ms}ms）'}, ensure_ascii=False)}\n\n" # 检查 LangGraph 中断状态（Human-in-the-Loop） interrupt_info = _check_interrupt(agent, config) if interrupt_info: yield f"data: {json.dumps({'type': 'interrupt', **interrupt_info}, ensure_ascii=False)}\n\n" # 中断时不发送 done 事件，等待用户审批后恢复 return yield f"data: {json.dumps({'type': 'done', 'thread_id': thread_id, 'session_id': session_id, 'duration_ms': duration_ms, 'message': f'回答完成（{duration_ms}ms）'}, ensure_ascii=False)}\n\n" except Exception as e: traceback.print_exc() yield f"data: {json.dumps({'type': 'error', 'message': str(e)}, ensure_ascii=False)}\n\n" # ── 文本提取 / 解析 / 清理 ── def _extract_text_content(msg_obj) -> str | None: """从 LangChain 消息对象中提取文本内容（兼容 dict 和对象两种形式）。""" if isinstance(msg_obj, dict): return msg_obj.get("content") elif hasattr(msg_obj, "content"): raw = getattr(msg_obj, "content", None) if isinstance(raw, str): return raw elif isinstance(raw, list): # 多模态内容块：合并所有 text 类型的块 parts = [] for block in raw: if isinstance(block, dict) and block.get("type") == "text": parts.append(block.get("text", "")) elif hasattr(block, "type") and getattr(block, "type", "") == "text": parts.append(getattr(block, "text", "")) return "".join(parts) if parts else None return None # ── 调试开关：设为 True 时打印推理内容摘要（排查完毕后关闭）── _DEBUG_REASONING = True _debug_dump_done = False # 只 dump 第一个非空 AI chunk 的完整属性 def _extract_reasoning_content(msg_obj) -> str | None: """从 LangChain 消息对象中提取推理/思考内容（DeepSeek 等模型的 reasoning_content）。 支持三种来源（按优先级）： 1. msg_obj.additional_kwargs["reasoning_content"]（OpenAI 兼容流式 delta） 2. msg_obj.reasoning_content（LangChain 直接属性） 3. content 中的 thinking 类型块 Returns: 推理文本字符串，无推理内容时返回 None """ global _debug_dump_done msg_type = getattr(msg_obj, "type", "?") is_tool = (msg_type == "tool") content_raw = getattr(msg_obj, "content", "") # ── 调试：dump 第一个非空非工具 chunk 的所有属性 ── if _DEBUG_REASONING and not _debug_dump_done and not is_tool and content_raw: _debug_dump_done = True print(f"\n[reasoning-dump] === 第一个非空 AI chunk 完整属性 ===") print(f" type: {type(msg_obj).__name__}") # 打印所有属性（包括私有） for attr in sorted(dir(msg_obj)): if attr.startswith('_') and not attr.startswith('__'): continue try: val = getattr(msg_obj, attr) if callable(val): continue s = repr(val) if len(s) > 400: s = s[:400] + f"... (total {len(s)} chars)" print(f" {attr}: {s}") except Exception as e: print(f" {attr}: ") # 特别检查 additional_kwargs ak = getattr(msg_obj, "additional_kwargs", {}) if isinstance(ak, dict) and ak: print(f" >>> additional_kwargs has keys: {list(ak.keys())}") for k, v in ak.items(): sv = repr(v) if len(sv) > 500: sv = sv[:500] + f"... ({len(sv)} total)" print(f" >>> [{k}]: {sv}") else: print(f" >>> additional_kwargs: EMPTY or not dict (type={type(ak).__name__})") # 检查 response_metadata rm = getattr(msg_obj, "response_metadata", {}) if isinstance(rm, dict) and rm: print(f" >>> response_metadata keys: {list(rm.keys())}") for k, v in rm.items(): sv = repr(v) if len(sv) > 300: sv = sv[:300] + f"..." print(f" >>> [{k}]: {sv}") print(f"[reasoning-dump] === dump 完毕 ===\n") # 方式 1：additional_kwargs 中的 reasoning_content（最常见） if hasattr(msg_obj, "additional_kwargs") and isinstance(msg_obj.additional_kwargs, dict): reasoning = msg_obj.additional_kwargs.get("reasoning_content", "") if reasoning: if _DEBUG_REASONING: print(f"[reasoning] ✅ additional_kwargs ({len(reasoning)} chars): {reasoning[:120]}...") return reasoning # 方式 2：直接属性 reasoning_content reasoning = getattr(msg_obj, "reasoning_content", None) if reasoning: if _DEBUG_REASONING: print(f"[reasoning] ✅ direct attr ({len(reasoning)} chars): {reasoning[:120]}...") return reasoning # 方式 3：content 为 list 时，提取 thinking 类型的块 if hasattr(msg_obj, "content"): raw = getattr(msg_obj, "content", None) if isinstance(raw, list): parts = [] for block in raw: if isinstance(block, dict) and block.get("type") == "thinking": parts.append(block.get("thinking", "")) elif hasattr(block, "type") and getattr(block, "type", "") == "thinking": parts.append(getattr(block, "thinking", "")) if parts: result = "".join(parts) if _DEBUG_REASONING: print(f"[reasoning] ✅ content blocks ({len(result)} chars): {result[:120]}...") return result return None def _capture_token_meta(msg_obj, usage_ref: dict): """从 LangChain 消息对象中提取 LLM token 用量元数据。 优先读取 usage_metadata（langchain ≥0.3）， 其次读取 response_metadata.usage（OpenAI 兼容格式）。 结果写入 usage_ref dict（原地修改）。 """ # 方式 1：usage_metadata（langchain 标准字段） um = getattr(msg_obj, "usage_metadata", None) if um and isinstance(um, dict): inp = um.get("input_tokens", 0) out = um.get("output_tokens", 0) if inp or out: usage_ref["prompt"] = inp usage_ref["completion"] = out return # 方式 2：response_metadata.usage（OpenAI / DeepSeek 兼容） rm = getattr(msg_obj, "response_metadata", None) if rm and isinstance(rm, dict): usage = rm.get("usage", {}) or rm.get("token_usage", {}) if isinstance(usage, dict): inp = usage.get("prompt_tokens", 0) out = usage.get("completion_tokens", 0) if inp or out: usage_ref["prompt"] = inp usage_ref["completion"] = out return def _parse_todo_list(text: str) -> list | None: """从 write_todos 输出中提取 todo 列表。 write_todos 输出格式: "Updated todo list to [{'content': '...', 'status': '...'}, ...]" 返回 JSON-serializable list of dicts，失败返回 None。 """ import ast # 贪婪匹配最外层 [...]（内容中的 [ ] 在字符串字面量内，ast.literal_eval 可正确处理） match = re.search(r"\[.*\]", text) if not match: return None try: python_list = ast.literal_eval(match.group()) if isinstance(python_list, list): return python_list except (ValueError, SyntaxError, TypeError): pass return None def _clean_response(text: str) -> str: """清理 Agent 回复中的格式噪音""" # 移除 ANSI 转义序列 text = re.sub(r'\x1b\[[0-9;]*m', '', text) text = re.sub(r'\x1b\[[0-9;]*[a-zA-Z]', '', text) text = re.sub(r'\x1b\][^\x07]*\x07', '', text) # 移除工具调用残留（大括号 JSON 块如果独立成行则移除） text = re.sub(r'^\s*\{[^}]*\}\s*$', '', text, flags=re.MULTILINE) return text.strip() # ── LangGraph Human-in-the-Loop 中断处理 ── def _check_interrupt(agent, config: dict) -> dict | None: """ 检查 LangGraph 状态是否被中断。 支持两种中断检测： 1. 程序化中断（节点内调用 interrupt()）→ state.interrupts 非空 2. interrupt_before / interrupt_after 暂停 → state.next 非空 当中断发生时返回中断信息 dict，否则返回 None。 返回格式: {"node": "tools", "message": "智能体准备执行工具，等待审批...", "interrupts": [...], "plan": "计划文本（程序化中断时）"} """ try: state = agent.get_state(config) except Exception: return None if state is None: return None # 1. 检测程序化中断（节点内调用 interrupt()） interrupts = getattr(state, "interrupts", None) if not interrupts: values = getattr(state, "values", {}) or {} interrupts = values.get("__interrupt__", []) # 2. 检测 interrupt_before / interrupt_after 暂停 # 图暂停时 state.next 非空（有待执行节点）；完成时为空 next_nodes = getattr(state, "next", None) is_paused_by_interrupt_config = bool(next_nodes) if next_nodes is not None else False if not interrupts and not is_paused_by_interrupt_config: return None # 提取中断节点名 if interrupts: interrupt_data = interrupts[0] if interrupts else {} # 处理 interrupt() 返回值可能是 Interrupt 对象或 dict if hasattr(interrupt_data, "value"): interrupt_data = interrupt_data.value if isinstance(interrupt_data, dict): node = interrupt_data.get("type", "tools") plan = interrupt_data.get("plan", "") message = interrupt_data.get("message", "智能体暂停执行，等待您的审批...") else: node = str(next_nodes or "tools") plan = "" message = "智能体暂停执行，等待您的审批..." else: node = next_nodes plan = "" message = "智能体暂停执行，等待您的审批..." if isinstance(node, (list, tuple)): node = node[0] if node else "unknown" result = { "node": str(node), "message": message, "interrupts": [str(i) for i in interrupts] if interrupts else [], } if plan: result["plan"] = plan return result async def resume_stream( agent, session_id: str, thread_id: str, config: dict, agent_cn_name: str = "智能助手", action: str = "approve", interrupt_before: list | None = None, interrupt_after: list | None = None, ): """ 恢复被中断的 LangGraph 流式执行。 参数: action: "approve" 或 "reject" Yields: SSE 事件字符串 """ if action == "reject": yield f"data: {json.dumps({'type': 'error', 'message': '用户拒绝了工具执行'}, ensure_ascii=False)}\n\n" yield f"data: {json.dumps({'type': 'done', 'thread_id': thread_id, 'session_id': session_id, 'message': '已取消执行'}, ensure_ascii=False)}\n\n" return full_response = "" start_time = time.time() try: # 检测中断类型：程序化中断需用 Command(resume=...)，配置式中断传 None state = agent.get_state(config) has_programmatic = bool(getattr(state, "interrupts", None)) if state else False stream_input = Command(resume={"action": action}) if has_programmatic else None # 恢复执行 async for chunk in agent.astream( stream_input, stream_mode=["updates", "messages"], config=config, interrupt_before=interrupt_before, interrupt_after=interrupt_after, version="v2", ): is_subagent = any(s.startswith("tools:") for s in chunk.get("ns", [])) if chunk["type"] == "updates": for node_name in chunk["data"]: if node_name == "model": yield f"data: {json.dumps({'type': 'thinking', 'source': 'main', 'cn_agent': agent_cn_name, 'message': '继续执行...'}, ensure_ascii=False)}\n\n" elif node_name == "tools": tools_data = chunk["data"].get(node_name, {}) tool_names = [] for msg in tools_data.get("messages", []): if hasattr(msg, "name"): tool_names.append(msg.name) elif isinstance(msg, dict) and msg.get("name"): tool_names.append(msg["name"]) cn_names = [_cn_tool_desc(t) for t in tool_names] if tool_names else ["工具调用"] yield f"data: {json.dumps({'type': 'executing', 'source': 'main', 'cn_agent': agent_cn_name, 'tools': tool_names, 'cn_tools': cn_names, 'message': f"正在{'、'.join(cn_names)}..."}, ensure_ascii=False)}\n\n" elif chunk["type"] == "messages": token_data = chunk["data"] if isinstance(token_data, (list, tuple)) and len(token_data) >= 1: msg_obj = token_data[0] else: msg_obj = token_data # 工具结果 is_tool_msg = hasattr(msg_obj, "type") and msg_obj.type == "tool" if is_tool_msg: tool_name = getattr(msg_obj, "name", "unknown") if tool_name != "write_todos": yield f"data: {json.dumps({'type': 'tool_result', 'source': 'main', 'cn_agent': agent_cn_name, 'tool': tool_name, 'cn_tool': _cn_tool_desc(tool_name), 'message': f'{_cn_tool_desc(tool_name)} 完成'}, ensure_ascii=False)}\n\n" # AI 文本 if not is_tool_msg: # 先检查推理/思考内容（DeepSeek 等模型的 reasoning_content） reasoning = _extract_reasoning_content(msg_obj) if reasoning: if _DEBUG_REASONING: print(f"[reasoning] ✅ YIELD reasoning event resume ({len(reasoning)} chars)") yield f"data: {json.dumps({'type': 'reasoning', 'source': 'main', 'content': reasoning}, ensure_ascii=False)}\n\n" # 再提取常规文本内容 content = _extract_text_content(msg_obj) if content and isinstance(content, str): full_response += content yield f"data: {json.dumps({'type': 'token', 'source': 'main', 'content': content}, ensure_ascii=False)}\n\n" # 保存助手回复 if full_response.strip(): cleaned = _clean_response(full_response) save_message(session_id, "assistant", cleaned) # 再次检查中断（可能有多次中断） interrupt_info = _check_interrupt(agent, config) if interrupt_info: yield f"data: {json.dumps({'type': 'interrupt', **interrupt_info}, ensure_ascii=False)}\n\n" return duration_ms = int((time.time() - start_time) * 1000) yield f"data: {json.dumps({'type': 'done', 'thread_id': thread_id, 'session_id': session_id, 'duration_ms': duration_ms, 'message': f'回答完成（{duration_ms}ms）'}, ensure_ascii=False)}\n\n" except Exception as e: traceback.print_exc() yield f"data: {json.dumps({'type': 'error', 'message': str(e)}, ensure_ascii=False)}\n\n"