From 78454bbb536b84ff37837272792f775805c191a5 Mon Sep 17 00:00:00 2001 From: cx-hub666 Date: Sat, 4 Jul 2026 20:22:30 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat(agent-eval):=20=E6=B7=BB=E5=8A=A0?= =?UTF-8?q?=E5=8F=AF=E9=80=89=E8=BD=A8=E8=BF=B9=E6=91=98=E8=A6=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../yuxi/services/agent_invocation_service.py | 141 +++++++++++- .../server/routers/agent_invocation_router.py | 2 + .../services/test_agent_call_run_service.py | 215 ++++++++++++++++++ docs/develop-guides/changelog.md | 2 +- 4 files changed, 358 insertions(+), 2 deletions(-) diff --git a/backend/package/yuxi/services/agent_invocation_service.py b/backend/package/yuxi/services/agent_invocation_service.py index 2c34c5256..7194aba74 100644 --- a/backend/package/yuxi/services/agent_invocation_service.py +++ b/backend/package/yuxi/services/agent_invocation_service.py @@ -34,11 +34,14 @@ build_chat_input_message, build_chat_input_message_from_openai_content, ) +from yuxi.services.run_queue_service import list_run_stream_events from yuxi.storage.postgres.models_business import User +from yuxi.utils.logging_config import logger EVALUATION_SOURCE = "agent_evaluation" EVALUATION_FIELDS = ("dataset_name", "dataset_item_id", "experiment_name") MAX_REQUEST_ID_LENGTH = 64 +TRAJECTORY_SUMMARY_EVENT_LIMIT = 500 async def create_agent_call_run_view( @@ -116,6 +119,7 @@ async def create_agent_eval_run_view( model_spec: str | None, current_user: User, db: AsyncSession, + include_trajectory_summary: bool = False, ) -> dict[str, Any]: """创建一次评估样例运行，并阻塞等待最终 AgentRun 结果。 @@ -143,7 +147,7 @@ async def create_agent_eval_run_view( attachment_file_ids=meta.get("attachment_file_ids") or [], ) try: - return await await_agent_run_result(run_id=run_response["run_id"], current_uid=str(current_user.uid)) + result = await await_agent_run_result(run_id=run_response["run_id"], current_uid=str(current_user.uid)) except AgentRunWaitTimeout as exc: raise HTTPException( status_code=504, @@ -152,6 +156,15 @@ async def create_agent_eval_run_view( "run": exc.result, }, ) from exc + if include_trajectory_summary: + try: + trajectory_summary = await _load_trajectory_summary(run_response["run_id"]) + if result.get("langfuse_trace_id"): + trajectory_summary["langfuse_trace_id"] = result["langfuse_trace_id"] + result["trajectory_summary"] = trajectory_summary + except Exception as e: + logger.warning(f"Failed to load trajectory summary for run {run_response['run_id']}: {e}") + return result async def create_agent_invocation_run_view( @@ -238,6 +251,132 @@ async def get_agent_call_run_result_view( return _build_agent_call_response(result) +def _sequence_value(event: dict[str, Any] | None) -> str | None: + if not isinstance(event, dict): + return None + seq = event.get("seq") + return str(seq) if seq is not None else None + + +def _empty_trajectory_summary(events: list[dict[str, Any]]) -> dict[str, Any]: + return { + "schema_version": 1, + "source": "run_events", + "event_count": len(events), + "events_truncated": len(events) >= TRAJECTORY_SUMMARY_EVENT_LIMIT, + "event_range": { + "first_seq": _sequence_value(events[0]) if events else None, + "last_seq": _sequence_value(events[-1]) if events else None, + }, + "tool_call_count": 0, + "tool_error_count": 0, + "interrupt_count": 0, + "tools": [], + } + + +def _event_payload(event: dict[str, Any]) -> dict[str, Any]: + envelope = event.get("payload") + if not isinstance(envelope, dict): + return {} + payload = envelope.get("payload") + return payload if isinstance(payload, dict) else {} + + +def _iter_event_chunks(event: dict[str, Any]): + payload = _event_payload(event) + items = payload.get("items") + if isinstance(items, list): + for item in items: + if isinstance(item, dict): + yield item + + chunk = payload.get("chunk") + if isinstance(chunk, dict): + yield chunk + + +def _tool_key(tool_call_id: str | None, name: str, fallback_index: int) -> str: + if tool_call_id: + return str(tool_call_id) + return f"name:{name}:{fallback_index}" + + +def _build_trajectory_summary(events: list[dict[str, Any]]) -> dict[str, Any]: + summary = _empty_trajectory_summary(events) + tool_calls: dict[str, str] = {} + tool_errors: set[str] = set() + open_tool_keys: dict[str, list[str]] = {} + fallback_index = 0 + + def _next_tool_key(tool_call_id: str | None, name: str, *, is_start: bool, is_finish: bool) -> str: + nonlocal fallback_index + if tool_call_id: + return str(tool_call_id) + if is_finish and open_tool_keys.get(name): + return open_tool_keys[name].pop(0) + + key = _tool_key(None, name, fallback_index) + fallback_index += 1 + if is_start and not is_finish: + open_tool_keys.setdefault(name, []).append(key) + return key + + for event in events: + event_type = event.get("event_type") + if event_type == "interrupt": + summary["interrupt_count"] += 1 + + for chunk in _iter_event_chunks(event): + if event_type != "interrupt" and chunk.get("status") in { + "ask_user_question_required", + "human_approval_required", + "interrupted", + }: + summary["interrupt_count"] += 1 + + stream_event = chunk.get("stream_event") + if isinstance(stream_event, dict) and stream_event.get("type") == "tool_call": + name = str(stream_event.get("name") or "unknown") + key = _next_tool_key(stream_event.get("tool_call_id"), name, is_start=True, is_finish=False) + tool_calls.setdefault(key, name) + + tool_event = chunk.get("event") + tool_event_data = tool_event.get("data") if isinstance(tool_event, dict) else None + if not isinstance(tool_event_data, dict): + continue + + name = str(tool_event_data.get("tool_name") or tool_event_data.get("name") or "unknown") + tool_event_type = tool_event_data.get("event") + key = _next_tool_key( + tool_event_data.get("tool_call_id"), + name, + is_start=tool_event_type == "tool-started", + is_finish=tool_event_type == "tool-finished", + ) + if tool_event_type == "tool-started" or key not in tool_calls: + tool_calls.setdefault(key, name) + if tool_event_data.get("error") or event_type == "error": + tool_errors.add(key) + + tools_by_name: dict[str, dict[str, Any]] = {} + for key, name in tool_calls.items(): + item = tools_by_name.setdefault(name, {"name": name, "call_count": 0, "error_count": 0}) + item["call_count"] += 1 + if key in tool_errors: + item["error_count"] += 1 + + summary["tool_call_count"] = len(tool_calls) + summary["tool_error_count"] = len(tool_errors) + summary["tools"] = sorted(tools_by_name.values(), key=lambda item: item["name"]) + return summary + + +async def _load_trajectory_summary(run_id: str) -> dict[str, Any]: + events = await list_run_stream_events(run_id, after_seq="0-0", limit=TRAJECTORY_SUMMARY_EVENT_LIMIT) + return _build_trajectory_summary(events) + + def _normalize_required_text(value: str | None, *, field_name: str) -> str: normalized = str(value or "").strip() if not normalized: diff --git a/backend/server/routers/agent_invocation_router.py b/backend/server/routers/agent_invocation_router.py index fb5004e1d..2ece6ac39 100644 --- a/backend/server/routers/agent_invocation_router.py +++ b/backend/server/routers/agent_invocation_router.py @@ -49,6 +49,7 @@ class AgentEvalRunCreate(BaseModel): meta: dict = Field(default_factory=dict, description="可选，请求追踪信息，例如 request_id、attachment_file_ids") image_content: str | None = Field(None, description="可选，base64 图片内容") model_spec: str | None = Field(None, description="可选，对话级模型覆盖，优先级高于智能体配置") + include_trajectory_summary: bool = Field(False, description="是否返回轻量工具调用轨迹摘要") @agent_invocation_router.post("/agent-call/runs") @@ -101,6 +102,7 @@ async def create_agent_eval_run( meta=dict(payload.meta or {}), image_content=payload.image_content, model_spec=payload.model_spec, + include_trajectory_summary=payload.include_trajectory_summary, current_user=current_user, db=db, ) diff --git a/backend/test/unit/services/test_agent_call_run_service.py b/backend/test/unit/services/test_agent_call_run_service.py index 99032dfda..ff54fb679 100644 --- a/backend/test/unit/services/test_agent_call_run_service.py +++ b/backend/test/unit/services/test_agent_call_run_service.py @@ -296,3 +296,218 @@ async def fake_await_agent_run_result(*, run_id: str, current_uid: str): assert calls["run_kwargs"]["requested_thread_id"] == "" assert calls["run_kwargs"]["request_id"] == "eval-req" assert calls["await_kwargs"] == {"run_id": "run-1", "current_uid": "user-1"} + + +@pytest.mark.asyncio +async def test_create_agent_eval_run_adds_trajectory_summary_when_requested(monkeypatch: pytest.MonkeyPatch): + calls: dict[str, object] = {} + current_user = SimpleNamespace(uid="user-1", role="user") + + async def fake_create_agent_invocation_run_view(**kwargs): + calls["run_kwargs"] = kwargs + return { + "run_id": "run-1", + "thread_id": "thread-1", + "status": "pending", + "request_id": kwargs["request_id"], + } + + async def fake_await_agent_run_result(*, run_id: str, current_uid: str): + return { + "status": "completed", + "agent_run_id": run_id, + "request_id": "eval-req", + "output": "ok", + "langfuse_trace_id": "trace-1", + } + + async def fake_load_trajectory_summary(run_id: str): + calls["trajectory_run_id"] = run_id + return {"tool_call_count": 1, "tools": [{"name": "web_search", "call_count": 1, "error_count": 0}]} + + monkeypatch.setattr(svc, "create_agent_invocation_run_view", fake_create_agent_invocation_run_view) + monkeypatch.setattr(svc, "await_agent_run_result", fake_await_agent_run_result) + monkeypatch.setattr(svc, "_load_trajectory_summary", fake_load_trajectory_summary) + + result = await svc.create_agent_eval_run_view( + query="question", + agent_slug="default-chatbot", + evaluation={}, + meta={"request_id": "eval-req"}, + image_content=None, + model_spec=None, + current_user=current_user, + db=object(), + include_trajectory_summary=True, + ) + + assert calls["trajectory_run_id"] == "run-1" + assert result["trajectory_summary"] == { + "tool_call_count": 1, + "tools": [{"name": "web_search", "call_count": 1, "error_count": 0}], + "langfuse_trace_id": "trace-1", + } + + +@pytest.mark.asyncio +async def test_create_agent_eval_run_ignores_trajectory_summary_load_errors(monkeypatch: pytest.MonkeyPatch): + calls: dict[str, object] = {} + current_user = SimpleNamespace(uid="user-1", role="user") + + async def fake_create_agent_invocation_run_view(**kwargs): + return { + "run_id": "run-1", + "thread_id": "thread-1", + "status": "pending", + "request_id": kwargs["request_id"], + } + + async def fake_await_agent_run_result(*, run_id: str, current_uid: str): + del current_uid + return {"status": "completed", "agent_run_id": run_id, "request_id": "eval-req", "output": "ok"} + + async def fake_load_trajectory_summary(run_id: str): + calls["trajectory_run_id"] = run_id + raise RuntimeError("redis unavailable") + + monkeypatch.setattr(svc, "create_agent_invocation_run_view", fake_create_agent_invocation_run_view) + monkeypatch.setattr(svc, "await_agent_run_result", fake_await_agent_run_result) + monkeypatch.setattr(svc, "_load_trajectory_summary", fake_load_trajectory_summary) + + result = await svc.create_agent_eval_run_view( + query="question", + agent_slug="default-chatbot", + evaluation={}, + meta={"request_id": "eval-req"}, + image_content=None, + model_spec=None, + current_user=current_user, + db=object(), + include_trajectory_summary=True, + ) + + assert calls["trajectory_run_id"] == "run-1" + assert result == {"status": "completed", "agent_run_id": "run-1", "request_id": "eval-req", "output": "ok"} + + +def test_build_trajectory_summary_counts_tools_interrupts_and_errors(): + summary = svc._build_trajectory_summary( + [ + { + "seq": "1-0", + "event_type": "messages", + "payload": { + "payload": { + "items": [ + { + "stream_event": { + "type": "tool_call", + "tool_call_id": "call-search", + "name": "web_search", + } + } + ] + } + }, + }, + { + "seq": "2-0", + "event_type": "error", + "payload": { + "payload": { + "chunk": { + "event": { + "data": { + "event": "tool-finished", + "tool_call_id": "call-search", + "tool_name": "web_search", + "error": "timeout", + } + } + } + } + }, + }, + {"seq": "3-0", "event_type": "interrupt", "payload": {"payload": {"reason": "human_approval"}}}, + ] + ) + + assert summary == { + "schema_version": 1, + "source": "run_events", + "event_count": 3, + "events_truncated": False, + "event_range": {"first_seq": "1-0", "last_seq": "3-0"}, + "tool_call_count": 1, + "tool_error_count": 1, + "interrupt_count": 1, + "tools": [{"name": "web_search", "call_count": 1, "error_count": 1}], + } + + +def test_build_trajectory_summary_matches_no_id_tool_finish_to_start(): + summary = svc._build_trajectory_summary( + [ + { + "seq": None, + "event_type": "messages", + "payload": { + "payload": { + "items": [ + { + "stream_event": { + "type": "tool_call", + "name": "read_file", + } + } + ] + } + }, + }, + { + "seq": None, + "event_type": "error", + "payload": { + "payload": { + "chunk": { + "event": { + "data": { + "event": "tool-finished", + "tool_name": "read_file", + "error": "file not found", + } + } + } + } + }, + }, + ] + ) + + assert summary["event_range"] == {"first_seq": None, "last_seq": None} + assert summary["tool_call_count"] == 1 + assert summary["tool_error_count"] == 1 + assert summary["tools"] == [{"name": "read_file", "call_count": 1, "error_count": 1}] + + +@pytest.mark.asyncio +async def test_load_trajectory_summary_reads_run_events_with_limit(monkeypatch: pytest.MonkeyPatch): + calls: dict[str, object] = {} + + async def fake_list_run_stream_events(run_id: str, *, after_seq: str, limit: int): + calls["run_id"] = run_id + calls["after_seq"] = after_seq + calls["limit"] = limit + return [] + + monkeypatch.setattr(svc, "list_run_stream_events", fake_list_run_stream_events) + + summary = await svc._load_trajectory_summary("run-1") + + assert calls == { + "run_id": "run-1", + "after_seq": "0-0", + "limit": svc.TRAJECTORY_SUMMARY_EVENT_LIMIT, + } + assert summary["event_count"] == 0 + assert summary["tools"] == [] diff --git a/docs/develop-guides/changelog.md b/docs/develop-guides/changelog.md index c15b33e04..5e4c4f44b 100644 --- a/docs/develop-guides/changelog.md +++ b/docs/develop-guides/changelog.md @@ -56,7 +56,7 @@ - 新增系统配置 Redis 快照同步：管理员保存配置时仍以 `saves/config/base.toml` 作为唯一持久化来源，成功写入后将可运行时同步的公开配置字段写入 `yuxi:runtime_config`；API 与 worker 进程在启动时各拉起一个后台同步线程，按 5 秒间隔从快照刷新内存值，读取端按普通属性访问、无需感知，Redis 不可用时继续使用当前内存值。`save_dir` 是启动期内部路径配置，不在管理员配置中展示、不从 `base.toml` 读取、不写入 Redis 快照且不支持通过管理员配置接口修改；sandbox 相关配置仍属于启动期敏感配置，运行中的已初始化组件不承诺完整热更新，修改后仍需重启保证生效；移除已无运行时调用点的 `enable_reranker` 与 `default_agent_id` 配置字段。 - 优化 FastAPI 请求链路并发能力：Milvus 知识库检索中的同步 embedding、向量/BM25/混合检索调用，以及图谱查询中的同步 Milvus/Neo4j 读操作（含连接建立）统一通过有界 `asyncio.to_thread` 在线程中执行，避免阻塞 API 事件循环；并发上限按事件循环懒加载信号量控制，不改变检索默认行为与参数上限。 - 改进 OpenAI 兼容提供商流式工具调用兼容（替代 v0.7.0 的按 provider 禁流式处理）：根因是 LangGraph v3 流式累积对 tool_call 字段“后值覆盖”，SiliconFlow、阿里云百炼等在参数续片里把 `name`/`id` 下发为空字符串覆盖首片真实值。改为 `_ToolCallChunkFixChatOpenAI` 把续片空串 `name`/`id` 归一化为 `None`，对所有 OpenAI 兼容 provider 通用生效且保留流式，移除原 `_NON_STREAMING_TOOL_CALL_PROVIDERS` 名单。 -- 新增 Agent 评估运行入口：`POST /api/agent-invocation/eval/runs` 会创建正常对话与 AgentRun，复用 worker 执行链路，并以 `source=agent_evaluation` 与 `agent_invocation_meta.evaluation` 标记写入 conversation、AgentRun 输入消息与 Langfuse trace；接口阻塞至运行结束后直接返回最终结果（状态、最终 assistant 输出、Langfuse trace id）。`yuxi-cli` 新增 `yuxi agent eval` 命令，用于从 Langfuse 数据集读取输入并回传实验输出 +- 新增 Agent 评估运行入口：`POST /api/agent-invocation/eval/runs` 会创建正常对话与 AgentRun，复用 worker 执行链路，并以 `source=agent_evaluation` 与 `agent_invocation_meta.evaluation` 标记写入 conversation、AgentRun 输入消息与 Langfuse trace；接口阻塞至运行结束后直接返回最终结果（状态、最终 assistant 输出、Langfuse trace id），并支持通过 `include_trajectory_summary` 按需返回轻量工具调用轨迹摘要。`yuxi-cli` 新增 `yuxi agent eval` 命令，用于从 Langfuse 数据集读取输入并回传实验输出 - 对话消息点赞/点踩反馈接入 Langfuse score：本地 `MessageFeedback` 保存成功后，如助手消息已关联 Langfuse trace，则同步写入 `user-feedback` score，点赞为 `1`、点踩为 `0`，点踩原因写入 comment，便于在 Langfuse 中按用户反馈筛选 trace。 - 新增外部系统 Agent 调用入口：独立 `agent-invocation` router 提供 `POST /api/agent-invocation/agent-call/runs` 与 `POST /api/agent-invocation/agent-call/runs/result`，字段沿用 Yuxi 命名（`agent_slug/thread_id/request_id/model_spec`），复用 AgentRun 队列和结果读取能力；支持非流式同步等待或 `async_mode=true` 立即返回 `run_id`，Agent Call 不允许通过 `agent_call_meta.context` 覆盖 Agent context，运行时模型覆盖只允许走独立 `model_spec`；修复无 `thread_id` 且模型校验失败时提前提交空对话，导致孤儿对话和 `request_id` 失败重试非幂等的问题；Agent Call 的 `messages[].content` 兼容 OpenAI 风格的 `text`/`image_url` 多模态数组，纯文本数组不再误报 422，图片输入会保留原始 LangChain 多模态消息供 AgentRun worker 恢复；Agent Eval 与 Agent Call 统一通过 conversation-backed invocation helper 创建 run，后续定时任务等入口只需做请求解析和结果出口适配。 - 修复 Agent Invocation 创建的 eval/call 对话进入用户对话导航的问题：侧边栏最近对话与对话搜索会按 conversation metadata `source` 排除 `agent_evaluation` 与 `agent_call`，保留 run/conversation 持久化与结果追踪能力。 From e2aab2ccbe702c59f1a1dd73b3d4ec6fa9f3b570 Mon Sep 17 00:00:00 2001 From: cx-hub666 Date: Sun, 5 Jul 2026 14:12:27 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix(agent-eval):=20=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=E4=B8=AD=E6=96=AD=E8=BD=A8=E8=BF=B9=E9=87=8D=E5=A4=8D=E8=AE=A1?= =?UTF-8?q?=E6=95=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../yuxi/services/agent_invocation_service.py | 119 +++++++++--------- .../services/test_agent_call_run_service.py | 24 ++++ 2 files changed, 81 insertions(+), 62 deletions(-) diff --git a/backend/package/yuxi/services/agent_invocation_service.py b/backend/package/yuxi/services/agent_invocation_service.py index 7194aba74..5c25f7889 100644 --- a/backend/package/yuxi/services/agent_invocation_service.py +++ b/backend/package/yuxi/services/agent_invocation_service.py @@ -42,6 +42,7 @@ EVALUATION_FIELDS = ("dataset_name", "dataset_item_id", "experiment_name") MAX_REQUEST_ID_LENGTH = 64 TRAJECTORY_SUMMARY_EVENT_LIMIT = 500 +INTERRUPT_STATUSES = {"ask_user_question_required", "human_approval_required", "interrupted"} async def create_agent_call_run_view( @@ -251,72 +252,28 @@ async def get_agent_call_run_result_view( return _build_agent_call_response(result) -def _sequence_value(event: dict[str, Any] | None) -> str | None: - if not isinstance(event, dict): - return None - seq = event.get("seq") - return str(seq) if seq is not None else None - - -def _empty_trajectory_summary(events: list[dict[str, Any]]) -> dict[str, Any]: - return { - "schema_version": 1, - "source": "run_events", - "event_count": len(events), - "events_truncated": len(events) >= TRAJECTORY_SUMMARY_EVENT_LIMIT, - "event_range": { - "first_seq": _sequence_value(events[0]) if events else None, - "last_seq": _sequence_value(events[-1]) if events else None, - }, - "tool_call_count": 0, - "tool_error_count": 0, - "interrupt_count": 0, - "tools": [], - } - - -def _event_payload(event: dict[str, Any]) -> dict[str, Any]: - envelope = event.get("payload") - if not isinstance(envelope, dict): - return {} - payload = envelope.get("payload") - return payload if isinstance(payload, dict) else {} - - -def _iter_event_chunks(event: dict[str, Any]): - payload = _event_payload(event) - items = payload.get("items") - if isinstance(items, list): - for item in items: - if isinstance(item, dict): - yield item - - chunk = payload.get("chunk") - if isinstance(chunk, dict): - yield chunk - - -def _tool_key(tool_call_id: str | None, name: str, fallback_index: int) -> str: - if tool_call_id: - return str(tool_call_id) - return f"name:{name}:{fallback_index}" +async def _load_trajectory_summary(run_id: str) -> dict[str, Any]: + """从 run event stream 读取有限事件并生成轻量轨迹摘要。""" + events = await list_run_stream_events(run_id, after_seq="0-0", limit=TRAJECTORY_SUMMARY_EVENT_LIMIT) + return _build_trajectory_summary(events) def _build_trajectory_summary(events: list[dict[str, Any]]) -> dict[str, Any]: - summary = _empty_trajectory_summary(events) + """聚合工具调用、工具错误和人工中断计数，避免暴露完整事件载荷。""" + summary = _trajectory_summary_base(events) tool_calls: dict[str, str] = {} tool_errors: set[str] = set() open_tool_keys: dict[str, list[str]] = {} fallback_index = 0 - def _next_tool_key(tool_call_id: str | None, name: str, *, is_start: bool, is_finish: bool) -> str: + def next_tool_key(tool_call_id: str | None, name: str, *, is_start: bool, is_finish: bool) -> str: nonlocal fallback_index if tool_call_id: return str(tool_call_id) if is_finish and open_tool_keys.get(name): return open_tool_keys[name].pop(0) - key = _tool_key(None, name, fallback_index) + key = f"name:{name}:{fallback_index}" fallback_index += 1 if is_start and not is_finish: open_tool_keys.setdefault(name, []).append(key) @@ -328,17 +285,13 @@ def _next_tool_key(tool_call_id: str | None, name: str, *, is_start: bool, is_fi summary["interrupt_count"] += 1 for chunk in _iter_event_chunks(event): - if event_type != "interrupt" and chunk.get("status") in { - "ask_user_question_required", - "human_approval_required", - "interrupted", - }: + if _is_interrupt_status_event(event_type, chunk): summary["interrupt_count"] += 1 stream_event = chunk.get("stream_event") if isinstance(stream_event, dict) and stream_event.get("type") == "tool_call": name = str(stream_event.get("name") or "unknown") - key = _next_tool_key(stream_event.get("tool_call_id"), name, is_start=True, is_finish=False) + key = next_tool_key(stream_event.get("tool_call_id"), name, is_start=True, is_finish=False) tool_calls.setdefault(key, name) tool_event = chunk.get("event") @@ -348,7 +301,7 @@ def _next_tool_key(tool_call_id: str | None, name: str, *, is_start: bool, is_fi name = str(tool_event_data.get("tool_name") or tool_event_data.get("name") or "unknown") tool_event_type = tool_event_data.get("event") - key = _next_tool_key( + key = next_tool_key( tool_event_data.get("tool_call_id"), name, is_start=tool_event_type == "tool-started", @@ -372,9 +325,51 @@ def _next_tool_key(tool_call_id: str | None, name: str, *, is_start: bool, is_fi return summary -async def _load_trajectory_summary(run_id: str) -> dict[str, Any]: - events = await list_run_stream_events(run_id, after_seq="0-0", limit=TRAJECTORY_SUMMARY_EVENT_LIMIT) - return _build_trajectory_summary(events) +def _trajectory_summary_base(events: list[dict[str, Any]]) -> dict[str, Any]: + """创建轨迹摘要的固定字段，后续只填充聚合计数。""" + first_seq = _event_seq(events[0]) if events else None + last_seq = _event_seq(events[-1]) if events else None + return { + "schema_version": 1, + "source": "run_events", + "event_count": len(events), + "events_truncated": len(events) >= TRAJECTORY_SUMMARY_EVENT_LIMIT, + "event_range": {"first_seq": first_seq, "last_seq": last_seq}, + "tool_call_count": 0, + "tool_error_count": 0, + "interrupt_count": 0, + "tools": [], + } + + +def _event_seq(event: dict[str, Any]) -> str | None: + seq = event.get("seq") + return str(seq) if seq is not None else None + + +def _iter_event_chunks(event: dict[str, Any]): + payload = _event_payload(event) + items = payload.get("items") + if isinstance(items, list): + for item in items: + if isinstance(item, dict): + yield item + + chunk = payload.get("chunk") + if isinstance(chunk, dict): + yield chunk + + +def _event_payload(event: dict[str, Any]) -> dict[str, Any]: + envelope = event.get("payload") + if not isinstance(envelope, dict): + return {} + payload = envelope.get("payload") + return payload if isinstance(payload, dict) else {} + + +def _is_interrupt_status_event(event_type: str | None, chunk: dict[str, Any]) -> bool: + return event_type not in {"interrupt", "end"} and chunk.get("status") in INTERRUPT_STATUSES def _normalize_required_text(value: str | None, *, field_name: str) -> str: diff --git a/backend/test/unit/services/test_agent_call_run_service.py b/backend/test/unit/services/test_agent_call_run_service.py index ff54fb679..8e991f6fc 100644 --- a/backend/test/unit/services/test_agent_call_run_service.py +++ b/backend/test/unit/services/test_agent_call_run_service.py @@ -445,6 +445,30 @@ def test_build_trajectory_summary_counts_tools_interrupts_and_errors(): } +def test_build_trajectory_summary_counts_human_interrupt_once_with_end_event(): + interrupt_chunk = { + "status": "human_approval_required", + "message": "approve?", + } + + summary = svc._build_trajectory_summary( + [ + { + "seq": "1-0", + "event_type": "interrupt", + "payload": {"payload": {"reason": "human_approval", "chunk": interrupt_chunk}}, + }, + { + "seq": "2-0", + "event_type": "end", + "payload": {"payload": {"status": "interrupted", "chunk": interrupt_chunk}}, + }, + ] + ) + + assert summary["interrupt_count"] == 1 + + def test_build_trajectory_summary_matches_no_id_tool_finish_to_start(): summary = svc._build_trajectory_summary( [