"""Agent 执行内核：产出结构化事件，统一处理落库、计费、取消与收尾。 updates 通道处理工具、待办和子 Agent；messages 通道处理正文和用量。 本模块不依赖 HTTP、SSE 或 API 请求模型。 """ from __future__ import annotations import asyncio import json from contextlib import aclosing import time from dataclasses import dataclass, field from datetime import datetime import anyio from typing import Any, AsyncIterator import config from runtime import concurrency from runtime.interrupts import pack, interrupt_event from runtime.context_meter import ContextMeter from runtime.events import EventBuilder, ExecutionEvent from runtime.source_tracker import SourceTracker from runtime.subagent_tracker import SubagentTracker from billing.meter import UsageMeter from db import chat_store, db_conn, run_store from obs import get_logger, get_trace_id, normalize_trace_id, bind_log_context, unbind_log_context logger = get_logger(__name__) MAIN_AGENT_NAME = "deep-agent" MAIN_AGENT_CN = "深度智能体" # 摘要中间件的模型调用也在 messages 通道里推 token，靠这个标记过滤掉， # 否则摘要正文会混进给用户看的回答里 _SUMMARIZATION_SOURCE = "summarization" class ExecutionOutputError(RuntimeError): """执行流已结束，但没有可交付的回答或业务提交。""" def __init__(self, code: str, message: str) -> None: super().__init__(message) self.code = code @dataclass class ExecutionRequest: """一次流式请求的全部入参。""" agent: Any session_id: str thread_id: str config: dict[str, Any] message: str user_name: str = "default" kind: str = "chat" model: str = "" original_user_message: str = "" trace_id: str = "" slot: concurrency.TaskSlot | None = None started_at: float = field(default_factory=time.time) agent_name: str = MAIN_AGENT_NAME agent_cn_name: str = MAIN_AGENT_CN # 非 None 表示这是一次中断恢复：不组装消息，直接把 Command(resume=...) 喂回图 resume_payload: Any = None session_lease: Any = None business_lease: Any = None result_validator: Any = None run_id: str = field(default_factory=db_conn.new_id) @dataclass(frozen=True) class ExecutionResult: run_id: str session_id: str status: str output: str error: str usage: dict[str, Any] citations: list[dict[str, Any]] interrupts: list[Any] structured_output: dict[str, Any] | None = None diagnostics: dict[str, Any] = field(default_factory=dict) def normalize_chunk(chunk: Any) -> tuple[str | None, tuple[Any, ...], Any]: """把不同版本的 astream chunk 归一成 ``(mode, ns, data)``。 兼容 v2 的 ``{"type","ns","data"}`` 字典，以及 v1 的 ``(mode, data)`` 和启用 subgraphs 时的 ``(ns, mode, data)`` 元组。 """ if isinstance(chunk, dict) and "data" in chunk and "type" in chunk: return str(chunk["type"]), tuple(chunk.get("ns") or ()), chunk["data"] if isinstance(chunk, tuple) and len(chunk) == 2 and isinstance(chunk[0], str): return chunk[0], (), chunk[1] if (isinstance(chunk, tuple) and len(chunk) == 3 and isinstance(chunk[0], (tuple, list)) and isinstance(chunk[1], str)): return chunk[1], tuple(chunk[0]), chunk[2] return None, (), chunk def _message_text(message: Any) -> str: """把消息 content 拍平成纯文本（content 可能是字符串或内容块列表）。""" content = getattr(message, "content", None) if isinstance(content, str): return content if isinstance(content, list): parts: list[str] = [] for block in content: if isinstance(block, str): parts.append(block) elif isinstance(block, dict): text = block.get("text") if isinstance(text, str): parts.append(text) return "".join(parts) return "" def _reasoning_text(message: Any) -> str: """抽取思考链文本（deepseek 的 reasoning_content）。""" extra = getattr(message, "additional_kwargs", None) if isinstance(extra, dict): value = extra.get("reasoning_content") if isinstance(value, str): return value return "" class Execution: """一次任务执行，可逐事件消费，也可等待最终结果。""" def __init__(self, request: ExecutionRequest) -> None: self.req = request request.trace_id = normalize_trace_id(request.trace_id or get_trace_id()) self.writer = EventBuilder(request.session_id, request.trace_id or get_trace_id(), encode=lambda event: event) self.meter = _meter_for_request() self.tracker = SubagentTracker() self.sources = SourceTracker() self.billing = UsageMeter( user_name=request.user_name, session_id=request.session_id, kind=request.kind, model=request.model or config.MODEL, trace_id=request.trace_id, usage_id=request.run_id, usage_day=db_conn.day_key(datetime.fromtimestamp(request.started_at)), ) # 累计状态 self.answer_parts: list[str] = [] self.tool_calls: list[dict[str, Any]] = [] self.todos: list[dict[str, Any]] = [] self.interrupt_payloads: list[Any] = [] self.finished = False self.status = "running" self.error = "" self._billing_added = False self._started = False self._snapshot = None self._answer_saved = False self._settlement_task: asyncio.Task | None = None self._consumed = False self.structured_output: dict[str, Any] | None = None self._model_text_seen = False self._last_model_text = "" self.diagnostics: dict[str, Any] = { 'stream_frames': 0, 'message_frames': 0, 'update_frames': 0, 'unrecognized_frames': 0, 'model_responses': 0, 'last_response': {}, } async def run(self) -> ExecutionResult: """消费事件并完成资源清理后返回结果，供定时任务、批处理等入口复用。""" async with aclosing(self.events()) as events: async for _ in events: pass return self.result() def result(self) -> ExecutionResult: if not self.finished: raise RuntimeError("任务尚未完成，不能读取最终结果") return ExecutionResult(self.req.run_id, self.req.session_id, self.status, "".join(self.answer_parts), self.error, self._usage_payload(), self.sources.build_citations(), list(self.interrupt_payloads), self.structured_output, dict(self.diagnostics)) # ======================================================== # 主入口 # ======================================================== async def events(self) -> AsyncIterator[ExecutionEvent]: """产出结构化执行事件。保证：计费幂等入账、并发位释放、图游标关闭。 这里显式绑定 trace_id 而不是依赖 HTTP 中间件设的 ContextVar： 响应体的迭代在 Starlette 的中间件调用栈之外执行，中间件里绑的值 在生成器里不保证可见。绑定发生在生成器内部，生命周期与本次流一致。 """ from obs import bind_trace_id, unbind_trace_id if self._consumed: raise RuntimeError("同一任务只能执行一次，请创建新的 Execution") self._consumed = True token = bind_trace_id(self.req.trace_id) context_token = bind_log_context(run_id=self.req.run_id, session_id=self.req.session_id, user_name=self.req.user_name, kind=self.req.kind) started = time.perf_counter() logger.info("任务开始", extra={"kv": {"event": "execution.started", "model": self.req.model or config.MODEL}}) stream = None try: if self.req.business_lease is not None: self.req.business_lease.bind() await asyncio.to_thread(run_store.start, self.req.run_id, self.req.session_id, self.req.user_name, self.req.kind) self._started = True yield self.writer.agent_start(self.req.agent_name, self.req.agent_cn_name) yield self.writer.thinking("正在理解您的问题...", self.req.agent_cn_name) graph_input = await self._prepare_input() stream = self.req.agent.astream( graph_input, stream_mode=["updates", "messages"], config=self.req.config, subgraphs=True, version="v2", ) async for frame in self._drive(stream): yield frame # 中断检测必须在流结束后做：interrupt 会让流自然结束 async for frame in self._emit_interrupts(): yield frame if self.structured_output is not None and not self.interrupt_payloads and self.req.result_validator: self.structured_output = await asyncio.to_thread(self.req.result_validator, self.structured_output, self.req.user_name, self.req.session_id) self._validate_completion() async for frame in self._completion(): yield frame except (asyncio.CancelledError, GeneratorExit): self.status = "cancelled" raise except Exception as exc: self.status = "failed" self.error = exc.code if isinstance(exc, ExecutionOutputError) else type(exc).__name__ async for frame in self._error_events(exc): yield frame finally: cleanup_failed = False try: # 清理与发帧独立。ASGI 断连的取消作用域不能打断保存和释放。 with anyio.CancelScope(shield=True): cleanup = asyncio.create_task(self._cleanup(stream)) try: await asyncio.shield(cleanup) except asyncio.CancelledError: await cleanup except Exception: cleanup_failed = True logger.exception("任务收尾失败", extra={"kv": {"event": "execution.cleanup_failed"}}) raise finally: try: status = "cleanup_failed" if cleanup_failed else self.status level = "error" if status in {"failed", "cleanup_failed"} else "warning" if status == "cancelled" else "info" getattr(logger, level)("任务结束", extra={"kv": { "event": "execution.completed", "status": status, "duration_ms": round((time.perf_counter() - started) * 1000, 1), "input_tokens": self.billing.input_tokens, "output_tokens": self.billing.output_tokens, "usage_source": self.billing.usage_source, "tool_calls": len(self.tool_calls), "error_code": self.error or None, "diagnostics": self.diagnostics, }}) finally: unbind_log_context(context_token) unbind_trace_id(token) async def _cleanup(self, stream: Any) -> None: try: if stream is not None: close = getattr(stream, "aclose", None) if close is not None: try: await close() except Exception: logger.exception("关闭图事件流失败 session=%s", self.req.session_id) await self._settle(aborted=self.status in {"cancelled", "failed"}) finally: try: await asyncio.to_thread(self._finalize) finally: if self.req.session_lease is not None: self.req.session_lease.release() if self.req.business_lease is not None: self.req.business_lease.release() # ======================================================== # 输入准备 # ======================================================== async def _prepare_input(self) -> Any: """组装喂给图的输入。 两种情形： - **恢复**（``resume_payload`` 非 None）：喂 ``Command(resume=...)``， 不组装消息——图会从挂起的那个节点继续。 - **新请求**：热线程（checkpoint 里已有状态）只喂当轮新消息；冷线程 需要从会话历史补种。 补种这一步不能省，也不能对所有线程都做：热线程若把历史重复追加， 上下文会随轮次按 O(N²) 膨胀——这是长会话最容易被忽略的膨胀来源。 """ if self.req.resume_payload is not None: from langgraph.types import Command logger.info("恢复中断 session=%s", self.req.session_id) return Command(resume=self.req.resume_payload) from langchain_core.messages import AIMessage, HumanMessage from agents import agent_memory messages: list[Any] = [] warm = await agent_memory.has_checkpoint(self.req.thread_id) if not warm: history = await asyncio.to_thread(chat_store.get_messages, self.req.session_id, limit=config.CTX_RECENT_LIMIT ) for row in history: content = row.get("content") or "" if not content: continue if row.get("msg_role") == "user": messages.append(HumanMessage(content=content)) else: messages.append(AIMessage(content=content)) if messages: logger.info( "冷线程补种历史 session=%s 条数=%d", self.req.session_id, len(messages) ) else: logger.debug("热线程，仅喂当轮消息 session=%s", self.req.session_id) # 用户消息先落库：即使这一轮流到一半挂了，历史里也得有用户问了什么 await asyncio.to_thread(chat_store.save_message, self.req.session_id, "user", self.req.original_user_message or self.req.message) self.meter.note_text(self.req.message) messages.append(HumanMessage(content=self.req.message)) # 必须是 {"messages": [...]} 字典：直接传 list 会被当成 __start__ 节点的 # 返回值，图会报 InvalidUpdateError: Expected dict return {"messages": messages} # ======================================================== # 流驱动 # ======================================================== async def _drive(self, stream: Any) -> AsyncIterator[ExecutionEvent]: """消费 astream，派发事件帧。带墙钟上限。""" remaining = config.TASK_WALL_TIMEOUT - (time.time() - self.req.started_at) deadline = time.monotonic() + max(remaining, 0) timed_out = False iterator = stream.__aiter__() while True: if time.monotonic() >= deadline: timed_out = True break try: chunk = await asyncio.wait_for( iterator.__anext__(), timeout=max(deadline - time.monotonic(), 0.001) ) except StopAsyncIteration: break except asyncio.TimeoutError: timed_out = True break mode, ns, data = normalize_chunk(chunk) self.diagnostics['stream_frames'] += 1 if mode == "updates": self.diagnostics['update_frames'] += 1 for frame in self._handle_updates(data, ns): yield frame elif mode == "messages": self.diagnostics['message_frames'] += 1 for frame in self._handle_messages(data, ns): yield frame elif mode is None: self.diagnostics['unrecognized_frames'] += 1 if timed_out: raise concurrency.TaskWallTimeoutError( f"任务超过墙钟上限 {config.TASK_WALL_TIMEOUT}s 已被中止" ) # 子智能体可能因为异常/中断没收到结果，收尾时统一关掉，避免前端卡片长转 for run in self.tracker.finish_all(): yield self.writer.progress( f"子智能体 {run.subagent_type} 未正常返回，已中止", subagent=run.subagent_type, subagent_id=run.call_id, ) # -------- updates 通道 -------- def _handle_updates(self, data: Any, ns: tuple[Any, ...]) -> list[ExecutionEvent]: """处理 updates 通道：工具事件、待办、子智能体台账。""" from langchain_core.messages import AIMessage, ToolMessage frames: list[ExecutionEvent] = [] if not isinstance(data, dict): return frames # 中断会以 __interrupt__ 键出现在 updates 里，比 aget_state 更及时 if "__interrupt__" in data: for item in data.get("__interrupt__") or (): self.interrupt_payloads.append(pack(item)) return frames # 子智能体身份：updates 的 ns 指向它所属的 task 调用 run = self.tracker.note_namespace(ns) if ns else None subagent_type = run.subagent_type if run else "" subagent_id = run.call_id if run else "" for node, update in data.items(): if not isinstance(update, dict): continue if not ns and "structured_response" in update: value = update["structured_response"] if hasattr(value, "model_dump"): value = value.model_dump(mode="json") if isinstance(value, dict): self.structured_output = value todos = update.get("todos") if isinstance(todos, list): self.todos = [t for t in todos if isinstance(t, dict)] frames.append(self.writer.todo_list(self.todos)) for message in update.get("messages") or (): if isinstance(message, ToolMessage): frames.extend(self._on_tool_message(message, subagent_type, subagent_id)) elif isinstance(message, AIMessage): if not ns: self._note_model_response(message) frames.extend(self._on_ai_message(message, subagent_type, subagent_id)) return frames def _note_model_response(self, message: Any) -> None: """只记录响应形状与结束原因，不写入正文、工具参数或请求头。""" text = _message_text(message) self._last_model_text = text self._model_text_seen |= bool(text.strip()) metadata = getattr(message, 'response_metadata', None) or {} self.diagnostics['model_responses'] += 1 self.diagnostics['last_response'] = { 'content_chars': len(text), 'reasoning_chars': len(_reasoning_text(message)), 'tool_calls': len(getattr(message, 'tool_calls', None) or []), 'invalid_tool_calls': len(getattr(message, 'invalid_tool_calls', None) or []), **{key: str(metadata[key])[:256] for key in ('finish_reason', 'id', 'model_name') if metadata.get(key) is not None}, } def _validate_completion(self) -> None: if self.interrupt_payloads or self.structured_output is not None: return if self.diagnostics['last_response'].get('invalid_tool_calls'): raise ExecutionOutputError('invalid_model_tool_calls', '模型返回了无法解析的工具调用，未完成任务；请检查模型服务的工具调用格式。') if not self._model_text_seen and not self.answer_parts and not self.tool_calls: if self.diagnostics['unrecognized_frames'] and not self.diagnostics['model_responses']: raise ExecutionOutputError('unsupported_agent_stream', '执行流返回了无法识别的事件格式，未取得有效输出；请核对框架版本与流式事件格式。') raise ExecutionOutputError('empty_model_response', '执行流结束，但未收到有效的模型正文、工具调用或正式结果；请检查模型服务响应及流式兼容性。') if self.req.result_validator: raise ExecutionOutputError('missing_business_result', 'Agent 已结束，但未通过提交工具交付正式业务结果。') def _on_ai_message( self, message: Any, subagent_type: str, subagent_id: str ) -> list[ExecutionEvent]: """处理一条完整的 AIMessage：登记 task 调用、发工具调用卡。""" from tools.registry import get_label frames: list[ExecutionEvent] = [] tool_calls = list(getattr(message, "tool_calls", None) or []) # 先让跟踪器登记 task 调用，这样后续子智能体事件能立刻带上身份 for run in self.tracker.note_task_calls(tool_calls): frames.append( self.writer.progress( f"委派子智能体：{run.subagent_type}", subagent=run.subagent_type, subagent_id=run.call_id, description=run.description, ) ) names = [c.get("name", "") for c in tool_calls if isinstance(c, dict)] if names: frames.append( self.writer.executing( f"调用工具：{'、'.join(get_label(n) for n in names)}", cn_agent=self.req.agent_cn_name, tools=names, cn_tools=[get_label(n) for n in names], ) ) for call in tool_calls: if not isinstance(call, dict): continue name = call.get("name") or "" if not name: continue args = call.get("args") self.tool_calls.append({"name": name, "args": args}) frames.append( self.writer.tool_call( name, args=args, cn_tool=get_label(name), agent_id=subagent_id, tool_call_id=str(call.get("id") or ""), ) ) return frames def _on_tool_message( self, message: Any, subagent_type: str, subagent_id: str ) -> list[ExecutionEvent]: """处理一条 ToolMessage：发工具结果卡、抽取来源、结子智能体。""" from tools.registry import get_label frames: list[ExecutionEvent] = [] name = getattr(message, "name", None) or "tool" content = _message_text(message) # task 的返回即子智能体结束 call_id = getattr(message, "tool_call_id", None) or "" if name == "task" and call_id: run = self.tracker.note_result(call_id, ok=True) if run is not None: frames.append( self.writer.agent_done( run.subagent_type, cn_agent=run.subagent_type, duration_ms=run.duration_ms, progress="已完成", agent_id=run.call_id, ) ) self.meter.note_text(content) self.sources.note_tool_result(name, content) frames.append( self.writer.tool_result( name, content=content[:4000], cn_tool=get_label(name), agent_id=subagent_id, error=str(getattr(message, "status", "") or "") == "error", tool_call_id=call_id, ) ) return frames # -------- messages 通道 -------- def _handle_messages(self, data: Any, ns: tuple[Any, ...]) -> list[ExecutionEvent]: """处理 messages 通道：正文 token、思考 token、用量计量。""" from langchain_core.messages import AIMessage, AIMessageChunk frames: list[ExecutionEvent] = [] if not isinstance(data, tuple) or len(data) != 2: return frames chunk, metadata = data metadata = metadata if isinstance(metadata, dict) else {} # 摘要中间件的模型调用也在这条通道上：它的 token 不能给用户看 source = metadata.get("lc_source") # 汇总全部模型费用；只有主线程更新当前上下文窗口占用。 self.meter.note_message(chunk, is_main_context=not ns and source != _SUMMARIZATION_SOURCE) if source == _SUMMARIZATION_SOURCE or ns: return frames # 只有 AIMessageChunk 的内容才是「模型正在说的话」。ToolMessage 也走 # 这条通道，它的 content 是工具返回的原始数据——不加这个判断，工具 # 结果的 JSON 会混进给用户看的正文里。 if not isinstance(chunk, (AIMessage, AIMessageChunk)): return frames reasoning = _reasoning_text(chunk) if reasoning: frames.append(self.writer.thinking_token(reasoning)) text = _message_text(chunk) if text: self._model_text_seen |= bool(text.strip()) self.meter.note_text(text) # 有结果验证器的业务以恢复后的原件为最终正文；模型转述不能先行发布。 if not self.req.result_validator: self.answer_parts.append(text) frames.append(self.writer.token(text)) return frames # ======================================================== # 中断 # ======================================================== async def _emit_interrupts(self) -> AsyncIterator[ExecutionEvent]: """把中断转成 1.5 形状的 interrupt 事件。 两个来源：流里捕获到的 ``__interrupt__``，以及流结束后从图状态里 读出的 pending interrupt。优先用前者（更及时），两者都空则无中断。 """ payloads = list(self.interrupt_payloads) if not payloads: payloads = await self._interrupts_from_state() if not payloads: return self.interrupt_payloads = payloads for payload in payloads: frame = self._adapt_interrupt(payload) if frame: yield frame async def _interrupts_from_state(self) -> list[Any]: """从图状态里读出挂起的中断载荷。""" get_state = getattr(self.req.agent, "aget_state", None) if get_state is None: return [] try: state = await get_state(self.req.config) except Exception as exc: # pragma: no cover - 取决于底层库 logger.warning("读取图状态失败：%s: %s", type(exc).__name__, exc) return [] found: list[Any] = [] for task in getattr(state, "tasks", None) or (): for item in getattr(task, "interrupts", None) or (): found.append(pack(item)) return found def _adapt_interrupt(self, payload: Any) -> ExecutionEvent: return interrupt_event(self.writer, payload) # ======================================================== # 收尾 # ======================================================== async def _settle(self, aborted: bool = False) -> None: # 断连恰逢数据库写入时继续等待同一次收尾，避免在线程仍写入时重做。 if self._settlement_task is None: self._settlement_task = asyncio.create_task(self._settle_once(aborted)) await asyncio.shield(self._settlement_task) async def _settle_once(self, aborted: bool = False) -> None: """落库 → 计费 → 来源 → 上下文用量 → agent_done/done。 无论正常结束、异常还是断连都走这里（幂等，重复调用只生效一次）。 """ if self.finished: return duration_ms = int((time.time() - self.req.started_at) * 1000) answer = "".join(self.answer_parts) if not answer and self.structured_output is not None: value = self.structured_output if isinstance(value.get("result"), str): report = value.get("result_format") == "report" title = value.get("result_title", "") answer = f"{'# ' if report and title else ''}{title}\n\n{value['result']}".strip() if value.get("missing_materials") and not report: answer += "\n\n需要补充：" + "、".join(value["missing_materials"]) else: answer = json.dumps(value, ensure_ascii=False, allow_nan=False) self.answer_parts.append(answer) if aborted and answer: answer += "\n\n（本次回复被中断，内容可能不完整）" # 1) 上下文用量快照。**只取一次**：snapshot() 会消费掉压缩存档的待办标记， # 取两次会导致第二次拿不到存档内容。 self._apply_estimate_fallback() if self._snapshot is None: self._snapshot = self.meter.snapshot() snapshot = self._snapshot # 2) 计费：把上下文计量器的数字转给计费计量器。 # 这是两个计量器唯一的连接点——漏了这一步，交互式对话全程不会计费， # 而 /usage/today 会显示 0，表面上「一切正常」。 if not self._billing_added: self.billing.add(snapshot.input_tokens, snapshot.output_tokens, snapshot.usage_source) self._billing_added = True # 3) 落库：assistant 消息是会话历史的事实源 citations = self.sources.build_citations() if (answer or citations) and not self._answer_saved: try: await asyncio.to_thread( chat_store.save_message, self.req.session_id, "assistant", answer, tool_calls=self.tool_calls or None, citations=citations or None, ) self._answer_saved = True except Exception as exc: self.status = "failed" self.error = "message_persistence_failed" logger.exception("回复落库失败", extra={"kv": {"event": "execution.persistence_failed"}}) # 4) 上下文用量 + 压缩存档 await self._persist_context(snapshot) # 5) 计费入账（幂等） await asyncio.to_thread(self.billing.settle) if self.status == "running": self.status = "waiting_input" if self.interrupt_payloads else "success" if self._started: await asyncio.to_thread(run_store.finish, self.req.run_id, self.status, self.error) # 用户在普通恢复接口完成定时任务审批后，同步其执行记录。 if self.req.kind == "resume": await asyncio.to_thread(db_conn.execute, "UPDATE schedule_runs SET status = ?, finished_at = ? WHERE thread_id = ? AND status = 'waiting_input'", (self.status, None if self.status == "waiting_input" else db_conn.fmt_dt(), self.req.session_id)) self.finished = True async def _completion(self, aborted: bool = False) -> AsyncIterator[ExecutionEvent]: had_answer = bool(self.answer_parts) if not had_answer and not self.req.result_validator and self._last_model_text.strip(): # 某些模型只在 updates 发完整消息，仍须保存并展示最终正文。 self.answer_parts.append(self._last_model_text) await self._settle(aborted) duration_ms = int((time.time() - self.req.started_at) * 1000) citations = self.sources.build_citations() # 6) 来源标注 yield self.writer.citations(citations, model_only=not citations) if aborted or self.status in {"failed", "cancelled"}: if self.error == "message_persistence_failed": yield self.writer.error("回复未能保存，请稍后检查历史记录", code=self.error) yield self.writer.done(duration_ms=duration_ms, aborted=True) return if self.status == "waiting_input": yield self.writer.done(duration_ms=duration_ms, waiting_input=True, run_id=self.req.run_id, usage=self._usage_payload()) return if not had_answer and self.answer_parts: yield self.writer.token("".join(self.answer_parts)) yield self.writer.agent_done( self.req.agent_name, cn_agent=self.req.agent_cn_name, duration_ms=duration_ms, progress="已完成", **self.meter.summary_fields(), ) yield self.writer.done(duration_ms=duration_ms, run_id=self.req.run_id, usage=self._usage_payload()) def _usage_payload(self) -> dict[str, Any]: """给前端的用量摘要（便于测试页显示，1.5 前端会忽略未知字段）。""" return { "input_tokens": self.meter.input_tokens, "output_tokens": self.meter.output_tokens, "total_tokens": self.meter.total_tokens, "usage_source": self.meter.usage_source, "context_tokens": self.meter.context_tokens, } def _apply_estimate_fallback(self) -> None: """provider 没返回任何 usage 时，用缓冲区文本估算兜底。 不兜底的话，这类请求会以 0 token 入账——用户白用、额度不受控， 而且从账面上完全看不出来。兜底后至少 usage_source=estimate 能 标示这批数字不可尽信。 """ if self.meter.total_tokens > 0: return estimated = self.meter.estimate_from_buffer() if estimated > 0: logger.warning( "本次未取到 provider usage，按缓冲区估算 %d tokens（usage_source=estimate）" " session=%s", estimated, self.req.session_id, ) self.meter.note_estimated(estimated) async def _persist_context(self, snapshot: Any) -> None: """落 session_context_usage，并在发生压缩时登记原文存档位置。 存档优先用**框架自己的产物**：摘要中间件会把被淘汰的历史追加写到 ``/conversation_history/{thread_id}.md``，并把路径嵌进摘要正文—— 那份文件是权威的，agent 也能自己读回来。只有框架产物不存在时才 回落写我们自己的缓冲区，避免「以为有存档其实没有」。 """ archive_path = "" try: if snapshot.compacted: archive_path = await asyncio.to_thread( self._archive_compaction, snapshot.archive_text, snapshot.archive_tokens ) await asyncio.to_thread(chat_store.save_context_usage, session_id=self.req.session_id, turn=chat_store.count_messages(self.req.session_id) // 2, input_tokens=snapshot.input_tokens, output_tokens=snapshot.output_tokens, cache_read_tokens=snapshot.cache_read_tokens, usage_source=snapshot.usage_source, compacted=snapshot.compacted, compact_count=snapshot.compact_count, snapshot={ "context_tokens": snapshot.context_tokens, "context_window": self.meter.window, "model_calls": snapshot.model_calls, "archive_path": archive_path, }, ) except Exception as exc: logger.exception( "上下文用量落库失败 session=%s：%s: %s", self.req.session_id, type(exc).__name__, exc, ) def _archive_compaction(self, buffered_text: str | None, buffered_tokens: int) -> str: """登记（必要时写出）压缩前原文，返回存档路径或空串。""" from agents.workspace import session_root history_dir = session_root(self.req.user_name, self.req.session_id) / "conversation_history" framework_file = history_dir / f"{self.req.thread_id}.md" if framework_file.exists(): path = str(framework_file) chat_store.save_context_archive( self.req.session_id, 0, path, buffered_tokens ) logger.info("压缩原文由框架落盘：%s", path) return path if not buffered_text: logger.warning( "检测到压缩但既无框架落盘也无本地缓冲，无法存档 session=%s", self.req.session_id, ) return "" archive_dir = history_dir archive_dir.mkdir(parents=True, exist_ok=True) stamp = time.strftime("%Y%m%d-%H%M%S") target = archive_dir / f"{self.req.session_id}-{stamp}.md" target.write_text(buffered_text, encoding="utf-8") chat_store.save_context_archive( self.req.session_id, 0, str(target), buffered_tokens ) logger.info("压缩原文由本进程落盘：%s（%d tokens）", target, buffered_tokens) return str(target) # ======================================================== # 异常 # ======================================================== async def _error_events(self, exc: Exception) -> AsyncIterator[ExecutionEvent]: """把异常翻译成 error 帧 + 收尾帧。""" from langgraph.errors import GraphRecursionError if isinstance(exc, ExecutionOutputError): logger.error(str(exc), extra={'kv': {'event': 'execution.output.failed', 'error_code': exc.code, 'diagnostics': self.diagnostics}}) yield self.writer.error(str(exc), code=exc.code) async for frame in self._completion(aborted=True): yield frame return if isinstance(exc, GraphRecursionError): logger.warning("图递归上限触发 session=%s：%s", self.req.session_id, exc) yield self.writer.error( "任务步骤过多已被中止，请把任务拆成更小的步骤重试", code="recursion_limit" ) async for frame in self._completion(aborted=True): yield frame return if isinstance(exc, concurrency.TaskWallTimeoutError): logger.warning("任务墙钟超时 session=%s：%s", self.req.session_id, exc) yield self.writer.error(str(exc), code="task_timeout") async for frame in self._completion(aborted=True): yield frame return logger.exception("流式执行异常 session=%s", self.req.session_id) yield self.writer.error( f"服务内部错误：{type(exc).__name__}", code="internal_error" ) async for frame in self._completion(aborted=True): yield frame # ======================================================== # 资源 # ======================================================== def _finalize(self) -> None: """释放并发位并兜底结算。幂等。""" try: if not self.billing.settled: self.billing.settle() finally: if self.req.slot is not None: self.req.slot.release() def _meter_for_request() -> ContextMeter: """给本轮请求建一个上下文计量器，窗口取自模型配置。""" from agents import model_factory try: window = model_factory.context_window() except Exception: # pragma: no cover - 配置异常不该阻断对话 window = 0 return ContextMeter(window=window)