diff --git a/app/agent/tools/impl/ask_user_choice.py b/app/agent/tools/impl/ask_user_choice.py index 4a82a6e8e..df3bf10b1 100644 --- a/app/agent/tools/impl/ask_user_choice.py +++ b/app/agent/tools/impl/ask_user_choice.py @@ -6,7 +6,7 @@ from pydantic import BaseModel, Field, model_validator from app.agent.tools.base import MoviePilotTool from app.agent.tools.tags import ToolTag -from app.application.messaging.agent import ( +from app.application.messaging.agent_interaction import ( AgentInteractionOption, agent_interaction_manager, build_agent_choice_callback, diff --git a/app/application/messaging/agent.py b/app/application/messaging/agent.py index bee76d4b0..e9cbeb4d5 100644 --- a/app/application/messaging/agent.py +++ b/app/application/messaging/agent.py @@ -7,7 +7,7 @@ import time import uuid from collections import deque from dataclasses import dataclass, field -from datetime import datetime, timedelta +from datetime import datetime from pathlib import Path from queue import Empty, Queue from threading import Lock @@ -16,12 +16,9 @@ from typing import ( AsyncIterator, Awaitable, Callable, - Dict, Iterable, - List, Optional, Protocol, - Tuple, Union, cast, ) @@ -40,6 +37,17 @@ from app.application.messaging.chat import ( get_configured_agent_chat_persistence, get_configured_agent_chat_service, ) +from app.application.messaging.agent_interaction import ( + AGENT_CHOICE_PREFIX, + LEGACY_AGENT_CHOICE_PREFIX, + AgentInteractionManager, + AgentInteractionOption, + PendingAgentInteraction, + agent_interaction_manager, + build_agent_choice_button_rows, + build_agent_choice_callback, + parse_agent_choice_callback, +) from app.application.messaging.router import has_pending_interaction from app.runtime.execution import run_in_threadpool from app.runtime.log import logger @@ -48,170 +56,6 @@ from app.schemas.message import Message from app.schemas.types import NotificationChannel __all__ = ["dispatch_command"] -# Agent 选择按钮回调前缀（新旧两种格式都必须继续兼容） -AGENT_CHOICE_PREFIX = "agent_interaction:choice:" -LEGACY_AGENT_CHOICE_PREFIX = "agent_choice:" - - -def build_agent_choice_callback(request_id: str, option_index: int) -> str: - """构造 Agent 选择按钮回调数据。""" - return f"{AGENT_CHOICE_PREFIX}{request_id}:{option_index}" - - -def parse_agent_choice_callback( - callback_data: str, -) -> Optional[Tuple[str, int]]: - """解析新旧两种 Agent 选择回调，格式无效时返回 None。""" - if callback_data.startswith(AGENT_CHOICE_PREFIX): - try: - _, _, request_id, option_index = callback_data.split(":", 3) - except ValueError: - return None - elif callback_data.startswith(LEGACY_AGENT_CHOICE_PREFIX): - # 兼容旧格式，避免已发送的按钮失效 - try: - _, request_id, option_index = callback_data.split(":", 2) - except ValueError: - return None - else: - return None - if not request_id or not option_index.isdigit(): - return None - return request_id, int(option_index) - - -def build_agent_choice_button_rows( - request: "PendingAgentInteraction", -) -> Tuple[List[dict[str, Any]], List[List[dict[str, Any]]]]: - """根据待选择请求构造 WebAgent 和消息渠道共用的按钮。""" - buttons = [ - { - "label": option.label, - "callback_data": build_agent_choice_callback(request.request_id, index), - "description": option.description or option.label, - } - for index, option in enumerate(request.options, start=1) - ] - button_rows = [[button] for button in buttons] - return buttons, button_rows - - -@dataclass(frozen=True) -class AgentInteractionOption: - """ - Agent 交互选项。 - """ - - label: str - value: str - description: Optional[str] = None - - -@dataclass -class PendingAgentInteraction: - """ - 待处理的 Agent 客户端交互请求。 - """ - - request_id: str - session_id: str - user_id: str - channel: Optional[str] - source: Optional[str] - username: Optional[str] - title: Optional[str] - prompt: str - options: List[AgentInteractionOption] - created_at: datetime = field(default_factory=datetime.now) - - -class AgentInteractionManager: - """ - 管理 Agent 发起的客户端交互请求。 - """ - - _ttl = timedelta(hours=24) - - def __init__(self): - """初始化待处理的 Agent 交互请求表。""" - self._pending_interactions: Dict[str, PendingAgentInteraction] = {} - self._lock = Lock() - - def _cleanup_locked(self) -> None: - """在持锁状态下移除过期 Agent 交互。""" - expire_before = datetime.now() - self._ttl - expired_ids = [ - request_id - for request_id, request in self._pending_interactions.items() - if request.created_at < expire_before - ] - for request_id in expired_ids: - self._pending_interactions.pop(request_id, None) - - def create_request( - self, - session_id: str, - user_id: str, - channel: Optional[str], - source: Optional[str], - username: Optional[str], - title: Optional[str], - prompt: str, - options: List[AgentInteractionOption], - ) -> PendingAgentInteraction: - """ - 创建一条待用户确认的 Agent 交互请求。 - """ - with self._lock: - self._cleanup_locked() - request_id = uuid.uuid4().hex[:12] - while request_id in self._pending_interactions: - request_id = uuid.uuid4().hex[:12] - request = PendingAgentInteraction( - request_id=request_id, - session_id=session_id, - user_id=str(user_id), - channel=channel, - source=source, - username=username, - title=title, - prompt=prompt, - options=options, - ) - self._pending_interactions[request_id] = request - return request - - def resolve( - self, - request_id: str, - option_index: int, - user_id: Optional[str] = None, - ) -> Optional[tuple[PendingAgentInteraction, AgentInteractionOption]]: - """ - 消费一条 Agent 交互请求，并返回选中的选项。 - """ - with self._lock: - self._cleanup_locked() - request = self._pending_interactions.get(request_id) - if not request: - return None - if user_id is not None and str(request.user_id) != str(user_id): - return None - if option_index < 1 or option_index > len(request.options): - return None - option = request.options[option_index - 1] - self._pending_interactions.pop(request_id, None) - return request, option - - def clear(self) -> None: - """ - 清空所有 Agent 交互请求。 - """ - with self._lock: - self._pending_interactions.clear() - - -agent_interaction_manager = AgentInteractionManager() _WEB_AGENT_EDIT_QUEUES: dict[str, list[Queue[dict[str, Any]]]] = {} diff --git a/app/application/messaging/agent_interaction.py b/app/application/messaging/agent_interaction.py new file mode 100644 index 000000000..253c5972e --- /dev/null +++ b/app/application/messaging/agent_interaction.py @@ -0,0 +1,160 @@ +"""Agent 按钮选择交互契约与进程内待处理请求管理。""" + +from dataclasses import dataclass, field +from datetime import datetime, timedelta +from threading import Lock +from typing import Any, Optional +import uuid + +# Agent 选择按钮回调前缀，新旧两种格式都必须继续兼容。 +AGENT_CHOICE_PREFIX = "agent_interaction:choice:" +LEGACY_AGENT_CHOICE_PREFIX = "agent_choice:" + + +def build_agent_choice_callback(request_id: str, option_index: int) -> str: + """构造 Agent 选择按钮回调数据。""" + return f"{AGENT_CHOICE_PREFIX}{request_id}:{option_index}" + + +def parse_agent_choice_callback( + callback_data: str, +) -> Optional[tuple[str, int]]: + """解析新旧两种 Agent 选择回调，格式无效时返回 None。""" + if callback_data.startswith(AGENT_CHOICE_PREFIX): + try: + _, _, request_id, option_index = callback_data.split(":", 3) + except ValueError: + return None + elif callback_data.startswith(LEGACY_AGENT_CHOICE_PREFIX): + # 兼容旧格式，避免已发送的按钮失效。 + try: + _, request_id, option_index = callback_data.split(":", 2) + except ValueError: + return None + else: + return None + if not request_id or not option_index.isdigit(): + return None + return request_id, int(option_index) + + +@dataclass(frozen=True) +class AgentInteractionOption: + """Agent 交互选项。""" + + label: str + value: str + description: Optional[str] = None + + +@dataclass +class PendingAgentInteraction: + """待处理的 Agent 客户端交互请求。""" + + request_id: str + session_id: str + user_id: str + channel: Optional[str] + source: Optional[str] + username: Optional[str] + title: Optional[str] + prompt: str + options: list[AgentInteractionOption] + created_at: datetime = field(default_factory=datetime.now) + + +def build_agent_choice_button_rows( + request: PendingAgentInteraction, +) -> tuple[list[dict[str, Any]], list[list[dict[str, Any]]]]: + """根据待选择请求构造 WebAgent 和消息渠道共用的按钮。""" + buttons = [ + { + "label": option.label, + "callback_data": build_agent_choice_callback(request.request_id, index), + "description": option.description or option.label, + } + for index, option in enumerate(request.options, start=1) + ] + button_rows = [[button] for button in buttons] + return buttons, button_rows + + +class AgentInteractionManager: + """管理 Agent 发起的客户端交互请求。""" + + _ttl = timedelta(hours=24) + + def __init__(self) -> None: + """初始化待处理的 Agent 交互请求表。""" + self._pending_interactions: dict[str, PendingAgentInteraction] = {} + self._lock = Lock() + + def _cleanup_locked(self) -> None: + """在持锁状态下移除过期 Agent 交互。""" + expire_before = datetime.now() - self._ttl + expired_ids = [ + request_id + for request_id, request in self._pending_interactions.items() + if request.created_at < expire_before + ] + for request_id in expired_ids: + self._pending_interactions.pop(request_id, None) + + def create_request( + self, + session_id: str, + user_id: str, + channel: Optional[str], + source: Optional[str], + username: Optional[str], + title: Optional[str], + prompt: str, + options: list[AgentInteractionOption], + ) -> PendingAgentInteraction: + """创建一条待用户确认的 Agent 交互请求。""" + with self._lock: + self._cleanup_locked() + request_id = uuid.uuid4().hex[:12] + while request_id in self._pending_interactions: + request_id = uuid.uuid4().hex[:12] + request = PendingAgentInteraction( + request_id=request_id, + session_id=session_id, + user_id=str(user_id), + channel=channel, + source=source, + username=username, + title=title, + prompt=prompt, + options=options, + ) + self._pending_interactions[request_id] = request + return request + + def resolve( + self, + request_id: str, + option_index: int, + user_id: Optional[str] = None, + ) -> Optional[tuple[PendingAgentInteraction, AgentInteractionOption]]: + """消费一条 Agent 交互请求，并返回选中的选项。""" + with self._lock: + self._cleanup_locked() + request = self._pending_interactions.get(request_id) + if not request: + return None + if user_id is not None and str(request.user_id) != str(user_id): + return None + if option_index < 1 or option_index > len(request.options): + return None + option = request.options[option_index - 1] + self._pending_interactions.pop(request_id, None) + return request, option + + def clear(self) -> None: + """清空所有 Agent 交互请求。""" + with self._lock: + self._pending_interactions.clear() + + +agent_interaction_manager = AgentInteractionManager() diff --git a/app/chain/message.py b/app/chain/message.py index 7eef7c879..aca25e40c 100644 --- a/app/chain/message.py +++ b/app/chain/message.py @@ -19,7 +19,10 @@ from app.application.agent import ( transcribe_audio, ) from app.application.messaging import router as interaction_router -from app.application.messaging.agent import agent_interaction_manager, parse_agent_choice_callback +from app.application.messaging.agent_interaction import ( + agent_interaction_manager, + parse_agent_choice_callback, +) from app.application.messaging.interaction import InteractionContext, InteractionDispatch from app.application.messaging.media import media_interaction_manager from app.application.messaging.plugin import PluginInputInteractionHandler diff --git a/tests/test_application_messaging_agent_architecture.py b/tests/test_application_messaging_agent_architecture.py new file mode 100644 index 000000000..0909c13c9 --- /dev/null +++ b/tests/test_application_messaging_agent_architecture.py @@ -0,0 +1,41 @@ +"""Agent 消息应用层职责拆分门禁。""" + +import ast +import importlib +from pathlib import Path + +PROJECT_ROOT = Path(__file__).parents[1] +MESSAGING_ROOT = PROJECT_ROOT / "app" / "application" / "messaging" +AGENT_FACADE_PATH = MESSAGING_ROOT / "agent.py" +AGENT_INTERACTION_PATH = MESSAGING_ROOT / "agent_interaction.py" + + +def _top_level_owners(path: Path) -> set[str]: + """返回模块顶层定义的类和函数名称。""" + tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) + return { + node.name + for node in tree.body + if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)) + } + + +def test_agent_interaction_has_one_canonical_owner() -> None: + """Agent 选择交互只能在独立模块实现，旧入口保留对象级兼容导出。""" + names = { + "AgentInteractionManager", + "AgentInteractionOption", + "PendingAgentInteraction", + "build_agent_choice_button_rows", + "build_agent_choice_callback", + "parse_agent_choice_callback", + } + facade = importlib.import_module("app.application.messaging.agent") + interaction = importlib.import_module( + "app.application.messaging.agent_interaction" + ) + + assert names <= _top_level_owners(AGENT_INTERACTION_PATH) + assert names.isdisjoint(_top_level_owners(AGENT_FACADE_PATH)) + for name in names | {"agent_interaction_manager"}: + assert getattr(facade, name) is getattr(interaction, name)