|
|
@@ -1,819 +1,27 @@
|
|
|
-"""SSE 流式内核。
|
|
|
+"""HTTP 流适配器:编码事件并传播断连,不执行 Agent 业务逻辑。"""
|
|
|
+from contextlib import aclosing
|
|
|
+from typing import AsyncIterator
|
|
|
|
|
|
-职责:把 LangGraph 的 ``astream`` 事件流翻译成 1.5 兼容的 SSE 帧,并在
|
|
|
-结束时完成落库、计费、来源标注与上下文计量。
|
|
|
+from runtime.engine import Execution, ExecutionRequest
|
|
|
+from runtime.interrupts import pending_interrupts # 兼容旧导入
|
|
|
+from api.sse_events import serialize_event
|
|
|
|
|
|
-## 通道分工(重要,避免重复事件)
|
|
|
|
|
|
-``stream_mode=["updates", "messages"]`` 两条通道都在推同一个 agent 的进展,
|
|
|
-不加区分会导致同一个工具调用被发两次。这里的分工是:
|
|
|
+async def stream_sse(execution: Execution) -> AsyncIterator[str]:
|
|
|
+ async with aclosing(execution.events()) as events:
|
|
|
+ async for event in events:
|
|
|
+ yield serialize_event(event)
|
|
|
|
|
|
-- **updates 通道**:工具调用/结果、待办清单、子智能体生命周期、状态提示。
|
|
|
- 走这里的原因是它给的是**完整的消息对象**(``tool_calls`` 里 args 已解析
|
|
|
- 完毕、``ToolMessage`` 已成对),不需要在流式增量里拼半个 JSON。
|
|
|
-- **messages 通道**:正文 token、思考 token、真实 token 用量。
|
|
|
|
|
|
-``version="v2"`` 下每个 chunk 是 ``{"type", "ns", "data"}``;``subgraphs=True``
|
|
|
-让子智能体的产出带 ``ns``。两个格式都做了兼容(旧版是 ``(mode, data)`` 元组)。
|
|
|
+# 旧内部调用入口保留;新入口直接使用 runtime.execution 和 stream_sse。
|
|
|
+StreamRequest = ExecutionRequest
|
|
|
|
|
|
-## 与 1.5 的差异
|
|
|
|
|
|
-1.5 的 sse_core 是 1908 行,里面混着计费、压缩、Word 导出、来源刮取等多条
|
|
|
-业务线。2.0 把这些拆成独立组件(context_meter / source_tracker /
|
|
|
-subagent_tracker),内核只负责「读流 → 派事件 → 收尾」。
|
|
|
-"""
|
|
|
-
|
|
|
-from __future__ import annotations
|
|
|
-
|
|
|
-import asyncio
|
|
|
-import time
|
|
|
-from dataclasses import dataclass, field
|
|
|
-from pathlib import Path
|
|
|
-from datetime import datetime
|
|
|
-
|
|
|
-import anyio
|
|
|
-from typing import Any, AsyncIterator
|
|
|
-
|
|
|
-import config
|
|
|
-from api import concurrency
|
|
|
-from api.interrupts import pack, interrupt_frame
|
|
|
-from api.context_meter import ContextMeter
|
|
|
-from api.sse_events import EventWriter
|
|
|
-from api.source_tracker import SourceTracker
|
|
|
-from api.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
|
|
|
-
|
|
|
-logger = get_logger(__name__)
|
|
|
-
|
|
|
-MAIN_AGENT_NAME = "deep-agent"
|
|
|
-MAIN_AGENT_CN = "深度智能体"
|
|
|
-
|
|
|
-# 摘要中间件的模型调用也在 messages 通道里推 token,靠这个标记过滤掉,
|
|
|
-# 否则摘要正文会混进给用户看的回答里
|
|
|
-_SUMMARIZATION_SOURCE = "summarization"
|
|
|
-
|
|
|
-
|
|
|
-@dataclass
|
|
|
-class StreamRequest:
|
|
|
- """一次流式请求的全部入参。"""
|
|
|
-
|
|
|
- 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
|
|
|
- run_id: str = field(default_factory=db_conn.new_id)
|
|
|
-
|
|
|
-
|
|
|
-def normalize_chunk(chunk: Any) -> tuple[str | None, tuple[Any, ...], Any]:
|
|
|
- """把不同版本的 astream chunk 归一成 ``(mode, ns, data)``。
|
|
|
-
|
|
|
- 兼容三种形态:v2 的 ``{"type","ns","data"}`` 字典、旧版 ``(mode, data)``
|
|
|
- 元组、以及无法识别的原样返回(mode=None,上层据此忽略)。
|
|
|
- """
|
|
|
- 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]
|
|
|
- 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 ChatStream:
|
|
|
- """一次 /chat 请求的流式执行。"""
|
|
|
-
|
|
|
- def __init__(self, request: StreamRequest) -> None:
|
|
|
- self.req = request
|
|
|
- self.writer = EventWriter(request.session_id, request.trace_id or get_trace_id())
|
|
|
- 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
|
|
|
-
|
|
|
- # ========================================================
|
|
|
- # 主入口
|
|
|
- # ========================================================
|
|
|
-
|
|
|
- async def generate(self) -> AsyncIterator[str]:
|
|
|
- """产出整个 SSE 流。保证:计费幂等入账、并发位释放、图游标关闭。
|
|
|
-
|
|
|
- 这里显式绑定 trace_id 而不是依赖 HTTP 中间件设的 ContextVar:
|
|
|
- 响应体的迭代在 Starlette 的中间件调用栈之外执行,中间件里绑的值
|
|
|
- 在生成器里不保证可见。绑定发生在生成器内部,生命周期与本次流一致。
|
|
|
- """
|
|
|
- from obs import bind_trace_id, unbind_trace_id
|
|
|
-
|
|
|
- token = bind_trace_id(self.req.trace_id or None)
|
|
|
- stream = None
|
|
|
- try:
|
|
|
- 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
|
|
|
-
|
|
|
- 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 = type(exc).__name__
|
|
|
- async for frame in self._error_frames(exc):
|
|
|
- yield frame
|
|
|
- finally:
|
|
|
- # 清理与发帧独立。ASGI 断连的取消作用域不能打断保存和释放。
|
|
|
- with anyio.CancelScope(shield=True):
|
|
|
- cleanup = asyncio.create_task(self._cleanup(stream))
|
|
|
- try:
|
|
|
- await asyncio.shield(cleanup)
|
|
|
- except asyncio.CancelledError:
|
|
|
- await cleanup
|
|
|
- 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()
|
|
|
-
|
|
|
-
|
|
|
- # ========================================================
|
|
|
- # 输入准备
|
|
|
- # ========================================================
|
|
|
-
|
|
|
- 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.info("热线程,仅喂当轮消息 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[str]:
|
|
|
- """消费 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)
|
|
|
- if mode == "updates":
|
|
|
- for frame in self._handle_updates(data, ns):
|
|
|
- yield frame
|
|
|
- elif mode == "messages":
|
|
|
- for frame in self._handle_messages(data, ns):
|
|
|
- yield frame
|
|
|
- # mode 为 None:无法识别的 chunk,忽略而不是猜
|
|
|
-
|
|
|
- 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[str]:
|
|
|
- """处理 updates 通道:工具事件、待办、子智能体台账。"""
|
|
|
- from langchain_core.messages import AIMessage, ToolMessage
|
|
|
-
|
|
|
- frames: list[str] = []
|
|
|
- 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
|
|
|
-
|
|
|
- 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):
|
|
|
- frames.extend(self._on_ai_message(message, subagent_type, subagent_id))
|
|
|
-
|
|
|
- return frames
|
|
|
-
|
|
|
- def _on_ai_message(
|
|
|
- self, message: Any, subagent_type: str, subagent_id: str
|
|
|
- ) -> list[str]:
|
|
|
- """处理一条完整的 AIMessage:登记 task 调用、发工具调用卡。"""
|
|
|
- from tools.registry import get_label
|
|
|
-
|
|
|
- frames: list[str] = []
|
|
|
- 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[str]:
|
|
|
- """处理一条 ToolMessage:发工具结果卡、抽取来源、结子智能体。"""
|
|
|
- from tools.registry import get_label
|
|
|
-
|
|
|
- frames: list[str] = []
|
|
|
- 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[str]:
|
|
|
- """处理 messages 通道:正文 token、思考 token、用量计量。"""
|
|
|
- from langchain_core.messages import AIMessage, AIMessageChunk
|
|
|
-
|
|
|
- frames: list[str] = []
|
|
|
- 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.meter.note_text(text)
|
|
|
- self.answer_parts.append(text)
|
|
|
- frames.append(self.writer.token(text))
|
|
|
-
|
|
|
- return frames
|
|
|
-
|
|
|
- # ========================================================
|
|
|
- # 中断
|
|
|
- # ========================================================
|
|
|
-
|
|
|
- async def _emit_interrupts(self) -> AsyncIterator[str]:
|
|
|
- """把中断转成 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) -> str | None:
|
|
|
- return interrupt_frame(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 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.error("落库失败 session=%s:%s: %s", self.req.session_id, type(exc).__name__, exc)
|
|
|
-
|
|
|
- # 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[str]:
|
|
|
- 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
|
|
|
-
|
|
|
- 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.error(
|
|
|
- "上下文用量落库失败 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_frames(self, exc: Exception) -> AsyncIterator[str]:
|
|
|
- """把异常翻译成 error 帧 + 收尾帧。"""
|
|
|
- from langgraph.errors import GraphRecursionError
|
|
|
-
|
|
|
- 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)
|
|
|
+class ChatStream(Execution):
|
|
|
+ def generate(self) -> AsyncIterator[str]:
|
|
|
+ return stream_sse(self)
|
|
|
|
|
|
|
|
|
def sse_headers(session_id: str) -> dict[str, str]:
|
|
|
- """SSE 响应头。``X-Accel-Buffering: no`` 是给 nginx 的:不关掉的话
|
|
|
- 反代会缓冲整个响应,流式效果全无(表现为「一次性吐出全部内容」)。"""
|
|
|
- return {
|
|
|
- "Cache-Control": "no-cache",
|
|
|
- "Connection": "keep-alive",
|
|
|
- "X-Accel-Buffering": "no",
|
|
|
- "X-Session-Id": session_id,
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-async def pending_interrupts(agent: Any, config: dict[str, Any]) -> list[Any]:
|
|
|
- """读取当前挂起的中断载荷列表。
|
|
|
-
|
|
|
- 供两处复用:``GET /chat/interrupt``(刷新页面后补渲染卡片)与
|
|
|
- ``POST /chat/resume``(在构造 decisions 前先知道有几个待审批动作——
|
|
|
- 官方 HITL 要求 decisions 与 action_requests **数量严格一一对应**,
|
|
|
- 数量对不上会直接抛错)。
|
|
|
- """
|
|
|
- get_state = getattr(agent, "aget_state", None)
|
|
|
- if get_state is None:
|
|
|
- return []
|
|
|
- try:
|
|
|
- state = await get_state(config)
|
|
|
- except Exception as exc:
|
|
|
- logger.warning("读取图状态失败:%s: %s", type(exc).__name__, exc)
|
|
|
- raise
|
|
|
- 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
|
|
|
+ return {"Cache-Control": "no-cache", "Connection": "keep-alive",
|
|
|
+ "X-Accel-Buffering": "no", "X-Session-Id": session_id}
|