From 7358b4df14c67ed7d0611886dce806d713655dac Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sun, 21 Jun 2026 21:43:44 +0800 Subject: [PATCH] Record streaming summaries for agent tool calls --- app/agent/__init__.py | 2 + app/agent/middleware/activity_log.py | 37 ++++++ app/agent/middleware/skills.py | 43 ++++++- app/agent/middleware/subagents.py | 166 ++++++++++++-------------- tests/test_agent_activity_log.py | 46 +++++++ tests/test_agent_skills_middleware.py | 43 +++++++ tests/test_agent_subagents.py | 100 +++++++++++++--- 7 files changed, 333 insertions(+), 104 deletions(-) diff --git a/app/agent/__init__.py b/app/agent/__init__.py index 2c6f2aff..4f84d98b 100644 --- a/app/agent/__init__.py +++ b/app/agent/__init__.py @@ -1023,6 +1023,7 @@ class MoviePilotAgent: skills_middleware = SkillsMiddleware( sources=[str(agent_runtime_manager.skills_dir)], bundled_skills_dir=str(settings.ROOT_PATH / "skills"), + stream_handler=self.stream_handler, ) skill_tools = list(getattr(skills_middleware, "tools", []) or []) activity_log_middleware = None @@ -1030,6 +1031,7 @@ class MoviePilotAgent: if self.has_message_context: activity_log_middleware = ActivityLogMiddleware( activity_dir=str(agent_runtime_manager.activity_dir), + stream_handler=self.stream_handler, ) activity_log_tools = list( getattr(activity_log_middleware, "tools", []) or [] diff --git a/app/agent/middleware/activity_log.py b/app/agent/middleware/activity_log.py index 01698b7e..825b9dc4 100644 --- a/app/agent/middleware/activity_log.py +++ b/app/agent/middleware/activity_log.py @@ -22,6 +22,7 @@ from langchain.agents.middleware.types import ( ModelResponse, PrivateStateAttr, # noqa ResponseT, + ToolCallRequest, ) from langchain_core.messages import AIMessage, HumanMessage, ToolMessage from langchain_core.tools import StructuredTool @@ -495,10 +496,13 @@ class ActivityLogMiddleware(AgentMiddleware[ActivityLogState, ContextT, Response activity_dir: str, retention_days: int = DEFAULT_RETENTION_DAYS, prompt_load_days: int = PROMPT_LOAD_DAYS, + stream_handler: Optional[Any] = None, ) -> None: + """初始化活动日志中间件。""" self.activity_dir = activity_dir self.retention_days = retention_days self.prompt_load_days = prompt_load_days + self.stream_handler = stream_handler self._tool_provider = _ActivityLogToolProvider(activity_dir=activity_dir) self.tools = [ StructuredTool.from_function( @@ -646,6 +650,39 @@ class ActivityLogMiddleware(AgentMiddleware[ActivityLogState, ContextT, Response modified_request = self.modify_request(request) return await handler(modified_request) + async def awrap_tool_call( + self, + request: ToolCallRequest, + handler: Callable[[ToolCallRequest], Awaitable[Any]], + ) -> Any: + """在活动日志查询工具执行时记录聚合摘要。""" + tool = request.tool + tool_name = getattr(tool, "name", None) + if tool_name != QUERY_ACTIVITY_LOG_TOOL_NAME: + return await handler(request) + + tool_call = request.tool_call or {} + tool_args = tool_call.get("args") or {} + if not isinstance(tool_args, dict): + tool_args = {} + logger.info( + f"开始执行活动日志查询工具: keyword={tool_args.get('keyword') or '-'}, " + f"date={tool_args.get('date') or '-'}" + ) + if self.stream_handler and getattr(self.stream_handler, "is_streaming", False): + self.stream_handler.record_tool_call( + tool_name=QUERY_ACTIVITY_LOG_TOOL_NAME, + tool_message=QUERY_ACTIVITY_LOG_TOOL_DESCRIPTION, + tool_kwargs=tool_args, + ) + try: + result = await handler(request) + except Exception as err: + logger.error(f"活动日志查询工具执行失败: error={err}") + raise + logger.info("活动日志查询工具执行完成") + return result + async def aafter_agent( self, state: ActivityLogState, runtime: Runtime ) -> Optional[dict[str, Any]]: diff --git a/app/agent/middleware/skills.py b/app/agent/middleware/skills.py index 920de9fc..1c64eea9 100644 --- a/app/agent/middleware/skills.py +++ b/app/agent/middleware/skills.py @@ -3,7 +3,7 @@ import re import shutil from collections.abc import Awaitable, Callable from pathlib import Path -from typing import Annotated, List, Optional +from typing import Annotated, Any, List, Optional from typing import NotRequired, TypedDict import yaml # noqa @@ -15,6 +15,7 @@ from langchain.agents.middleware.types import ( ModelRequest, ModelResponse, ResponseT, + ToolCallRequest, ) from langchain.agents.middleware.types import PrivateStateAttr # noqa from langchain_core.runnables import RunnableConfig @@ -525,6 +526,7 @@ class SkillsMiddleware(AgentMiddleware[SkillsState, ContextT, ResponseT]): # no *, sources: list[str], bundled_skills_dir: str | None = None, + stream_handler: Optional[Any] = None, ) -> None: """初始化 Skill 中间件。 @@ -535,9 +537,12 @@ class SkillsMiddleware(AgentMiddleware[SkillsState, ContextT, ResponseT]): # no bundled_skills_dir : str | None 项目内置技能目录路径。若提供，在首次加载前会将其中不存在于 sources 首个目录的技能自动复制过去。 + stream_handler : Optional[Any] + 流式输出处理器，用于记录 skill 工具调用摘要。 """ self.sources = sources self.bundled_skills_dir = bundled_skills_dir + self.stream_handler = stream_handler self.system_prompt_template = SKILLS_SYSTEM_PROMPT self._skill_provider = _SkillToolProvider(sources=sources) self.tools = [ @@ -584,7 +589,8 @@ class SkillsMiddleware(AgentMiddleware[SkillsState, ContextT, ResponseT]): # no skills_catalog=_format_skill_tool_catalog(skills) ) - def _format_skills_list(self, skills: list[SkillMetadata]) -> str: + @staticmethod + def _format_skills_list(skills: list[SkillMetadata]) -> str: """格式化技能元数据列表用于系统提示词。""" if not skills: return "(No skills available yet.)" @@ -657,5 +663,38 @@ class SkillsMiddleware(AgentMiddleware[SkillsState, ContextT, ResponseT]): # no modified_request = self.modify_request(request) return await handler(modified_request) + async def awrap_tool_call( + self, + request: ToolCallRequest, + handler: Callable[[ToolCallRequest], Awaitable[Any]], + ) -> Any: + """在 skill 工具执行时记录聚合摘要。""" + tool = request.tool + tool_name = getattr(tool, "name", None) + if tool_name != SKILL_TOOL_NAME: + return await handler(request) + + tool_call = request.tool_call or {} + tool_args = tool_call.get("args") or {} + if not isinstance(tool_args, dict): + tool_args = {} + logger.info( + f"开始执行 Skill 工具: name={tool_args.get('name') or '-'}, " + f"explanation={tool_args.get('explanation') or '-'}" + ) + if self.stream_handler and getattr(self.stream_handler, "is_streaming", False): + self.stream_handler.record_tool_call( + tool_name=SKILL_TOOL_NAME, + tool_message="Skill loaded", + tool_kwargs=tool_args, + ) + try: + result = await handler(request) + except Exception as err: + logger.error(f"Skill 工具执行失败: error={err}") + raise + logger.info("Skill 工具执行完成") + return result + __all__ = ["SKILL_TOOL_NAME", "SkillMetadata", "SkillsMiddleware"] diff --git a/app/agent/middleware/subagents.py b/app/agent/middleware/subagents.py index 8aa11969..0b2e5159 100644 --- a/app/agent/middleware/subagents.py +++ b/app/agent/middleware/subagents.py @@ -313,6 +313,31 @@ def _format_datetime(value: Optional[datetime]) -> Optional[str]: return value.strftime("%Y-%m-%d %H:%M:%S") +def _extract_tool_call_args(request: ToolCallRequest) -> dict[str, Any]: + """提取工具调用参数，并规整为字典。""" + tool_call = request.tool_call or {} + tool_args = tool_call.get("args") or {} + if not isinstance(tool_args, dict): + return {} + return tool_args + + +def _record_subagent_tool_call( + *, + stream_handler: Any, + tool_name: str, + tool_args: dict[str, Any], +) -> None: + """在流式处理器中记录子代理工具调用摘要。""" + if not stream_handler or not getattr(stream_handler, "is_streaming", False): + return + stream_handler.record_tool_call( + tool_name=tool_name, + tool_message="Subagent invoked", + tool_kwargs=tool_args, + ) + + class _SubAgentAgentProvider: """子代理图懒加载与执行器。""" @@ -409,8 +434,10 @@ class MoviePilotSubAgentMiddleware(AgentMiddleware): tools: list[BaseTool], system_prompt: str = SUBAGENT_PARENT_PROMPT, task_description: str = SUBAGENT_TASK_DESCRIPTION, + stream_handler: Any = None, ) -> None: self.system_prompt = system_prompt + self.stream_handler = stream_handler self._provider = _SubAgentAgentProvider( model=model, profiles=profiles, @@ -453,6 +480,35 @@ class MoviePilotSubAgentMiddleware(AgentMiddleware): ) return await handler(request.override(system_message=new_system_message)) + async def awrap_tool_call( + self, + request: ToolCallRequest, + handler: Callable[[ToolCallRequest], Awaitable[Any]], + ) -> Any: + """在 task 子代理工具执行时记录聚合摘要。""" + tool = request.tool + tool_name = getattr(tool, "name", None) + if tool_name != SUBAGENT_TASK_TOOL_NAME: + return await handler(request) + + tool_args = _extract_tool_call_args(request) + logger.info( + f"开始执行子代理工具: tool_name={tool_name}, " + f"subagent_type={tool_args.get('subagent_type') or '-'}" + ) + _record_subagent_tool_call( + stream_handler=self.stream_handler, + tool_name=SUBAGENT_TASK_TOOL_NAME, + tool_args=tool_args, + ) + try: + result = await handler(request) + except Exception as err: + logger.error(f"子代理工具执行失败: tool_name={tool_name}, error={err}") + raise + logger.info(f"子代理工具执行完成: tool_name={tool_name}") + return result + class SubAgentTaskControlMiddleware(AgentMiddleware): """提供异步子代理任务调度工具的中间件。""" @@ -464,8 +520,10 @@ class SubAgentTaskControlMiddleware(AgentMiddleware): profiles: tuple[_SubAgentProfile, ...], tools: list[BaseTool], task_description: str = SUBAGENT_CONTROL_DESCRIPTION, + stream_handler: Any = None, ) -> None: """初始化异步子代理调度中间件。""" + self.stream_handler = stream_handler self._provider = _SubAgentAgentProvider( model=model, profiles=profiles, @@ -987,98 +1045,37 @@ class SubAgentTaskControlMiddleware(AgentMiddleware): logger.info(f"Agent 结束，取消未完成子代理任务: tasks={len(unfinished_records)}") await self._cancel_records(unfinished_records) - -class SubAgentCallSummaryMiddleware(AgentMiddleware): - """记录子代理调用次数的中间件。""" - - def __init__(self, *, stream_handler: Any = None) -> None: - self.stream_handler = stream_handler - self.tools = [] - async def awrap_tool_call( self, request: ToolCallRequest, handler: Callable[[ToolCallRequest], Awaitable[Any]], ) -> Any: - """在子代理任务工具执行时记录聚合摘要。""" + """在 subagent_task 子代理工具执行时记录聚合摘要。""" tool = request.tool tool_name = getattr(tool, "name", None) - is_subagent_tool = tool_name in { - SUBAGENT_TASK_TOOL_NAME, - SUBAGENT_CONTROL_TOOL_NAME, - } - if is_subagent_tool: - tool_call = request.tool_call or {} - tool_args = tool_call.get("args") or {} - if not isinstance(tool_args, dict): - tool_args = {} - logger.info( - f"开始执行子代理工具: tool_name={tool_name}, " - f"action={tool_args.get('action') or '-'}, " - f"subagent_type={tool_args.get('subagent_type') or '-'}" - ) - if ( - self.stream_handler - and getattr(self.stream_handler, "is_streaming", False) - ): - self.stream_handler.record_tool_call( - tool_name=tool_name or SUBAGENT_TASK_TOOL_NAME, - tool_message="Subagent invoked", - tool_kwargs=tool_args, - ) + if tool_name != SUBAGENT_CONTROL_TOOL_NAME: + return await handler(request) + + tool_args = _extract_tool_call_args(request) + logger.info( + f"开始执行子代理工具: tool_name={tool_name}, " + f"action={tool_args.get('action') or '-'}, " + f"subagent_type={tool_args.get('subagent_type') or '-'}" + ) + _record_subagent_tool_call( + stream_handler=self.stream_handler, + tool_name=SUBAGENT_CONTROL_TOOL_NAME, + tool_args=tool_args, + ) try: result = await handler(request) except Exception as err: - if is_subagent_tool: - logger.error(f"子代理工具执行失败: tool_name={tool_name}, error={err}") + logger.error(f"子代理工具执行失败: tool_name={tool_name}, error={err}") raise - if is_subagent_tool: - logger.info(f"子代理工具执行完成: tool_name={tool_name}") + logger.info(f"子代理工具执行完成: tool_name={tool_name}") return result -def _deepagents_spec( - profiles: tuple[_SubAgentProfile, ...], tools: list[BaseTool] -) -> list[dict[str, Any]]: - """将内置定义转换为 Deep Agents 子代理配置。""" - specs = [] - for profile in profiles: - specs.append( - { - "name": profile.name, - "description": profile.description, - "prompt": profile.prompt, - "tools": _select_tools(tools, profile), - } - ) - return specs - - -def _try_create_deepagents_middleware( - *, - profiles: tuple[_SubAgentProfile, ...], - tools: list[BaseTool], - model: BaseChatModel, -) -> Optional[AgentMiddleware]: - """优先创建 Deep Agents 官方子代理中间件。""" - try: - from deepagents.backends import StateBackend - from deepagents.middleware.subagents import SubAgentMiddleware - - return SubAgentMiddleware( - backend=StateBackend(), - subagents=_deepagents_spec(profiles, tools), - default_model=model, - system_prompt=SUBAGENT_PARENT_PROMPT, - task_description=SUBAGENT_TASK_DESCRIPTION, - ) - except ImportError: - return None - except Exception as err: - logger.debug(f"Deep Agents 子代理中间件不可用，使用本地实现: {err}") - return None - - def create_subagent_middlewares( *, model: BaseChatModel, @@ -1089,21 +1086,17 @@ def create_subagent_middlewares( _builtin_subagent_profiles.cache_clear() builtin_subagent_names.cache_clear() profiles = _builtin_subagent_profiles() - subagent_middleware = _try_create_deepagents_middleware( + subagent_middleware = MoviePilotSubAgentMiddleware( + model=model, profiles=profiles, tools=tools, - model=model, + stream_handler=stream_handler, ) - if subagent_middleware is None: - subagent_middleware = MoviePilotSubAgentMiddleware( - model=model, - profiles=profiles, - tools=tools, - ) control_middleware = SubAgentTaskControlMiddleware( model=model, profiles=profiles, tools=tools, + stream_handler=stream_handler, ) task_tools = [ @@ -1113,7 +1106,6 @@ def create_subagent_middlewares( return [ subagent_middleware, control_middleware, - SubAgentCallSummaryMiddleware(stream_handler=stream_handler), ], task_tools diff --git a/tests/test_agent_activity_log.py b/tests/test_agent_activity_log.py index d44dd224..5296daab 100644 --- a/tests/test_agent_activity_log.py +++ b/tests/test_agent_activity_log.py @@ -8,6 +8,7 @@ from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, Tool from app.agent.middleware.activity_log import ( ActivityLogMiddleware, + QUERY_ACTIVITY_LOG_TOOL_DESCRIPTION, QUERY_ACTIVITY_LOG_TOOL_NAME, _summarize_with_llm, load_activity_log_index, @@ -260,6 +261,51 @@ def test_activity_log_middleware_query_tool_returns_json_payload(tmp_path): assert payload["entries"][0]["summary"] == "帮用户整理了电影 A" +def test_activity_log_tool_call_records_streaming_summary(tmp_path): + """query_activity_log 工具执行时应记录流式聚合摘要。""" + + async def _run_test(): + calls = [] + stream_handler = SimpleNamespace( + is_streaming=True, + record_tool_call=lambda **kwargs: calls.append(kwargs), + ) + middleware = ActivityLogMiddleware( + activity_dir=str(tmp_path), + stream_handler=stream_handler, + ) + request = SimpleNamespace( + tool=SimpleNamespace(name=QUERY_ACTIVITY_LOG_TOOL_NAME), + tool_call={ + "args": { + "keyword": "整理", + "date": "2026-06-18", + } + }, + ) + + async def _fake_handler(_request): + """返回模拟工具结果。""" + return "ok" + + result = await middleware.awrap_tool_call(request, _fake_handler) + return result, calls + + result, calls = asyncio.run(_run_test()) + + assert result == "ok" + assert calls == [ + { + "tool_name": QUERY_ACTIVITY_LOG_TOOL_NAME, + "tool_message": QUERY_ACTIVITY_LOG_TOOL_DESCRIPTION, + "tool_kwargs": { + "keyword": "整理", + "date": "2026-06-18", + }, + } + ] + + def test_factory_does_not_register_activity_log_tool(): """活动日志查询工具应由中间件注册，不应进入全局工具工厂。""" with patch( diff --git a/tests/test_agent_skills_middleware.py b/tests/test_agent_skills_middleware.py index 9146c5ef..168e0da4 100644 --- a/tests/test_agent_skills_middleware.py +++ b/tests/test_agent_skills_middleware.py @@ -1,4 +1,5 @@ import json +from types import SimpleNamespace import pytest from anyio import Path as AsyncPath @@ -111,3 +112,45 @@ def test_modify_request_instructs_model_to_use_skill_tool_without_paths(tmp_path assert "moviepilot-cli" in system_content assert "Read `" not in system_content assert str(tmp_path) not in system_content + + +@pytest.mark.anyio +async def test_skill_tool_call_records_streaming_summary(tmp_path): + """skill 工具执行时应记录流式聚合摘要。""" + _write_skill(tmp_path, "moviepilot-cli") + calls = [] + stream_handler = SimpleNamespace( + is_streaming=True, + record_tool_call=lambda **kwargs: calls.append(kwargs), + ) + middleware = SkillsMiddleware( + sources=[str(tmp_path)], + stream_handler=stream_handler, + ) + request = SimpleNamespace( + tool=SimpleNamespace(name=SKILL_TOOL_NAME), + tool_call={ + "args": { + "name": "moviepilot-cli", + "explanation": "测试加载技能", + } + }, + ) + + async def _fake_handler(_request): + """返回模拟工具结果。""" + return "ok" + + result = await middleware.awrap_tool_call(request, _fake_handler) + + assert result == "ok" + assert calls == [ + { + "tool_name": SKILL_TOOL_NAME, + "tool_message": "Skill loaded", + "tool_kwargs": { + "name": "moviepilot-cli", + "explanation": "测试加载技能", + }, + } + ] diff --git a/tests/test_agent_subagents.py b/tests/test_agent_subagents.py index 50df5f23..f8b66e27 100644 --- a/tests/test_agent_subagents.py +++ b/tests/test_agent_subagents.py @@ -11,7 +11,6 @@ from app.agent.middleware.subagents import ( MoviePilotSubAgentMiddleware, SUBAGENT_CONTROL_TOOL_NAME, SUBAGENT_TASK_TOOL_NAME, - SubAgentCallSummaryMiddleware, SubAgentTaskControlMiddleware, create_subagent_middlewares, ) @@ -28,7 +27,9 @@ def test_create_subagent_middlewares_registers_task_tool(): stream_handler=None, ) - assert len(middlewares) == 3 + assert len(middlewares) == 2 + assert isinstance(middlewares[0], MoviePilotSubAgentMiddleware) + assert isinstance(middlewares[1], SubAgentTaskControlMiddleware) assert [tool.name for tool in task_tools] == [ SUBAGENT_TASK_TOOL_NAME, SUBAGENT_CONTROL_TOOL_NAME, @@ -140,17 +141,27 @@ def test_builtin_tools_declare_tags_in_implementation(): assert missing_tools == [] -def test_call_summary_middleware_logs_subagent_tool_operations(): - """子代理工具包装层应记录工具执行开始和完成日志。""" +def test_task_tool_call_records_streaming_summary(): + """task 子代理工具执行时应记录流式聚合摘要。""" async def _run_test(): - middleware = SubAgentCallSummaryMiddleware() + calls = [] + stream_handler = SimpleNamespace( + is_streaming=True, + record_tool_call=lambda **kwargs: calls.append(kwargs), + ) + middleware = MoviePilotSubAgentMiddleware( + model=FakeListChatModel(responses=["ok"]), + profiles=subagent_module._builtin_subagent_profiles(), + tools=[], + stream_handler=stream_handler, + ) request = SimpleNamespace( - tool=SimpleNamespace(name=SUBAGENT_CONTROL_TOOL_NAME), + tool=SimpleNamespace(name=SUBAGENT_TASK_TOOL_NAME), tool_call={ "args": { - "action": "status", - "subagent_type": "general-purpose", + "description": "检查媒体信息", + "subagent_type": "media-researcher", } }, ) @@ -158,15 +169,74 @@ def test_call_summary_middleware_logs_subagent_tool_operations(): async def _fake_handler(_request): return "ok" - with patch.object(subagent_module.logger, "info") as log_info: - result = await middleware.awrap_tool_call(request, _fake_handler) + result = await middleware.awrap_tool_call(request, _fake_handler) + return result, calls - messages = [call.args[0] for call in log_info.call_args_list] - assert result == "ok" - assert any("开始执行子代理工具" in message for message in messages) - assert any("子代理工具执行完成" in message for message in messages) + result, calls = asyncio.run(_run_test()) - asyncio.run(_run_test()) + assert result == "ok" + assert calls == [ + { + "tool_name": SUBAGENT_TASK_TOOL_NAME, + "tool_message": "Subagent invoked", + "tool_kwargs": { + "description": "检查媒体信息", + "subagent_type": "media-researcher", + }, + } + ] + + +def test_control_tool_call_records_streaming_summary(): + """subagent_task 子代理工具执行时应记录流式聚合摘要。""" + + async def _run_test(): + calls = [] + stream_handler = SimpleNamespace( + is_streaming=True, + record_tool_call=lambda **kwargs: calls.append(kwargs), + ) + middleware = SubAgentTaskControlMiddleware( + model=FakeListChatModel(responses=["ok"]), + profiles=subagent_module._builtin_subagent_profiles(), + tools=[], + stream_handler=stream_handler, + ) + request = SimpleNamespace( + tool=SimpleNamespace(name=SUBAGENT_CONTROL_TOOL_NAME), + tool_call={ + "args": { + "action": "start", + "tasks": [ + {"subagent_type": "media-researcher"}, + {"subagent_type": "download-diagnostician"}, + ], + } + }, + ) + + async def _fake_handler(_request): + return "ok" + + result = await middleware.awrap_tool_call(request, _fake_handler) + return result, calls + + result, calls = asyncio.run(_run_test()) + + assert result == "ok" + assert calls == [ + { + "tool_name": SUBAGENT_CONTROL_TOOL_NAME, + "tool_message": "Subagent invoked", + "tool_kwargs": { + "action": "start", + "tasks": [ + {"subagent_type": "media-researcher"}, + {"subagent_type": "download-diagnostician"}, + ], + }, + } + ] def test_control_tool_starts_tasks_concurrently_and_waits():