From e1b703aacc4589fec01a1d9753b54d96cf21e94d Mon Sep 17 00:00:00 2001 From: DDSRem <73049927+DDSRem@users.noreply.github.com> Date: Sun, 18 Oct 2026 21:49:57 +0800 Subject: [PATCH] refactor(agent): centralize graph resource ownership (#7062) --- app/agent/graph.py | 266 ++++++++ app/agent/orchestrator.py | 637 +++++++----------- app/api/endpoints/openai.py | 7 +- docs/architecture-overview.md | 4 +- docs/rules/05-architecture.md | 3 +- scripts/evaluation/runtime.py | 2 +- tests/agent_graph.py | 37 + .../architecture/dependency-baseline.json | 15 +- .../fixtures/architecture/mypy-baseline.json | 3 +- tests/test_agent_api_lazy_imports.py | 30 +- tests/test_agent_graph_cache.py | 17 +- tests/test_agent_graph_resources.py | 335 +++++++++ tests/test_agent_recovery.py | 17 +- tests/test_agent_side_effect_boundaries.py | 35 +- tests/test_agent_summarization_streaming.py | 5 +- tests/test_architecture_dependencies.py | 23 + tests/test_llm_client_lifecycle.py | 31 +- 17 files changed, 978 insertions(+), 489 deletions(-) create mode 100644 app/agent/graph.py create mode 100644 tests/agent_graph.py create mode 100644 tests/test_agent_graph_resources.py diff --git a/app/agent/graph.py b/app/agent/graph.py new file mode 100644 index 000000000..9da51224b --- /dev/null +++ b/app/agent/graph.py @@ -0,0 +1,266 @@ +""" +会话执行图的缓存及资源所有权。 + +构建中的模型、中间件与已发布图使用同一资源范围；发布同步移交所有权， +取消、构建失败和缓存替换只回收退休范围，未退出的消费者继续保留对应模型。 +""" + +import asyncio +import inspect +from collections.abc import AsyncIterator, Callable, Iterable +from contextlib import asynccontextmanager +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any, Optional + +from app.agent.llm.helper import LLMHelper +from app.agent.tools.catalog import ToolCatalogSnapshot +from app.runtime.log import logger + + +class GraphResources: + """ + 一次构图或临时模型调用拥有的资源，不负责创建模型或业务中间件。 + """ + + def __init__(self) -> None: + """ + 按对象身份记录资源，避免模型相等比较及重复释放。 + """ + self.models: dict[int, Any] = {} + self.middlewares: dict[int, object] = {} + self._sealed = False + + def own_model(self, model: Any) -> None: + """ + 在创建模型后的下一个异步等待前接管其客户端。 + + :param model: 当前范围使用的模型；重复登记同一对象不会增加关闭次数 + """ + self.models[id(model)] = model + + def own_middlewares(self, middlewares: Iterable[object]) -> None: + """ + 接管提供关闭能力的中间件；关停期间接管时立即封住新任务入口。 + + :param middlewares: 图使用的中间件，未声明 close 的无资源中间件不登记 + """ + for middleware in middlewares: + if callable(getattr(middleware, "close", None)): + self.middlewares[id(middleware)] = middleware + if self._sealed: + self.seal() + + def seal(self) -> None: + """ + 同步禁止已接管控制器提交新任务，仍由异步回收确认真实退出。 + """ + self._sealed = True + for middleware in self.middlewares.values(): + seal = getattr(middleware, "seal", None) + if callable(seal): + try: + seal() + except Exception as error: + logger.debug(f"封住子代理中间件失败: {error}") + + @property + def empty(self) -> bool: + """ + 所有消费者及模型都已经释放时返回 True。 + """ + return not self.models and not self.middlewares + + +@dataclass +class CompiledAgentBundle: + """ + 会话内可复用的执行图、构造签名及其唯一资源范围。 + + :param signature: 模型、工具和运行配置的构造签名 + :param agent: 已编译的执行图 + :param streaming: 是否按流式模式构造 + :param created_at: 图构造完成时间 + :param tool_catalog: 主智能体工具目录；None 表示尚未提供 + :param subagent_catalog: 子智能体工具目录；None 表示尚未提供 + :param plugin_revision: 插件目录修订号；-1 表示尚未记录 + :param mcp_config_signature: 外部 MCP 配置签名；空串表示尚未记录 + :param catalog_checked_at: 最近目录复核时间；None 表示尚未复核 + :param resources: 图的模型和中间件资源；默认建立空范围 + """ + + signature: tuple[Any, ...] + agent: Any + streaming: bool + created_at: datetime + tool_catalog: Optional[ToolCatalogSnapshot] = None + subagent_catalog: Optional[ToolCatalogSnapshot] = None + plugin_revision: int = -1 + mcp_config_signature: str = "" + catalog_checked_at: Optional[datetime] = None + resources: GraphResources = field(default_factory=GraphResources) + + +class AgentGraphCache: + """ + 单会话图资源 owner，统一跟踪构建、发布和退休范围。 + + 在会话所属事件循环内使用；资源登记与发布不等待，回收串行执行。 + 关闭失败或取消时保留尚未确认释放的对象，后续失效或清理可继续回收。 + """ + + def __init__(self, *, models_in_use: Callable[[], bool]) -> None: + """ + 创建空图缓存，不启动任务或取得外部资源。 + + :param models_in_use: 后台复盘仍使用共享模型时返回 True，暂缓退休模型释放 + """ + self._models_in_use = models_in_use + self._current: Optional[CompiledAgentBundle] = None + self._building: dict[int, GraphResources] = {} + self._retired: dict[int, GraphResources] = {} + self._collection_lock = asyncio.Lock() + self._sealed = False + + @property + def current(self) -> Optional[CompiledAgentBundle]: + """ + 返回当前已发布图；失效后立即为 None，即使资源仍待回收。 + """ + return self._current + + def get(self, signature: tuple[Any, ...], streaming: bool) -> Optional[Any]: + """ + 只复用构造签名和流式模式一致的当前图。 + """ + bundle = self._current + if bundle and bundle.signature == signature and bundle.streaming == streaming: + return bundle.agent + return None + + @asynccontextmanager + async def build(self) -> AsyncIterator[GraphResources]: + """ + 登记构建范围；未发布的范围在任意退出路径退休并尝试回收。 + + :yields: 模型和中间件创建后必须立即登记的资源范围 + :raises RuntimeError: 会话已经开始关闭，禁止创建新范围 + """ + if self._sealed: + raise RuntimeError("Agent 图资源正在关闭") + resources = GraphResources() + self._building[id(resources)] = resources + try: + yield resources + finally: + if self._building.pop(id(resources), None) is not None: + self._retired[id(resources)] = resources + await self.collect() + + def publish(self, bundle: CompiledAgentBundle) -> None: + """ + 同步发布完整图并移交资源，旧图在后续回收中退休。 + + :param bundle: 构造成功的图；资源应来自当前构建范围 + :raises RuntimeError: 会话已封闭，调用方范围负责回收未发布资源 + """ + if self._sealed: + raise RuntimeError("Agent 图资源正在关闭") + if self._current is not None and self._current.resources is not bundle.resources: + previous = self._current.resources + self._retired[id(previous)] = previous + self._current = bundle + self._building.pop(id(bundle.resources), None) + + def seal(self) -> None: + """ + 封住当前、构建中及退休控制器；随后完成的构建不能发布。 + """ + self._sealed = True + for resources in (*self._active(), *self._retired.values()): + resources.seal() + + def _active(self) -> tuple[GraphResources, ...]: + """ + 返回仍由前台构建或已发布图使用的资源范围。 + """ + current = (self._current.resources,) if self._current is not None else () + return (*current, *self._building.values()) + + def retire(self) -> None: + """ + 同步使当前图失效，资源保留到下次构建退出或异步清理时回收。 + """ + if self._current is not None: + resources = self._current.resources + self._retired[id(resources)] = resources + self._current = None + + async def invalidate(self) -> bool: + """ + 立即使图失效并回收资源；构建、消费者或关闭操作未收敛时返回 False。 + """ + self.retire() + await self.collect() + return self._current is None and not self._building and not self._retired + + @staticmethod + async def _close_middleware(middleware: object) -> bool: + """ + 适配中间件已有同步／异步关闭合同；False 或异常表示仍需保留资源。 + """ + try: + result = getattr(middleware, "close")() + if inspect.isawaitable(result): + result = await result + return result is not False + except Exception as error: + logger.debug(f"关闭子代理中间件失败: {error}") + return False + + async def _close_middlewares(self, retired: tuple[GraphResources, ...]) -> None: + """ + 先关闭退休消费者；同一控制器每轮最多尝试一次，取消不丢失未关闭对象。 + """ + attempted: set[int] = set() + for resources in retired: + for identity, middleware in tuple(resources.middlewares.items()): + if identity in attempted: + continue + attempted.add(identity) + active = {key for group in self._active() for key in group.middlewares} + if identity in active or await self._close_middleware(middleware): + for group in self._retired.values(): + group.middlewares.pop(identity, None) + + async def _close_models(self, retired: tuple[GraphResources, ...]) -> None: + """ + 消费者退出后释放模型；仍被复盘或其他资源范围使用的模型继续保留。 + """ + attempted: set[int] = set() + for resources in retired: + for identity, model in tuple(resources.models.items()): + if identity in attempted: + continue + attempted.add(identity) + active = {key for group in self._active() for key in group.models} + pending = {key for group in self._retired.values() if group.middlewares for key in group.models} + if identity not in active: + if identity in pending or self._models_in_use(): + continue + if not await LLMHelper.close_llm(model): + continue + for group in self._retired.values(): + group.models.pop(identity, None) + + async def collect(self) -> None: + """ + 串行回收退休范围；成功的单个释放立即登记，取消后可从剩余资源继续。 + """ + async with self._collection_lock: + retired = tuple(self._retired.values()) + await self._close_middlewares(retired) + await self._close_models(retired) + for resources in retired: + if resources.empty: + self._retired.pop(id(resources), None) diff --git a/app/agent/orchestrator.py b/app/agent/orchestrator.py index 255511286..67cf7a9a0 100644 --- a/app/agent/orchestrator.py +++ b/app/agent/orchestrator.py @@ -4,7 +4,6 @@ import asyncio import hashlib -import inspect import json import re import time @@ -28,6 +27,7 @@ from langgraph.checkpoint.memory import InMemorySaver from app.agent.callback import StreamingHandler from app.agent.contracts import ReplyMode, build_display_message +from app.agent.graph import AgentGraphCache, CompiledAgentBundle from app.agent.learning.session import LearningSession from app.agent.llm.helper import LLMHelper from app.agent.llm.tools import ServerToolRegistry @@ -285,35 +285,6 @@ class _SessionUsageSnapshot: } -@dataclass -class _CompiledAgentBundle: - """ - 会话内可复用的 Agent 图及其构造签名。 - - :param signature: 构造模型、工具和运行图时使用的配置签名，供复用判断 - :param agent: 已编译且可复用的 Agent 执行图 - :param streaming: 该执行图是否按流式模式构造 - :param created_at: 本次执行图构造完成的时间 - :param tool_catalog: 主智能体使用的工具目录快照；尚未提供时为 None - :param subagent_catalog: 子智能体使用的工具目录快照；尚未提供时为 None - :param plugin_revision: 构造时插件工具目录修订号；-1 表示尚未记录 - :param mcp_config_signature: 构造时外部 MCP 配置签名；尚未记录时为空字符串 - :param catalog_checked_at: 最近检查工具目录是否变化的时间；未检查时为 None - :param models: 本执行图持有的模型实例，用于统一释放客户端资源 - """ - - signature: tuple[Any, ...] - agent: Any - streaming: bool - created_at: datetime - tool_catalog: Optional[ToolCatalogSnapshot] = None - subagent_catalog: Optional[ToolCatalogSnapshot] = None - plugin_revision: int = -1 - mcp_config_signature: str = "" - catalog_checked_at: Optional[datetime] = None - models: tuple[Any, ...] = () - - class _ThinkTagStripper: """ 流式剥离 ... 标签的辅助类。 @@ -562,9 +533,7 @@ class MoviePilotAgent: self._llm_runtime_config: Optional[Dict[str, Any]] = None self._llm_provider_selection: Dict[str, Any] = {} self._agent_started_at: Optional[datetime] = None - self._compiled_agent_bundle: Optional[_CompiledAgentBundle] = None - self._retired_llm_models: list[Any] = [] - self._subagent_middlewares: tuple[Any, ...] = () + self._graphs = AgentGraphCache(models_in_use=self._learning_models_in_use) self._shutdown_started = False self._last_agent_cache_hit = False @@ -712,16 +681,15 @@ class MoviePilotAgent: """ if not str(message or "").strip(): return "" - model = await self._initialize_llm(streaming=False) - try: + async with self._graphs.build() as resources: + model = await self._initialize_llm(streaming=False) + resources.own_model(model) response = await model.ainvoke( [ SystemMessage(content=AGENT_CHAT_TITLE_PROMPT), HumanMessage(content=self._build_chat_title_message(message)), ] ) - finally: - await self._release_llm_models((model,)) content = LLMHelper.extract_text_content(getattr(response, "content", response)) title = self._parse_chat_title_response(content) if not self._is_valid_chat_title(title): @@ -1877,71 +1845,9 @@ class MoviePilotAgent: return get_tool_factory().catalog_factory_revision() - def _get_cached_agent(self, signature: tuple[Any, ...], streaming: bool) -> Optional[Any]: - """ - 按签名读取当前会话已编译的 Agent 图。 - """ - bundle = self._compiled_agent_bundle - if bundle and bundle.streaming == streaming and bundle.signature == signature: - return bundle.agent - return None - - @staticmethod - def _seal_subagent_middleware_instances( - middlewares: tuple[Any, ...], - ) -> None: - """ - 同步封住子代理控制器的新任务入口，不等待既有任务退出。 - """ - for middleware in middlewares: - seal = getattr(middleware, "seal", None) - if not callable(seal): - continue - try: - seal() - except Exception as error: - logger.debug(f"封住子代理中间件失败: {error}") - - @staticmethod - async def _close_subagent_middleware_instances( - middlewares: tuple[Any, ...], - ) -> tuple[Any, ...]: - """ - 关闭子代理控制器，并返回仍持有未收敛任务的实例。 - """ - pending_middlewares = [] - for middleware in middlewares: - close = getattr(middleware, "close", None) - if not callable(close): - continue - try: - result = close() - if inspect.isawaitable(result): - result = await result - if result is False: - pending_middlewares.append(middleware) - except Exception as error: - logger.debug(f"关闭子代理中间件失败: {error}") - pending_middlewares.append(middleware) - return tuple(pending_middlewares) - - @staticmethod - def _merge_subagent_middleware_owners( - *groups: tuple[Any, ...], - ) -> tuple[Any, ...]: - """ - 按对象身份合并当前图与延迟收敛控制器的 owner 集合。 - """ - merged = [] - for group in groups: - for middleware in group: - if not any(middleware is existing for existing in merged): - merged.append(middleware) - return tuple(merged) - def begin_shutdown(self) -> None: """ - 在任何异步等待前封住当前 Agent 的子代理提交和终端作用域。 + 在任何异步等待前封住图资源、后台复盘和终端作用域。 """ if self._learning: self._learning.seal() @@ -1949,86 +1855,20 @@ class MoviePilotAgent: for scope in self._scheduled_terminal_scopes: scope.seal() self._shutdown_started = True - self._seal_subagent_middleware_instances(self._subagent_middlewares) + self._graphs.seal() - async def _cache_agent( - self, - *, - signature: tuple[Any, ...], - agent: Any, - streaming: bool, - tool_catalog: ToolCatalogSnapshot, - subagent_catalog: ToolCatalogSnapshot, - mcp_config_signature: str, - subagent_middlewares: tuple[Any, ...] = (), - models: tuple[Any, ...] = (), - ) -> Any: + def _learning_models_in_use(self) -> bool: """ - 保存当前会话可复用的 Agent 图。 + 后台复盘尚未退出时保留可能与其共享客户端的退休模型。 """ - previous_middlewares = tuple( - middleware - for middleware in self._subagent_middlewares - if not any(middleware is replacement for replacement in subagent_middlewares) - ) - if self._shutdown_started: - self._seal_subagent_middleware_instances(subagent_middlewares) - if self._compiled_agent_bundle: - self._retired_llm_models.extend(self._compiled_agent_bundle.models) - pending_middlewares = await self._close_subagent_middleware_instances(previous_middlewares) - self._compiled_agent_bundle = _CompiledAgentBundle( - signature=signature, - agent=agent, - streaming=streaming, - created_at=datetime.now(), - tool_catalog=tool_catalog, - subagent_catalog=subagent_catalog, - plugin_revision=tool_catalog.plugin_revision, - mcp_config_signature=mcp_config_signature, - catalog_checked_at=datetime.now(), - models=models, - ) - self._subagent_middlewares = self._merge_subagent_middleware_owners( - pending_middlewares, - subagent_middlewares, - ) - await self._release_retired_llm_models(consumers_closed=not pending_middlewares) - return agent - - async def _release_llm_models(self, models: tuple[Any, ...]) -> bool: - """ - 释放已无消费者的模型；关闭失败或清理被取消时保留模型供下一次清理。 - """ - for model in models: - if not any(model is existing for existing in self._retired_llm_models): - self._retired_llm_models.append(model) - for model in models: - if await LLMHelper.close_llm(model): - self._retired_llm_models = [item for item in self._retired_llm_models if item is not model] - return not self._retired_llm_models - - async def _release_retired_llm_models(self, *, consumers_closed: bool) -> bool: - """ - 等子代理及共享模型的后台复盘退出后，再释放旧图连接。 - """ - learning_run = self._learning.run if self._learning else None - learning_active = bool(learning_run and learning_run.task and not learning_run.task.done()) - if not consumers_closed or learning_active: - return not self._retired_llm_models - return await self._release_llm_models(tuple(self._retired_llm_models)) + run = self._learning.run if self._learning else None + return bool(run and run.task and not run.task.done()) async def _invalidate_cached_agent(self) -> bool: """ - 使当前图失效，未收敛的消费者及其模型连接继续由 Agent 持有。 + 使图失效，由图资源 owner 保留未退出消费者和关闭失败的连接。 """ - subagent_middlewares = self._subagent_middlewares - if self._compiled_agent_bundle: - self._retired_llm_models.extend(self._compiled_agent_bundle.models) - self._compiled_agent_bundle = None - pending_middlewares = await self._close_subagent_middleware_instances(subagent_middlewares) - self._subagent_middlewares = pending_middlewares - models_closed = await self._release_retired_llm_models(consumers_closed=not pending_middlewares) - return not pending_middlewares and models_closed + return await self._graphs.invalidate() @staticmethod def _latest_turn_messages(messages: List[BaseMessage]) -> List[BaseMessage]: @@ -2123,259 +1963,234 @@ class MoviePilotAgent: 创建 LangGraph Agent（使用 create_agent + SummarizationMiddleware） :param streaming: 是否启用流式输出 """ - temporary_subagent_middlewares: tuple[Any, ...] = () - temporary_models: list[Any] = [] - pending_middlewares: tuple[Any, ...] = () try: - runtime_config = await self._resolve_llm_runtime_config() - plugin_revision = _get_plugin_tools_revision() - mcp_config_signature = agent_mcp_manager.config_signature() - cached_bundle = self._compiled_agent_bundle - catalog_is_fresh = bool( - cached_bundle - and cached_bundle.streaming == streaming - and cached_bundle.tool_catalog is not None - and cached_bundle.subagent_catalog is not None - and cached_bundle.plugin_revision == plugin_revision - and cached_bundle.mcp_config_signature == mcp_config_signature - and cached_bundle.catalog_checked_at is not None - and (datetime.now() - cached_bundle.catalog_checked_at).total_seconds() - < self.TOOL_CATALOG_REFRESH_SECONDS - ) - if catalog_is_fresh: + async with self._graphs.build() as resources: + runtime_config = await self._resolve_llm_runtime_config() + plugin_revision = _get_plugin_tools_revision() + mcp_config_signature = agent_mcp_manager.config_signature() + cached_bundle = self._graphs.current + catalog_is_fresh = bool( + cached_bundle + and cached_bundle.streaming == streaming + and cached_bundle.tool_catalog is not None + and cached_bundle.subagent_catalog is not None + and cached_bundle.plugin_revision == plugin_revision + and cached_bundle.mcp_config_signature == mcp_config_signature + and cached_bundle.catalog_checked_at is not None + and (datetime.now() - cached_bundle.catalog_checked_at).total_seconds() + < self.TOOL_CATALOG_REFRESH_SECONDS + ) + if catalog_is_fresh: + bundle_signature = await self._agent_bundle_signature( + streaming, + tool_catalog=cached_bundle.tool_catalog, + subagent_catalog=cached_bundle.subagent_catalog, + ) + cached_agent = self._graphs.get(bundle_signature, streaming) + self._last_agent_cache_hit = bool(cached_agent) + if cached_agent: + logger.debug(f"复用会话内 Agent 图: session_id={self.session_id}") + return cached_agent + web_search_resolution = ServerToolRegistry.resolve_web_search( + provider=str(runtime_config.get("provider") or ""), + model=str(runtime_config.get("model") or ""), + mode=runtime_config.get("web_search_mode"), + api_protocol=runtime_config.get("api_protocol"), + base_url=runtime_config.get("base_url"), + ) + base_tool_catalog, base_subagent_catalog = self._initialize_local_tool_catalogs() + mcp_specs = await agent_mcp_manager.list_enabled_tool_specs() + local_tools = self._filter_local_web_search_tools( + base_tool_catalog.tools, + enabled=web_search_resolution.use_local_web_search, + ) + mcp_tools = await self._initialize_mcp_tools(specs=mcp_specs) + tools = [*local_tools, *mcp_tools] + local_subagent_tools = self._filter_local_web_search_tools( + base_subagent_catalog.tools, + enabled=web_search_resolution.use_local_web_search, + ) + subagent_mcp_tools = await self._initialize_subagent_mcp_tools(specs=mcp_specs) + subagent_catalog = ToolCatalogSnapshot.from_tools( + [*local_subagent_tools, *subagent_mcp_tools], + plugin_revision=base_subagent_catalog.plugin_revision, + factory_revision=base_subagent_catalog.factory_revision, + ).require_unique() + subagent_tools = [*local_subagent_tools, *subagent_mcp_tools] + # 系统提示词 + system_prompt = prompt_manager.get_agent_prompt(channel=self.channel) + + # LLM 模型（用于 agent 执行） + agent_model = await self._initialize_llm(streaming=streaming) + resources.own_model(agent_model) + self._sync_model_profile(agent_model) + # 供应商原生工具不进入本地 ToolNode，宿主策略只覆盖 client-side tools。 + server_tools = LLMHelper.get_server_tools(agent_model) + + # 为内部模型调用准备非流式 LLM，避免与用户流式回复复用同一实例。 + non_streaming_model = agent_model if not streaming else await self._initialize_llm(streaming=False) + if non_streaming_model is not agent_model: + resources.own_model(non_streaming_model) + skills_middleware = SkillsMiddleware( + sources=[str(agent_runtime_manager.skills_dir)], + bundled_skills_dir=str(get_runtime_setting("ROOT_PATH") / "skills"), + stream_handler=self.stream_handler, + ) + skill_tools = list(getattr(skills_middleware, "tools", []) or []) + user_memory_dir = agent_runtime_manager.get_user_memory_dir(self.user_id) + memory_middleware = MemoryMiddleware( + memory_dir=str(agent_runtime_manager.memory_dir), + user_memory_dir=str(user_memory_dir) if user_memory_dir else None, + store=self._learning.tools.memory if self._learning else None, + stream_handler=self.stream_handler, + ) + memory_tools = list(getattr(memory_middleware, "tools", []) or []) + learning_middleware = LearningMiddleware(self._learning) if self._learning is not None and self._learning_enabled() else None + if learning_middleware: + memory_tools.extend(learning_middleware.tools) + recall_service = getattr(self._data, "recall", None) + recall_middlewares = [] + if recall_service and build_user_memory_key(self.user_id): + history_session = await self._recall_session(agent_model) + if history_session is not None: + recall_middlewares.append(RecallMiddleware(recall_service, str(self.user_id), history_session)) + memory_tools.extend(tool for middleware in recall_middlewares for tool in middleware.tools) + policy_context = self._build_policy_context() + subagent_middlewares, subagent_task_tools = create_subagent_middlewares( + model=non_streaming_model, + tools=subagent_tools, + server_tools=server_tools, + stream_handler=self.stream_handler, + policy_context=policy_context.for_subagent(), + catalog=subagent_catalog, + ) + resources.own_middlewares(subagent_middlewares) + plan_middleware = PlanMiddleware() + output_middleware = ToolOutputMiddleware(policy_context) + invocation_repository = getattr(self._data, "invocations", None) + invocation_middlewares = [InvocationMiddleware(policy_context, invocation_repository, tools)] if invocation_repository else [] + internal_tools = [ + *skill_tools, *memory_tools, *subagent_task_tools, *plan_middleware.tools, + *output_middleware.tools, + *(tool for middleware in invocation_middlewares for tool in middleware.tools), + ] + tool_selector = self._initialize_tool_selector(tools, internal_tools, non_streaming_model) + # 严格目录必须覆盖 LangGraph ToolNode 可执行的全部 client-side 工具。 + tool_catalog = ToolCatalogSnapshot.from_tools( + [ + *local_tools, + *mcp_tools, + *internal_tools, + *(tool_selector.tools if tool_selector else []), + ], + plugin_revision=base_tool_catalog.plugin_revision, + factory_revision=base_tool_catalog.factory_revision, + ).require_unique() bundle_signature = await self._agent_bundle_signature( streaming, - tool_catalog=cached_bundle.tool_catalog, - subagent_catalog=cached_bundle.subagent_catalog, + tool_catalog=tool_catalog, + subagent_catalog=subagent_catalog, ) - cached_agent = self._get_cached_agent(bundle_signature, streaming) + cached_agent = self._graphs.get(bundle_signature, streaming) self._last_agent_cache_hit = bool(cached_agent) if cached_agent: + # 签名相同表示已编译图中的精确工具实例仍有效；新建快照仅用于复核。 + cached_bundle.catalog_checked_at = datetime.now() logger.debug(f"复用会话内 Agent 图: session_id={self.session_id}") return cached_agent - web_search_resolution = ServerToolRegistry.resolve_web_search( - provider=str(runtime_config.get("provider") or ""), - model=str(runtime_config.get("model") or ""), - mode=runtime_config.get("web_search_mode"), - api_protocol=runtime_config.get("api_protocol"), - base_url=runtime_config.get("base_url"), - ) - base_tool_catalog, base_subagent_catalog = self._initialize_local_tool_catalogs() - mcp_specs = await agent_mcp_manager.list_enabled_tool_specs() - local_tools = self._filter_local_web_search_tools( - base_tool_catalog.tools, - enabled=web_search_resolution.use_local_web_search, - ) - mcp_tools = await self._initialize_mcp_tools(specs=mcp_specs) - tools = [*local_tools, *mcp_tools] - local_subagent_tools = self._filter_local_web_search_tools( - base_subagent_catalog.tools, - enabled=web_search_resolution.use_local_web_search, - ) - subagent_mcp_tools = await self._initialize_subagent_mcp_tools(specs=mcp_specs) - subagent_catalog = ToolCatalogSnapshot.from_tools( - [*local_subagent_tools, *subagent_mcp_tools], - plugin_revision=base_subagent_catalog.plugin_revision, - factory_revision=base_subagent_catalog.factory_revision, - ).require_unique() - subagent_tools = [*local_subagent_tools, *subagent_mcp_tools] - # 系统提示词 - system_prompt = prompt_manager.get_agent_prompt(channel=self.channel) - - # LLM 模型（用于 agent 执行） - agent_model = await self._initialize_llm(streaming=streaming) - temporary_models.append(agent_model) - self._sync_model_profile(agent_model) - # 供应商原生工具不进入本地 ToolNode，宿主策略只覆盖 client-side tools。 - server_tools = LLMHelper.get_server_tools(agent_model) - - # 为内部模型调用准备非流式 LLM，避免与用户流式回复复用同一实例。 - non_streaming_model = agent_model if not streaming else await self._initialize_llm(streaming=False) - if non_streaming_model is not agent_model: - temporary_models.append(non_streaming_model) - skills_middleware = SkillsMiddleware( - sources=[str(agent_runtime_manager.skills_dir)], - bundled_skills_dir=str(get_runtime_setting("ROOT_PATH") / "skills"), - stream_handler=self.stream_handler, - ) - skill_tools = list(getattr(skills_middleware, "tools", []) or []) - user_memory_dir = agent_runtime_manager.get_user_memory_dir(self.user_id) - memory_middleware = MemoryMiddleware( - memory_dir=str(agent_runtime_manager.memory_dir), - user_memory_dir=str(user_memory_dir) if user_memory_dir else None, - store=self._learning.tools.memory if self._learning else None, - stream_handler=self.stream_handler, - ) - memory_tools = list(getattr(memory_middleware, "tools", []) or []) - learning_middleware = LearningMiddleware(self._learning) if self._learning is not None and self._learning_enabled() else None - if learning_middleware: - memory_tools.extend(learning_middleware.tools) - recall_service = getattr(self._data, "recall", None) - recall_middlewares = [] - if recall_service and build_user_memory_key(self.user_id): - history_session = await self._recall_session(agent_model) - if history_session is not None: - recall_middlewares.append(RecallMiddleware(recall_service, str(self.user_id), history_session)) - memory_tools.extend(tool for middleware in recall_middlewares for tool in middleware.tools) - policy_context = self._build_policy_context() - subagent_middlewares, subagent_task_tools = create_subagent_middlewares( - model=non_streaming_model, - tools=subagent_tools, - server_tools=server_tools, - stream_handler=self.stream_handler, - policy_context=policy_context.for_subagent(), - catalog=subagent_catalog, - ) - temporary_subagent_middlewares = tuple(subagent_middlewares) - plan_middleware = PlanMiddleware() - output_middleware = ToolOutputMiddleware(policy_context) - invocation_repository = getattr(self._data, "invocations", None) - invocation_middlewares = [InvocationMiddleware(policy_context, invocation_repository, tools)] if invocation_repository else [] - internal_tools = [ - *skill_tools, *memory_tools, *subagent_task_tools, *plan_middleware.tools, - *output_middleware.tools, - *(tool for middleware in invocation_middlewares for tool in middleware.tools), - ] - tool_selector = self._initialize_tool_selector(tools, internal_tools, non_streaming_model) - # 严格目录必须覆盖 LangGraph ToolNode 可执行的全部 client-side 工具。 - tool_catalog = ToolCatalogSnapshot.from_tools( - [ - *local_tools, - *mcp_tools, - *internal_tools, - *(tool_selector.tools if tool_selector else []), - ], - plugin_revision=base_tool_catalog.plugin_revision, - factory_revision=base_tool_catalog.factory_revision, - ).require_unique() - bundle_signature = await self._agent_bundle_signature( - streaming, - tool_catalog=tool_catalog, - subagent_catalog=subagent_catalog, - ) - cached_agent = self._get_cached_agent(bundle_signature, streaming) - self._last_agent_cache_hit = bool(cached_agent) - if cached_agent: - # 签名相同表示已编译图中的精确工具实例仍有效；新建快照仅用于复核。 - cached_bundle.catalog_checked_at = datetime.now() - pending_middlewares = await self._close_subagent_middleware_instances(temporary_subagent_middlewares) - self._subagent_middlewares = self._merge_subagent_middleware_owners( - self._subagent_middlewares, - pending_middlewares, + summarization_middleware = SummarizationMiddleware( + model=non_streaming_model, + trigger=("fraction", 0.85), + keep=("messages", 20), ) - temporary_subagent_middlewares = () - logger.debug(f"复用会话内 Agent 图: session_id={self.session_id}") - return cached_agent - summarization_middleware = SummarizationMiddleware( - model=non_streaming_model, - trigger=("fraction", 0.85), - keep=("messages", 20), - ) - # 中间件 - guardrails = ToolGuardrailsMiddleware(policy_context) - policy_middleware = AgentPolicyMiddleware( - context=policy_context, catalog=tool_catalog, tools=tools, guardrails=guardrails, - ) - code_middleware = CodeExecutionMiddleware(policy_middleware) - middlewares = [ - # 宿主策略必须位于最外层，确保插件覆盖工具基类也不能绕过。 - policy_middleware, - code_middleware, - # 运行中补充消息只在模型回合边界进入同一张图，不启动并行 Agent。 - *([SteeringMiddleware()] if self._steering_inbox.running else []), - output_middleware, - *invocation_middlewares, - skills_middleware, - # Jobs 任务管理 - JobsMiddleware( - sources=[str(agent_runtime_manager.jobs_dir)], - ), - # 运行时人格与核心规则 - RuntimeConfigMiddleware(), - # 计划独立保存，最终请求压缩仍计入其系统上下文预算。 - plan_middleware, - # 稳定偏好与原始会话证据分别管理。 - memory_middleware, - *recall_middlewares, - # 以工具原文及模型请求顺序检测循环，外层归档仍保留真实回执。 - guardrails, - *([learning_middleware] if learning_middleware else []), - # 错误工具调用修复 - PatchToolCallsMiddleware(), - # 子代理委派 - *subagent_middlewares, - ] - - # 工具选择 - if tool_selector is not None: - middlewares.append(tool_selector) - middlewares.append(CodeCaptureMiddleware(code_middleware)) - - # 所有压缩都在最终请求边界完成，避免主模型失败前写入摘要状态。 - middlewares.append( - FinalRequestCompactionMiddleware( - summarizer=summarization_middleware, + # 中间件 + guardrails = ToolGuardrailsMiddleware(policy_context) + policy_middleware = AgentPolicyMiddleware( + context=policy_context, catalog=tool_catalog, tools=tools, guardrails=guardrails, ) - ) - middlewares.append(VisionMiddleware()) - if learning_middleware: - middlewares.append(LearningCaptureMiddleware(learning_middleware.session)) + code_middleware = CodeExecutionMiddleware(policy_middleware) + middlewares = [ + # 宿主策略必须位于最外层，确保插件覆盖工具基类也不能绕过。 + policy_middleware, + code_middleware, + # 运行中补充消息只在模型回合边界进入同一张图，不启动并行 Agent。 + *([SteeringMiddleware()] if self._steering_inbox.running else []), + output_middleware, + *invocation_middlewares, + skills_middleware, + # Jobs 任务管理 + JobsMiddleware( + sources=[str(agent_runtime_manager.jobs_dir)], + ), + # 运行时人格与核心规则 + RuntimeConfigMiddleware(), + # 计划独立保存，最终请求压缩仍计入其系统上下文预算。 + plan_middleware, + # 稳定偏好与原始会话证据分别管理。 + memory_middleware, + *recall_middlewares, + # 以工具原文及模型请求顺序检测循环，外层归档仍保留真实回执。 + guardrails, + *([learning_middleware] if learning_middleware else []), + # 错误工具调用修复 + PatchToolCallsMiddleware(), + # 子代理委派 + *subagent_middlewares, + ] - # 预算观察器必须位于最内层，才能看到动态 system 和最终筛选后的工具。 - middlewares.append( - UsageMiddleware( - on_usage=self._record_usage, - on_request_budget=self._record_request_budget, - next_request_sequence=self._next_request_sequence, + # 工具选择 + if tool_selector is not None: + middlewares.append(tool_selector) + middlewares.append(CodeCaptureMiddleware(code_middleware)) + + # 所有压缩都在最终请求边界完成，避免主模型失败前写入摘要状态。 + middlewares.append( + FinalRequestCompactionMiddleware( + summarizer=summarization_middleware, + ) ) - ) + middlewares.append(VisionMiddleware()) + if learning_middleware: + middlewares.append(LearningCaptureMiddleware(learning_middleware.session)) - agent = create_agent( - model=agent_model, - tools=[ - *tools, - *skill_tools, - *memory_tools, - *server_tools, - ], - system_prompt=system_prompt, - middleware=middlewares, - checkpointer=InMemorySaver(), - ) - cached_agent = await self._cache_agent( - signature=bundle_signature, - agent=agent, - streaming=streaming, - tool_catalog=tool_catalog, - subagent_catalog=subagent_catalog, - mcp_config_signature=mcp_config_signature, - subagent_middlewares=tuple(subagent_middlewares), - models=tuple(temporary_models), - ) - temporary_subagent_middlewares = () - temporary_models = [] - return cached_agent - except asyncio.CancelledError: - pending_middlewares = await self._close_subagent_middleware_instances(temporary_subagent_middlewares) - self._subagent_middlewares = self._merge_subagent_middleware_owners( - self._subagent_middlewares, - pending_middlewares, - ) + # 预算观察器必须位于最内层，才能看到动态 system 和最终筛选后的工具。 + middlewares.append( + UsageMiddleware( + on_usage=self._record_usage, + on_request_budget=self._record_request_budget, + next_request_sequence=self._next_request_sequence, + ) + ) + + agent = create_agent( + model=agent_model, + tools=[ + *tools, + *skill_tools, + *memory_tools, + *server_tools, + ], + system_prompt=system_prompt, + middleware=middlewares, + checkpointer=InMemorySaver(), + ) + self._graphs.publish(CompiledAgentBundle( + signature=bundle_signature, + agent=agent, + streaming=streaming, + created_at=datetime.now(), + tool_catalog=tool_catalog, + subagent_catalog=subagent_catalog, + plugin_revision=tool_catalog.plugin_revision, + mcp_config_signature=mcp_config_signature, + catalog_checked_at=datetime.now(), + resources=resources, + )) + return agent + except Exception as error: + logger.error(f"创建 Agent 失败: {error}") raise - except Exception as e: - pending_middlewares = await self._close_subagent_middleware_instances(temporary_subagent_middlewares) - self._subagent_middlewares = self._merge_subagent_middleware_owners( - self._subagent_middlewares, - pending_middlewares, - ) - logger.error(f"创建 Agent 失败: {e}") - raise - finally: - if pending_middlewares: - self._retired_llm_models.extend(temporary_models) - else: - await self._release_llm_models(tuple(temporary_models)) async def process( self, diff --git a/app/api/endpoints/openai.py b/app/api/endpoints/openai.py index 575769fec..bb50a3d30 100644 --- a/app/api/endpoints/openai.py +++ b/app/api/endpoints/openai.py @@ -49,6 +49,7 @@ from app.schemas.openai import OpenAIUsage as _SchemaOpenAIUsage from app.schemas.types import NotificationChannel if TYPE_CHECKING: + from app.agent.graph import AgentGraphCache from app.agent.manager import AgentManager from app.agent.orchestrator import MoviePilotAgent @@ -70,8 +71,12 @@ OpenAIStreamEvent = str | dict[str, str] | None class _CollectingMoviePilotAgentMixin: """ 捕获 Agent 最终输出，避免再通过消息渠道二次发送。 + + :ivar _graphs: 运行时 Agent 基类提供的会话图资源 owner """ + _graphs: AgentGraphCache + def __init__(self, *args, stream_mode: bool = False, **kwargs): """ 初始化协议 Agent 的消息收集与流式模式。 @@ -107,7 +112,7 @@ class _CollectingMoviePilotAgentMixin: self.stream_handler = _get_openai_streaming_handler_type()() self.stream_handler.bind_queue(event_queue) # 已编译工具持有旧 handler；identity 变化时必须重建图和工具目录。 - self._compiled_agent_bundle = None + self._graphs.retire() def release_protocol_request( self, diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 1879ecf95..a2fe6196a 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -816,8 +816,8 @@ flowchart LR | 指标 | 当前值 | |---|---:| -| Python 模块 | 1123 | -| 内部导入边 | 9,579 | +| Python 模块 | 1124 | +| 内部导入边 | 9,587 | | 非平凡 SCC | 1（29 个模块的 TMDB vendor 精确 containment） | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 51（精确 policy 准入，Application / Chain 均为 0） | diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index 35fc26dc9..39a638619 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -97,7 +97,8 @@ to make the directory tree look symmetrical. | `app/agent/session.py` | Per-session queues, workers, active Agent instances, status projection, cancellation and deferred cleanup | | `app/agent/lifecycle.py` | Runtime generation admission, initialization, idle-session collection and bounded manager shutdown | | `app/agent/tasks.py` | Background prompts, durable scheduled-task execution and heartbeat wakeups | -| `app/agent/orchestrator.py` | One `MoviePilotAgent` execution instance: prompt/tool/middleware assembly, model invocation, streaming and per-agent state | +| `app/agent/orchestrator.py` | One `MoviePilotAgent` execution instance: prompt/tool/middleware assembly, model invocation, streaming and per-agent state; graph resources are registered with the graph owner immediately after acquisition | +| `app/agent/graph.py` | Session-local compiled graph cache and model/controller ownership: one resource scope follows construction through publication and retirement; publication transfers ownership synchronously, cancellation retains unclosed resources, and consumers finish before their models close. Background learning delays model reclamation; shutdown seals in-flight scopes and refuses late publication. This is an internal owner, not an SDK or Compat export | | `app/agent/middleware/plan.py` | Current task objective, step status and evidence in graph state; sanitized snapshots travel through existing message persistence and never authorize tool effects | | `app/application/messaging/recall.py` / `app/agent/history/` | Application 持有独立消息库 DTO/Port 和后台维护服务；Agent 专属 SQLite/FTS5 schema、存储、检索、窗口与旧快照导入由 `agent/history/` 拥有，数据库位于 Agent 运行目录且不共享主库 Session。`db/adapters/recall.py` 仅负责一次性旧快照只读导出 | | `app/agent/learning/` / `app/agent/middleware/learning.py` | Agent 专属个人技能、稳定记忆和复盘生命周期；独立文件目录按用户隔离，模型仅能调用受控维护工具。`session.py` 管理计数及 TaskRegistry owner，`review.py` 复用最终模型请求并在派发处限制工具，`files.py` 只封装该存储的路径、进程锁与原子替换，不作为通用文件 Adapter。人工提案/技能控制命令经 messaging interaction 契约路由，只有宿主收到的用户原文可以触发，后台快照没有执行权限 | diff --git a/scripts/evaluation/runtime.py b/scripts/evaluation/runtime.py index c96a9b0e6..58bdaebc4 100644 --- a/scripts/evaluation/runtime.py +++ b/scripts/evaluation/runtime.py @@ -584,7 +584,7 @@ def _agent_type() -> type: 保留原始执行图的观察引用，生产失败恢复清缓存后仍可导出失败轨迹。 """ graph = await super()._create_agent(streaming=streaming) - self.evaluation_bundle = self._compiled_agent_bundle + self.evaluation_bundle = self._graphs.current return graph async def _initialize_llm(self, streaming: bool = False) -> Any: diff --git a/tests/agent_graph.py b/tests/agent_graph.py new file mode 100644 index 000000000..101a9096b --- /dev/null +++ b/tests/agent_graph.py @@ -0,0 +1,37 @@ +""" +通过真实图资源 owner 为 Agent 行为测试装配已发布图。 +""" + +from datetime import datetime +from typing import Any, Optional + +from app.agent.graph import CompiledAgentBundle, GraphResources +from app.agent.orchestrator import MoviePilotAgent + + +def publish_graph( + agent: MoviePilotAgent, + *, + graph: Optional[Any] = None, + models: tuple[Any, ...] = (), + middlewares: tuple[object, ...] = (), +) -> CompiledAgentBundle: + """ + 为无需真实构图的回归装配资源完整的缓存图，不模拟生命周期实现。 + + :param agent: 被测会话执行对象 + :param graph: 供执行的测试图；None 时使用占位对象 + :param models: 需观察客户端释放的模型 + :param middlewares: 需观察关闭及未收敛重试的控制器 + :return: 已发布图，供断言资源归属和缓存身份 + """ + resources = GraphResources() + for model in models: + resources.own_model(model) + resources.own_middlewares(middlewares) + bundle = CompiledAgentBundle( + signature=("test",), agent=graph if graph is not None else object(), + streaming=False, created_at=datetime.now(), resources=resources, + ) + agent._graphs.publish(bundle) + return bundle diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 518372127..8f47c9993 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -974,8 +974,8 @@ "runtime_only": true } }, - "edge_count": 9579, - "edge_sha256": "fbbb828033ea217b8461f87d24cba67d08d93086bac88007f6445dbf34ec2eb2", + "edge_count": 9587, + "edge_sha256": "c95b2c94aa5c65c69b1374ca4e0014241e841d8520004141ba39e12f3052286a", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -1300,6 +1300,13 @@ "app.agent.code.result -> app.agent.policy.sanitizer", "app.agent.contracts -> app.schemas", "app.agent.contracts -> app.schemas.types", + "app.agent.graph -> app.agent", + "app.agent.graph -> app.agent.llm", + "app.agent.graph -> app.agent.llm.helper", + "app.agent.graph -> app.agent.tools", + "app.agent.graph -> app.agent.tools.catalog", + "app.agent.graph -> app.runtime", + "app.agent.graph -> app.runtime.log", "app.agent.guardrails.controller -> app.agent", "app.agent.guardrails.controller -> app.agent.guardrails", "app.agent.guardrails.controller -> app.agent.guardrails.results", @@ -1679,6 +1686,7 @@ "app.agent.orchestrator -> app.agent", "app.agent.orchestrator -> app.agent.callback", "app.agent.orchestrator -> app.agent.contracts", + "app.agent.orchestrator -> app.agent.graph", "app.agent.orchestrator -> app.agent.learning", "app.agent.orchestrator -> app.agent.learning.session", "app.agent.orchestrator -> app.agent.llm", @@ -10557,7 +10565,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 1123, + "module_count": 1124, "modules": [ "app", "app.adapters", @@ -10632,6 +10640,7 @@ "app.agent.code.manager", "app.agent.code.result", "app.agent.contracts", + "app.agent.graph", "app.agent.guardrails", "app.agent.guardrails.completion", "app.agent.guardrails.controller", diff --git a/tests/fixtures/architecture/mypy-baseline.json b/tests/fixtures/architecture/mypy-baseline.json index 5091518f3..d874aea1d 100644 --- a/tests/fixtures/architecture/mypy-baseline.json +++ b/tests/fixtures/architecture/mypy-baseline.json @@ -114,8 +114,7 @@ "no-any-return": 6, "no-untyped-def": 6, "type-arg": 12, - "union-attr": 3, - "var-annotated": 1 + "union-attr": 3 }, "app/agent/skills/registry.py": { "misc": 3, diff --git a/tests/test_agent_api_lazy_imports.py b/tests/test_agent_api_lazy_imports.py index eb2385e0d..46f57fe32 100644 --- a/tests/test_agent_api_lazy_imports.py +++ b/tests/test_agent_api_lazy_imports.py @@ -304,6 +304,8 @@ import asyncio import json import sys import types +from datetime import datetime +from unittest.mock import Mock sites = types.ModuleType("app.application.site.sites") sites.SitesHelper = type("SitesHelper", (), {}) @@ -311,12 +313,13 @@ sites.__file__ = "" sys.modules["app.application.site.sites"] = sites from app.api.endpoints import openai +from app.agent.graph import AgentGraphCache, CompiledAgentBundle, GraphResources class RuntimeAgent: def __init__(self, **kwargs): self.__dict__.update(kwargs) self.stream_handler = object() - self._compiled_agent_bundle = object() + self._graphs = AgentGraphCache(models_in_use=lambda: False) class TestStreamingHandler(openai._OpenAIStreamingHandlerMixin): pass @@ -327,11 +330,23 @@ agent = agent_type(session_id="stable", user_id="api") first_queue = asyncio.Queue() second_queue = asyncio.Queue() +resources = GraphResources() +consumer = Mock(close=Mock(return_value=True)) +resources.own_middlewares([consumer]) +agent._graphs.publish(CompiledAgentBundle( + signature=("old",), agent=object(), streaming=False, + created_at=datetime.now(), resources=resources, +)) agent.configure_protocol_request(stream_mode=True, event_queue=first_queue) +invalidated = agent._graphs.current is None +retained = not resources.empty +asyncio.run(agent._graphs.collect()) handler = agent.stream_handler handler._event_queue.put_nowait("first") -compiled_bundle = object() -agent._compiled_agent_bundle = compiled_bundle +compiled_bundle = CompiledAgentBundle( + signature=("new",), agent=object(), streaming=True, created_at=datetime.now(), +) +agent._graphs.publish(compiled_bundle) agent.configure_protocol_request(stream_mode=True, event_queue=second_queue) agent.release_protocol_request(first_queue) @@ -339,8 +354,11 @@ handler._event_queue.put_nowait("second") agent.release_protocol_request(second_queue) print(json.dumps({ + "invalidated": invalidated, + "retained_until_collection": retained, + "consumer_closes": consumer.close.call_count, "same_handler": agent.stream_handler is handler, - "same_bundle": agent._compiled_agent_bundle is compiled_bundle, + "same_bundle": agent._graphs.current is compiled_bundle, "first": first_queue.get_nowait(), "first_empty": first_queue.empty(), "second": second_queue.get_nowait(), @@ -352,6 +370,9 @@ print(json.dumps({ ) assert result == { + "invalidated": True, + "retained_until_collection": True, + "consumer_closes": 1, "same_handler": True, "same_bundle": True, "first": "first", @@ -432,7 +453,6 @@ class RuntimeAgent: def __init__(self, **kwargs): self.__dict__.update(kwargs) self.stream_handler = object() - self._compiled_agent_bundle = None async def process(self, _prompt, **_kwargs): return "runtime reply" diff --git a/tests/test_agent_graph_cache.py b/tests/test_agent_graph_cache.py index 6084fbf68..79b7498ff 100644 --- a/tests/test_agent_graph_cache.py +++ b/tests/test_agent_graph_cache.py @@ -11,12 +11,13 @@ import pytest from langchain_core.messages import AIMessage, HumanMessage from app.agent.contracts import ReplyMode +from app.agent.graph import CompiledAgentBundle from app.agent.mcp import AgentMcpToolSpec from app.agent.middleware.invocation import GET_TOOL_EXECUTION_NAME from app.agent.middleware.output import READ_TOOL_RESULT_NAME from app.agent.middleware.plan import PLAN_TOOL_NAME from app.agent.middleware.selection import TOOL_DISCOVERY_NAME -from app.agent.orchestrator import MoviePilotAgent, _CompiledAgentBundle +from app.agent.orchestrator import MoviePilotAgent from app.agent.tools.catalog import ( ToolCatalogSnapshot, ToolIdentityAmbiguousError, @@ -103,7 +104,7 @@ async def test_create_agent_reuses_cached_graph_when_signature_matches(): cached_graph = object() catalog = ToolCatalogSnapshot.from_tools([], plugin_revision=0, factory_revision="factory-v1") agent = MoviePilotAgent(session_id="cache-hit", user_id="user-1") - agent._compiled_agent_bundle = _CompiledAgentBundle( + agent._graphs.publish(CompiledAgentBundle( signature=("sig",), agent=cached_graph, streaming=False, @@ -113,7 +114,7 @@ async def test_create_agent_reuses_cached_graph_when_signature_matches(): plugin_revision=0, mcp_config_signature="mcp-config", catalog_checked_at=datetime.now(), - ) + )) with ( patch.object( @@ -146,7 +147,7 @@ async def test_fresh_catalog_cache_hit_skips_tool_and_mcp_discovery() -> None: cached_graph = object() catalog = ToolCatalogSnapshot.from_tools([], plugin_revision=0, factory_revision="factory-v1") agent = MoviePilotAgent(session_id="catalog-cache-hit", user_id="user-1") - agent._compiled_agent_bundle = _CompiledAgentBundle( + agent._graphs.publish(CompiledAgentBundle( signature=("sig",), agent=cached_graph, streaming=False, @@ -156,7 +157,7 @@ async def test_fresh_catalog_cache_hit_skips_tool_and_mcp_discovery() -> None: plugin_revision=0, mcp_config_signature="mcp-config", catalog_checked_at=datetime.now(), - ) + )) with ( patch.object( @@ -197,7 +198,7 @@ async def test_expired_unchanged_catalog_renews_freshness() -> None: catalog = ToolCatalogSnapshot.from_tools([], plugin_revision=0, factory_revision="factory-v1") expired_at = datetime.now() - timedelta(minutes=5) agent = MoviePilotAgent(session_id="catalog-refresh", user_id="user-1") - agent._compiled_agent_bundle = _CompiledAgentBundle( + agent._graphs.publish(CompiledAgentBundle( signature=("sig",), agent=cached_graph, streaming=False, @@ -207,7 +208,7 @@ async def test_expired_unchanged_catalog_renews_freshness() -> None: plugin_revision=0, mcp_config_signature="mcp-config", catalog_checked_at=expired_at, - ) + )) fake_llm = SimpleNamespace( _llm_type="openai-chat", model="fake", @@ -283,7 +284,7 @@ async def test_expired_unchanged_catalog_renews_freshness() -> None: graph = await agent._create_agent(streaming=False) assert graph is cached_graph - assert agent._compiled_agent_bundle.catalog_checked_at > expired_at + assert agent._graphs.current.catalog_checked_at > expired_at temporary_middleware.close.assert_awaited_once() fake_llm.http_client.close.assert_called_once() fake_llm.http_async_client.aclose.assert_awaited_once() diff --git a/tests/test_agent_graph_resources.py b/tests/test_agent_graph_resources.py new file mode 100644 index 000000000..644ec0f16 --- /dev/null +++ b/tests/test_agent_graph_resources.py @@ -0,0 +1,335 @@ +""" +验证构建、图替换、取消和关闭之间的真实资源所有权转换。 +""" + +import asyncio +from datetime import datetime +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest + +from app.agent.graph import AgentGraphCache, CompiledAgentBundle, GraphResources +from app.agent.llm.helper import LLMHelper + + +@pytest.fixture +def anyio_backend(): + """ + 生命周期屏障使用 asyncio 事件循环。 + """ + return "asyncio" + + +def _bundle(resources: GraphResources, revision: int = 0) -> CompiledAgentBundle: + """ + 以真实资源范围构造不同签名的占位执行图。 + """ + return CompiledAgentBundle( + signature=(revision,), agent=object(), streaming=False, + created_at=datetime.now(), resources=resources, + ) + + +@pytest.mark.anyio +@pytest.mark.parametrize("failure", [None, RuntimeError("构建失败"), asyncio.CancelledError()]) +async def test_unpublished_build_closes_consumers_before_models(monkeypatch, failure): + """ + 正常复核返回、异常和取消均关闭未发布范围，消费者必须先于模型退出。 + """ + events = [] + cache = AgentGraphCache(models_in_use=lambda: False) + model = object() + + async def close_consumer(): + """ + 记录消费者退出，验证其使用模型期间不会提前释放连接。 + """ + events.append("consumer") + return True + + async def close_model(value): + """ + 只允许关闭本范围接管的模型。 + """ + assert value is model + events.append("model") + return True + + monkeypatch.setattr(LLMHelper, "close_llm", close_model) + + async def build(): + """ + 在登记完整资源后模拟三种构图退出路径。 + """ + async with cache.build() as resources: + resources.own_model(model) + resources.own_model(model) + resources.own_middlewares([SimpleNamespace(close=close_consumer), object()]) + if failure is not None: + raise failure + + if failure is None: + await build() + else: + with pytest.raises(type(failure)): + await build() + assert events == ["consumer", "model"] + assert await cache.invalidate() is True + assert events == ["consumer", "model"] + + +@pytest.mark.anyio +async def test_cancel_during_post_publish_collection_keeps_new_graph_alive(monkeypatch): + """ + 新图已发布后取消旧图回收，不能把新图误当作未提交的临时资源关闭。 + """ + cache = AgentGraphCache(models_in_use=lambda: False) + old_model, new_model = object(), object() + started, release = asyncio.Event(), asyncio.Event() + closed = [] + + async def close_model(model): + """ + 在旧模型关闭前阻塞，使取消准确落在图发布后的回收阶段。 + """ + if model is old_model: + started.set() + await release.wait() + closed.append(model) + return True + + monkeypatch.setattr(LLMHelper, "close_llm", close_model) + async with cache.build() as resources: + resources.own_model(old_model) + cache.publish(_bundle(resources)) + + async def replace(): + """ + 在真实构建范围内同步发布新图，再由退出逻辑回收旧图。 + """ + async with cache.build() as replacement: + replacement.own_model(new_model) + cache.publish(_bundle(replacement, 1)) + + task = asyncio.create_task(replace()) + await started.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert cache.get((1,), False) is cache.current.agent + assert closed == [] + release.set() + await cache.collect() + assert closed == [old_model] + assert await cache.invalidate() is True + assert closed == [old_model, new_model] + + +@pytest.mark.anyio +@pytest.mark.parametrize("first_result", [False, RuntimeError("关闭失败")]) +async def test_pending_consumers_retain_only_their_own_models(monkeypatch, first_result): + """ + 某代消费者未退出时保留对应模型，但不能阻塞另一独立范围释放。 + """ + cache = AgentGraphCache(models_in_use=lambda: False) + close_model = AsyncMock(return_value=True) + monkeypatch.setattr(LLMHelper, "close_llm", close_model) + model, independent_model = object(), object() + consumer = SimpleNamespace(close=AsyncMock(side_effect=[first_result, False, True])) + async with cache.build() as resources: + resources.own_model(model) + resources.own_middlewares([consumer]) + close_model.assert_not_awaited() + async with cache.build() as independent: + independent.own_model(independent_model) + close_model.assert_awaited_once_with(independent_model) + assert await cache.invalidate() is True + assert [call.args[0] for call in close_model.await_args_list] == [independent_model, model] + + +@pytest.mark.anyio +@pytest.mark.parametrize("phase", ["consumer", "model"]) +async def test_interrupted_collection_can_retry_without_losing_resources(monkeypatch, phase): + """ + 消费者或模型关闭中再次取消，资源必须仍由原 owner 保留并可重试。 + """ + cache = AgentGraphCache(models_in_use=lambda: False) + started, release = asyncio.Event(), asyncio.Event() + model = object() + + async def blocked_close(*_args): + """ + 在真实关闭边界等待，用屏障代替不稳定的时间等待。 + """ + started.set() + await release.wait() + return True + + consumer = SimpleNamespace(close=AsyncMock(return_value=True)) + close_model = AsyncMock(return_value=True) + if phase == "consumer": + consumer.close.side_effect = blocked_close + else: + close_model.side_effect = blocked_close + monkeypatch.setattr(LLMHelper, "close_llm", close_model) + async with cache.build() as resources: + resources.own_model(model) + resources.own_middlewares([consumer]) + cache.publish(_bundle(resources)) + task = asyncio.create_task(cache.invalidate()) + await started.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert cache.current is None + assert not resources.empty + release.set() + assert await cache.invalidate() is True + assert resources.empty + assert consumer.close.await_count == (2 if phase == "consumer" else 1) + assert close_model.await_count == (1 if phase == "consumer" else 2) + + +@pytest.mark.anyio +async def test_shutdown_tracks_inflight_build_and_seals_late_controller(monkeypatch): + """ + 关停与构建交错时不提前释放在用模型，迟到控制器立即封口，迟到图拒绝发布。 + """ + cache = AgentGraphCache(models_in_use=lambda: False) + close_model = AsyncMock(return_value=True) + monkeypatch.setattr(LLMHelper, "close_llm", close_model) + model = object() + consumer = SimpleNamespace(seal=Mock(), close=AsyncMock(return_value=True)) + async with cache.build() as resources: + resources.own_model(model) + cache.seal() + assert await cache.invalidate() is False + close_model.assert_not_awaited() + resources.own_middlewares([consumer]) + consumer.seal.assert_called_once() + with pytest.raises(RuntimeError, match="正在关闭"): + cache.publish(_bundle(resources)) + close_model.assert_awaited_once_with(model) + assert await cache.invalidate() is True + with pytest.raises(RuntimeError, match="正在关闭"): + async with cache.build(): + pytest.fail("关闭后不得进入构建范围") + + +@pytest.mark.anyio +async def test_concurrent_collection_does_not_double_close(monkeypatch): + """ + 两个清理请求并发时只执行一次实际模型关闭。 + """ + cache = AgentGraphCache(models_in_use=lambda: False) + started, release = asyncio.Event(), asyncio.Event() + + async def close_model(_model): + """ + 保持首次关闭在途，让第二次清理与其真实交错。 + """ + started.set() + await release.wait() + return True + + close = AsyncMock(side_effect=close_model) + monkeypatch.setattr(LLMHelper, "close_llm", close) + async with cache.build() as resources: + resources.own_model(object()) + cache.publish(_bundle(resources)) + first = asyncio.create_task(cache.invalidate()) + await started.wait() + second = asyncio.create_task(cache.invalidate()) + release.set() + assert await asyncio.gather(first, second) == [True, True] + close.assert_awaited_once() + + +@pytest.mark.anyio +async def test_retirement_during_collection_keeps_shared_model_for_late_consumer(monkeypatch): + """ + 清理等待期间另一图退休时，也必须等待其消费者退出后再释放共享模型。 + """ + cache = AgentGraphCache(models_in_use=lambda: False) + started, release = asyncio.Event(), asyncio.Event() + model = object() + close = AsyncMock(return_value=True) + monkeypatch.setattr(LLMHelper, "close_llm", close) + + async def close_old_consumer(): + """ + 阻塞旧消费者退出，为新图退休提供确定的交错点。 + """ + started.set() + await release.wait() + return True + + async with cache.build() as resources: + resources.own_model(model) + resources.own_middlewares([SimpleNamespace(close=close_old_consumer)]) + cache.publish(_bundle(resources)) + cache.retire() + newer = GraphResources() + consumer = SimpleNamespace(close=AsyncMock(return_value=True)) + newer.own_model(model) + newer.own_middlewares([consumer]) + cache.publish(_bundle(newer, 1)) + collecting = asyncio.create_task(cache.collect()) + await started.wait() + cache.retire() + release.set() + await collecting + close.assert_not_awaited() + consumer.close.assert_not_awaited() + assert await cache.invalidate() is True + consumer.close.assert_awaited_once() + close.assert_awaited_once_with(model) + + +@pytest.mark.anyio +async def test_repeated_replacement_preserves_shared_resources_and_releases_once(monkeypatch): + """ + 多轮替换共享同一模型与控制器时，只有最终失效才释放各一次。 + """ + cache = AgentGraphCache(models_in_use=lambda: False) + model = object() + consumer = SimpleNamespace(close=AsyncMock(return_value=True)) + close = AsyncMock(return_value=True) + monkeypatch.setattr(LLMHelper, "close_llm", close) + for revision in range(20): + async with cache.build() as resources: + resources.own_model(model) + resources.own_middlewares([consumer]) + bundle = _bundle(resources, revision) + cache.publish(bundle) + cache.publish(bundle) + assert cache.get((revision,), False) is bundle.agent + assert cache.get((revision,), True) is None + close.assert_not_awaited() + consumer.close.assert_not_awaited() + assert await cache.invalidate() is True + close.assert_awaited_once_with(model) + consumer.close.assert_awaited_once() + + +@pytest.mark.anyio +async def test_learning_and_close_failure_keep_resource_until_both_converge(monkeypatch): + """ + 后台复盘退出后才能尝试关闭；关闭失败仍保留资源至后续成功。 + """ + learning = asyncio.get_running_loop().create_future() + cache = AgentGraphCache(models_in_use=lambda: not learning.done()) + close = AsyncMock(side_effect=[False, True]) + monkeypatch.setattr(LLMHelper, "close_llm", close) + async with cache.build() as resources: + resources.own_model(object()) + cache.publish(_bundle(resources)) + assert await cache.invalidate() is False + close.assert_not_awaited() + learning.set_result(None) + assert await cache.invalidate() is False + assert not resources.empty + assert await cache.invalidate() is True + assert resources.empty + assert close.await_count == 2 diff --git a/tests/test_agent_recovery.py b/tests/test_agent_recovery.py index 677d51333..e33d783b8 100644 --- a/tests/test_agent_recovery.py +++ b/tests/test_agent_recovery.py @@ -18,6 +18,7 @@ from app.agent.memory import MemoryManager from app.agent.middleware.patching import PatchToolCallsMiddleware from app.agent.middleware.plan import PLAN_SNAPSHOT_KEY from app.agent.orchestrator import MoviePilotAgent +from tests.agent_graph import publish_graph class _InterruptedGraph: @@ -114,12 +115,12 @@ async def test_failure_preserves_completed_receipts_and_marks_unknown_outcomes(r original = _completed_tool_messages() graph = _InterruptedGraph(original) agent._create_agent = AsyncMock(return_value=graph) - agent._compiled_agent_bundle = SimpleNamespace(models=()) + publish_graph(agent) result, _ = await agent._execute_agent(original[:1]) assert result == "智能助手执行失败，请稍后重试" - assert agent._compiled_agent_bundle is None + assert agent._graphs.current is None recovered = await agent._memory.async_get_agent_messages(agent.session_id, agent.user_id) receipts = [message for message in recovered if isinstance(message, ToolMessage)] assert receipts[0].content == original[2].content @@ -151,7 +152,7 @@ async def test_cancellation_preserves_snapshot_and_still_propagates(recovery_age agent._should_stream = lambda: streaming graph = _InterruptedGraph(_completed_tool_messages(), blocked=True) agent._create_agent = AsyncMock(return_value=graph) - agent._compiled_agent_bundle = SimpleNamespace(models=()) + publish_graph(agent) execution = asyncio.create_task(agent._execute_agent(graph.messages[:1])) await graph.started.wait() execution.cancel() @@ -159,7 +160,7 @@ async def test_cancellation_preserves_snapshot_and_still_propagates(recovery_age with pytest.raises(asyncio.CancelledError): await execution - assert agent._compiled_agent_bundle is None + assert agent._graphs.current is None recovered = await agent._memory.async_get_agent_messages(agent.session_id, agent.user_id) assert "already-created" in str(messages_to_dict(recovered)) assert "取消" in recovered[-1].content @@ -210,12 +211,12 @@ async def test_snapshot_failure_does_not_replace_original_error(recovery_agent, agent._get_recursion_limit = lambda: int("invalid configuration") if failure == "persistence": agent._memory.async_save_agent_messages = AsyncMock(side_effect=RuntimeError("save failed")) - agent._compiled_agent_bundle = SimpleNamespace(models=()) + publish_graph(agent) result, _ = await agent._execute_agent(graph.messages[:1]) assert result == "智能助手执行失败，请稍后重试" - assert agent._compiled_agent_bundle is None + assert agent._graphs.current is None agent.stream_handler.stop_streaming.assert_awaited_once() @@ -269,7 +270,7 @@ async def test_second_cancellation_during_snapshot_still_invalidates_graph(recov agent, persistence = recovery_agent graph = _InterruptedGraph(_completed_tool_messages(), blocked=True) agent._create_agent = AsyncMock(return_value=graph) - agent._compiled_agent_bundle = SimpleNamespace(models=()) + publish_graph(agent) saving = asyncio.Event() async def block_save(**_kwargs): @@ -289,7 +290,7 @@ async def test_second_cancellation_during_snapshot_still_invalidates_graph(recov with pytest.raises(asyncio.CancelledError): await execution - assert agent._compiled_agent_bundle is None + assert agent._graphs.current is None agent.stream_handler.stop_streaming.assert_awaited_once() diff --git a/tests/test_agent_side_effect_boundaries.py b/tests/test_agent_side_effect_boundaries.py index 998eada72..570d55402 100644 --- a/tests/test_agent_side_effect_boundaries.py +++ b/tests/test_agent_side_effect_boundaries.py @@ -29,7 +29,7 @@ from app.agent.terminal.manager import _TerminalSessionManager from app.agent.terminal.ownership import current_terminal_scope from app.agent.terminal.session import _TerminalSession from app.agent.tools.base import MoviePilotTool -from app.agent.tools.catalog import ToolCatalogSnapshot +from tests.agent_graph import publish_graph pytestmark = pytest.mark.usefixtures("terminal_scope") @@ -327,12 +327,12 @@ async def test_agent_cleanup_closes_subagent_middlewares() -> None: """ closed.append(True) - agent._subagent_middlewares = (_Middleware(),) + bundle = publish_graph(agent, middlewares=(_Middleware(),)) assert await agent.cleanup() is True assert closed == [True] - assert agent._subagent_middlewares == () + assert bundle.resources.empty @pytest.mark.anyio @@ -343,22 +343,14 @@ async def test_agent_cache_replacement_closes_previous_subagent_middleware() -> agent = MoviePilotAgent(session_id="session-1", user_id="user-1") old_middleware = SimpleNamespace(close=AsyncMock()) new_middleware = SimpleNamespace(close=AsyncMock()) - catalog = ToolCatalogSnapshot.from_tools([], plugin_revision=0, factory_revision="factory-v1") - agent._subagent_middlewares = (old_middleware,) + publish_graph(agent, middlewares=(old_middleware,)) - await agent._cache_agent( - signature=("new",), - agent=object(), - streaming=False, - tool_catalog=catalog, - subagent_catalog=catalog, - mcp_config_signature="mcp-config", - subagent_middlewares=(new_middleware,), - ) + bundle = publish_graph(agent, middlewares=(new_middleware,)) + await agent._graphs.collect() old_middleware.close.assert_awaited_once() new_middleware.close.assert_not_awaited() - assert agent._subagent_middlewares == (new_middleware,) + assert tuple(bundle.resources.middlewares.values()) == (new_middleware,) @pytest.mark.anyio @@ -369,8 +361,7 @@ async def test_agent_execution_failure_closes_cached_subagent_middleware() -> No agent = MoviePilotAgent(session_id="session-1", user_id="user-1") middleware = SimpleNamespace(close=AsyncMock()) graph = SimpleNamespace(ainvoke=AsyncMock(side_effect=RuntimeError("failed"))) - agent._compiled_agent_bundle = SimpleNamespace(agent=graph, models=()) - agent._subagent_middlewares = (middleware,) + bundle = publish_graph(agent, graph=graph, middlewares=(middleware,)) agent._should_stream = lambda: False agent._create_agent = AsyncMock(return_value=graph) agent._dispatch_execution_notice = AsyncMock() @@ -380,8 +371,8 @@ async def test_agent_execution_failure_closes_cached_subagent_middleware() -> No assert result == "智能助手执行失败，请稍后重试" middleware.close.assert_awaited_once() - assert agent._compiled_agent_bundle is None - assert agent._subagent_middlewares == () + assert agent._graphs.current is None + assert bundle.resources.empty def test_subagent_control_middleware_close_is_idempotent() -> None: @@ -478,13 +469,13 @@ async def test_agent_cleanup_retains_nonconverged_subagent_middleware() -> None: seal=MagicMock(), close=AsyncMock(side_effect=[False, True]), ) - agent._subagent_middlewares = (middleware,) + bundle = publish_graph(agent, middlewares=(middleware,)) assert await agent.cleanup() is False - assert agent._subagent_middlewares == (middleware,) + assert tuple(bundle.resources.middlewares.values()) == (middleware,) assert await agent.cleanup() is True - assert agent._subagent_middlewares == () + assert bundle.resources.empty assert middleware.seal.call_count == 2 assert middleware.close.await_count == 2 diff --git a/tests/test_agent_summarization_streaming.py b/tests/test_agent_summarization_streaming.py index 981c0e27e..2bdfce5b5 100644 --- a/tests/test_agent_summarization_streaming.py +++ b/tests/test_agent_summarization_streaming.py @@ -40,6 +40,7 @@ from app.agent.middleware.summarization import ( FinalRequestCompactionMiddleware, ) from app.agent.middleware.usage import UsageMiddleware +from tests.agent_graph import publish_graph class _FakeLLM: @@ -1338,7 +1339,7 @@ def test_summary_failure_preserves_database_history(): user_id=user_id, memory=isolated_memory, ) - agent._compiled_agent_bundle = SimpleNamespace(models=()) + publish_graph(agent) agent._should_stream = lambda: False agent._create_agent = AsyncMock(return_value=_FailingGraph()) agent.stream_handler = SimpleNamespace(stop_streaming=AsyncMock(return_value=(False, ""))) @@ -1352,7 +1353,7 @@ def test_summary_failure_preserves_database_history(): isolated_memory.clear_memory(session_id, user_id) recovered_messages = isolated_memory.get_agent_messages(session_id, user_id) assert result == "智能助手执行失败，请稍后重试" - assert agent._compiled_agent_bundle is None + assert agent._graphs.current is None assert [message.content for message in recovered_messages] == ["数据库中的旧事实"] send_usage_event.assert_called_once() assert not send_usage_event.call_args.args[1].success diff --git a/tests/test_architecture_dependencies.py b/tests/test_architecture_dependencies.py index a05dba737..2d9a3ba74 100644 --- a/tests/test_architecture_dependencies.py +++ b/tests/test_architecture_dependencies.py @@ -374,6 +374,29 @@ def test_agent_manager_uses_focused_owner_modules_and_precise_facade(): assert "from app.agent.orchestrator import AgentManager" not in startup_source +def test_agent_graph_resources_have_one_owner() -> None: + """ + 图资源只归 graph 所有，执行对象不得重新建立平行资源集合或私有兼容导出。 + """ + graph_path = APP_ROOT / "agent" / "graph.py" + assert graph_path.is_file() + graph_tree = _parse_source(graph_path) + assert {node.name for node in graph_tree.body if isinstance(node, ast.ClassDef)} == { + "GraphResources", "CompiledAgentBundle", "AgentGraphCache", + } + orchestrator_path = APP_ROOT / "agent" / "orchestrator.py" + orchestrator = orchestrator_path.read_text(encoding="utf-8") + for retired in ("_compiled_agent_bundle", "_retired_llm_models", "self._subagent_middlewares", + "def _cache_agent", "def _release_llm_models", "class _CompiledAgentBundle"): + assert retired not in orchestrator + assert "from app.agent.graph import AgentGraphCache, CompiledAgentBundle" in orchestrator + assert not any( + isinstance(node, ast.ImportFrom) and node.module == "app.agent.orchestrator" + for node in ast.walk(graph_tree) + ) + assert "app/agent/graph.py" in (PROJECT_ROOT / "docs/rules/05-architecture.md").read_text(encoding="utf-8") + + def test_mcp_schema_projection_has_one_pure_owner() -> None: """ MCP schema 投影使用单词 owner，且不引入运行时或端点依赖。 diff --git a/tests/test_llm_client_lifecycle.py b/tests/test_llm_client_lifecycle.py index 1ea2a6f47..6b247528a 100644 --- a/tests/test_llm_client_lifecycle.py +++ b/tests/test_llm_client_lifecycle.py @@ -4,7 +4,6 @@ LLM 独占 HTTP 客户端必须随请求和 Agent 图确定性释放。 import asyncio import json -from datetime import datetime from types import SimpleNamespace from unittest.mock import AsyncMock, Mock @@ -14,8 +13,9 @@ import pytest from app.agent import orchestrator as agent_module from app.agent.llm import helper as llm_module from app.agent.llm.helper import LLMHelper, LLMTestError, LLMTestTimeout -from app.agent.orchestrator import MoviePilotAgent, _CompiledAgentBundle +from app.agent.orchestrator import MoviePilotAgent from app.agent.tools.catalog import ToolCatalogSnapshot +from tests.agent_graph import publish_graph @pytest.fixture @@ -37,16 +37,6 @@ def _model(*, failure=None): ) -def _bundle(model): - """ - 构造持有实际模型 owner 的已编译图快照。 - """ - return _CompiledAgentBundle( - signature=("old",), agent=object(), streaming=False, - created_at=datetime.now(), models=(model,), - ) - - @pytest.mark.anyio @pytest.mark.parametrize("failure", [None, RuntimeError("failed"), TimeoutError(), asyncio.CancelledError()]) async def test_settings_test_closes_clients_on_every_exit(monkeypatch, failure): @@ -91,13 +81,9 @@ async def test_graph_replacement_closes_only_retired_clients(): """ old_model, new_model = _model(), _model() agent = MoviePilotAgent(session_id="replace-clients", user_id="1") - agent._compiled_agent_bundle = _bundle(old_model) - catalog = ToolCatalogSnapshot.from_tools([], plugin_revision=0, factory_revision="test") - await agent._cache_agent( - signature=("new",), agent=object(), streaming=False, - tool_catalog=catalog, subagent_catalog=catalog, mcp_config_signature="", - models=(new_model,), - ) + publish_graph(agent, models=(old_model,)) + publish_graph(agent, models=(new_model,)) + await agent._graphs.collect() old_model.http_async_client.aclose.assert_awaited_once() new_model.http_async_client.aclose.assert_not_awaited() assert await agent._invalidate_cached_agent() is True @@ -113,8 +99,7 @@ async def test_unfinished_subagent_retains_model_until_cleanup_converges(): """ model = _model() agent = MoviePilotAgent(session_id="pending-clients", user_id="1") - agent._compiled_agent_bundle = _bundle(model) - agent._subagent_middlewares = (SimpleNamespace(close=AsyncMock(side_effect=[False, True])),) + publish_graph(agent, models=(model,), middlewares=(SimpleNamespace(close=AsyncMock(side_effect=[False, True])),)) assert await agent._invalidate_cached_agent() is False model.http_async_client.aclose.assert_not_awaited() assert await agent._invalidate_cached_agent() is True @@ -128,7 +113,7 @@ async def test_unfinished_learning_retains_model_until_task_finishes(): """ model = _model() agent = MoviePilotAgent(session_id="learning-clients", user_id="1") - agent._compiled_agent_bundle = _bundle(model) + publish_graph(agent, models=(model,)) task = asyncio.get_running_loop().create_future() agent._learning = SimpleNamespace(run=SimpleNamespace(task=task)) assert await agent._invalidate_cached_agent() is False @@ -146,7 +131,7 @@ async def test_close_failure_retains_owner_and_still_closes_other_client(): model = _model() model.http_async_client.aclose.side_effect = [RuntimeError("close failed"), None] agent = MoviePilotAgent(session_id="retry-clients", user_id="1") - agent._compiled_agent_bundle = _bundle(model) + publish_graph(agent, models=(model,)) assert await agent._invalidate_cached_agent() is False model.http_client.close.assert_called_once() assert await agent._invalidate_cached_agent() is True