From 6d31b226ba506a180348411fa71f5e5d26c4892f Mon Sep 17 00:00:00 2001 From: HandsomeAq <310925901@qq.com> Date: Wed, 5 Aug 2026 23:26:00 +0800 Subject: [PATCH] =?UTF-8?q?bug=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- task_receiver.py | 105 +++++++++++++++++++++++++++++++++-------------- 1 file changed, 74 insertions(+), 31 deletions(-) diff --git a/task_receiver.py b/task_receiver.py index cca0ffb..c114d4a 100644 --- a/task_receiver.py +++ b/task_receiver.py @@ -12,18 +12,45 @@ """ import asyncio import logging -import threading +import os from fastapi import APIRouter, HTTPException, Request from fastapi.responses import JSONResponse import gateway_client -from google.adk.runners import InMemoryRunner +from google.adk.runners import Runner +from google.adk.sessions.sqlite_session_service import SqliteSessionService +from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService +from google.adk.agents.run_config import RunConfig, StreamingMode from google.genai import types as genai_types logger = logging.getLogger(__name__) +def _sessions_db_path() -> str: + """返回会话数据库路径（与 chat.py 一致，位于项目 data 目录）。""" + here = os.path.dirname(os.path.abspath(__file__)) + data_dir = os.path.join(here, "data") + os.makedirs(data_dir, exist_ok=True) + return os.path.join(data_dir, "sessions.db") + + +def _first_sentence(text: str) -> str: + """从文本中提取第一个完整句子（用于"Agent 接受任务回复"展示）。 + + ADK 流式事件会把回复拆成多个 text 片段，第一个片段常只有一两个字。这里把 + 累积文本按句子结束符切分，返回第一句完整内容；若没有句子结束符则回退为 + 完整文本。 + """ + if not text: + return "" + for sep in ("。", "！", "？", "!", "?", "\n", ";"): + idx = text.find(sep) + if idx != -1: + return text[: idx + 1].strip() + return text.strip() + + def _payload_to_prompt(payload: dict) -> str: """将网关任务 payload 转换为 agent 的用户指令。""" if not payload: @@ -48,47 +75,53 @@ def create_task_router(app): pass async def _run_agent_once(prompt: str, request_id: str) -> tuple[str, str]: - """用 ADK InMemoryRunner 运行一次 agent，返回 (接受任务后的首条回复, 最终总结)。 + """运行一次 agent，返回 (接受任务后的首条回复, 最终总结)。 - app 是 App 容器（根 agent 非裸 LlmAgent），run_async 不会自动创建 - session，需先用 runner.session_service 显式创建。 + 必须使用 Runner + SqliteSessionService + streaming_mode=SSE（与 chat.py + 一致）：InMemoryRunner 无法驱动带 compaction 配置的 App 容器，会导致 + LLM 不调用、回复为空（"Root node was cancelled"）。 """ - runner = InMemoryRunner(app=app) - session_id = f"task-{request_id}" - await runner.session_service.create_session( - app_name=runner.app_name, - user_id="gateway", - session_id=session_id, + runner = Runner( + app=app, + session_service=SqliteSessionService(db_path=_sessions_db_path()), + artifact_service=InMemoryArtifactService(), + auto_create_session=True, ) + session_id = f"task-{request_id}" + message = genai_types.Content(parts=[genai_types.Part(text=prompt)]) texts: list[str] = [] final_text = "" async for event in runner.run_async( user_id="gateway", session_id=session_id, - new_message=genai_types.Content(role="user", parts=[genai_types.Part(text=prompt)]), + new_message=message, + run_config=RunConfig(streaming_mode=StreamingMode.SSE), ): # 可中断：每收到一个事件检查一次停止标志 if gateway_client.is_stop_requested(request_id): raise _TaskCancelled() - if event.is_final_response(): - if event.content and event.content.parts: - for part in event.content.parts: - text = getattr(part, "text", None) - if text: - final_text += text - break + # 只收集非思考（thought）的用户可见文本：过滤掉 thought 片段 if event.content and event.content.parts: for part in event.content.parts: text = getattr(part, "text", None) - if text: + is_thought = getattr(part, "thought", False) + if not text or is_thought: + continue + if event.is_final_response(): + final_text += text + else: texts.append(text) - # 首条回复 = 接受任务后的第一条回应；最终总结 = final response - reply = (texts[0] if texts else final_text).strip() - summary = final_text.strip() or "\n".join(t for t in texts if t).strip() or "(无输出)" + # 最终总结 = final response 文本；首条回复 = 累积中间文本直到完整句子 + summary = final_text.strip() or "".join(texts).strip() or "(无输出)" + reply = _first_sentence("".join(texts)) or summary return reply, summary - def _execute_and_report(request_id: str, payload: dict) -> None: - """后台线程：执行 agent，成功后回传 success，异常回传 failed，被取消时回传 cancelled。 + async def _execute_and_report(request_id: str, payload: dict) -> None: + """后台执行：执行 agent，成功后回传 success，异常回传 failed，被取消时回传 cancelled。 + + 此协程通过 asyncio.create_task 在主事件循环中调度，与 agent 的 MCP + session / opentelemetry 上下文保持同一事件循环，避免跨线程/跨 loop 导致的 + "Root node was cancelled" / "Failed to detach context" 崩溃。 执行过程中每步都检查停止标志（gateway_client.is_stop_requested），一旦收到 取消指令（task_stop）即中断并回传失败（cancelled），网关 on_result 终态保护 @@ -97,7 +130,7 @@ def create_task_router(app): gateway_client.clear_stop_requested(request_id) try: prompt = _payload_to_prompt(payload) - reply, summary = asyncio.run(_run_agent_once(prompt, request_id)) + reply, summary = await _run_agent_once(prompt, request_id) if gateway_client.is_stop_requested(request_id): raise _TaskCancelled() gateway_client.report_result( @@ -107,6 +140,15 @@ def create_task_router(app): progress=100, result={"reply": reply, "output": summary}, ) + except asyncio.CancelledError: + logger.info("agent task cancelled (loop) request=%s", request_id) + gateway_client.report_result( + request_id, + agent_id=app.name, + status="failed", + progress=100, + error_info="cancelled by user", + ) except _TaskCancelled: logger.info("agent task cancelled request=%s", request_id) gateway_client.report_result( @@ -132,6 +174,9 @@ def create_task_router(app): body 中若携带 cli_session_id，则同时启动该会话的 SSE 停止指令订阅线程， 用于接收网关取消任务时下发的 task_stop。 + + 执行在 asyncio.create_task 中调度（与 MCP session 同事件循环）， + 不再使用新线程 + asyncio.run，避免跨事件循环导致 agent 崩溃。 """ body = await request.json() if body.get("auth") != gateway_client.GATEWAY_AUTH: @@ -140,12 +185,10 @@ def create_task_router(app): cli_session_id = body.get("cli_session_id") if cli_session_id: gateway_client.start_stop_listener(cli_session_id) - threading.Thread( - target=_execute_and_report, - args=(request_id, payload), + asyncio.create_task( + _execute_and_report(request_id, payload), name=f"task-{request_id[:8]}", - daemon=True, - ).start() + ) logger.info("task received request=%s payload=%s", request_id, payload) return JSONResponse(status_code=202, content={"ok": True, "request_id": request_id, "status": "accepted"})