""" WebAgent 附件处理、展示快照与流式对话的应用编排。 """ import asyncio import copy import hashlib import mimetypes import shutil import time import uuid from collections import deque from dataclasses import dataclass, field, replace from datetime import datetime from functools import partial from pathlib import Path from queue import Empty, Queue from typing import ( Any, AsyncIterator, Awaitable, Callable, Coroutine, Optional, Protocol, Union, cast, ) import aiofiles from app.application import agent as agent_application from app.application.agent import is_audio_input_available, transcribe_audio from app.application.commands import dispatch_command, get_command, get_commands from app.application.configuration import get_api_runtime_config_snapshot from app.application.messaging.chat import ( AgentChatPersistenceService, AgentChatPrincipal, AgentChatRecord, AgentChatService, get_configured_agent_chat_persistence, get_configured_agent_chat_service, ) from app.application.messaging.interaction import agent as agent_interaction from app.application.messaging.router import has_pending_interaction from app.application.messaging.webagent import events as web_agent_events from app.runtime.execution import run_in_threadpool from app.runtime.log import logger from app.runtime.tasks import get_task_registry from app.schemas.message import Message from app.schemas.types import NotificationChannel __all__ = ["dispatch_command"] _WEB_AGENT_BACKGROUND_TASKS: set[asyncio.Task[object]] = set() def create_web_agent_background_task( coroutine: Coroutine[Any, Any, object], ) -> asyncio.Task[object]: """ 登记 Web Agent 后台任务，使应用关闭时可以统一收口。 """ task = cast( asyncio.Task[object], get_task_registry().create( coroutine, owner="api.agent.web_execution", ), ) _WEB_AGENT_BACKGROUND_TASKS.add(task) task.add_done_callback(_WEB_AGENT_BACKGROUND_TASKS.discard) return task async def shutdown_web_agent_background_tasks() -> None: """ 取消并等待 Web Agent 后台任务，避免关闭数据库后仍提交快照。 """ tasks = tuple(_WEB_AGENT_BACKGROUND_TASKS) for task in tasks: task.cancel() if tasks: # asyncio.wait 不会因关闭阶段自身被取消而再次取消这些任务；仍在收尾的 # Agent 任务会保留在注册表中，直到自己的数据库操作取得确定终态。 await asyncio.wait(tasks) async def wait_web_agent_background_tasks() -> None: """ 等待已登记的 Web Agent 任务完成最终收尾。 """ tasks = tuple(_WEB_AGENT_BACKGROUND_TASKS) if tasks: await asyncio.gather(*tasks, return_exceptions=True) WEB_AGENT_SESSION_PREFIX = "web-agent:" WEB_AGENT_SOURCE = "web-agent" WEB_AGENT_FILE_TTL_SECONDS = 6 * 60 * 60 WEB_AGENT_FILE_MAX_ITEMS = 256 WEB_AGENT_UPLOAD_MAX_BYTES = 32 * 1024 * 1024 WEB_AGENT_UPLOAD_CHUNK_SIZE = 1024 * 1024 WEB_AGENT_AUDIO_CONVERSION_TIMEOUT_SECONDS = 60.0 WEB_AGENT_BROWSER_AUDIO_SUFFIXES = {".aac", ".m4a", ".mp3", ".mp4", ".wav", ".wave"} WEB_AGENT_TRADITIONAL_IDLE_TIMEOUT_SECONDS = 2.0 WEB_AGENT_TRADITIONAL_MAX_WAIT_SECONDS = 60.0 WEB_AGENT_STREAM_HEARTBEAT_SECONDS = 15.0 WEB_AGENT_STREAM_COALESCE_SECONDS = 0.03 WEB_AGENT_STREAM_COALESCE_MAX_CHARS = 256 WEB_AGENT_STREAM_QUEUE_MAX_SIZE = 64 _WEB_AGENT_FILE_REGISTRY: dict[str, dict[str, Any]] = {} _web_agent_message_handler: Optional[Callable[..., object]] = None _web_agent_session_binder: Optional[Callable[[str, str], None]] = None def configure_web_agent_message_runtime( *, message_handler: Callable[..., object], session_binder: Callable[[str, str], None], ) -> None: """ 由启动组合根装配传统消息执行与会话绑定端口。 """ global _web_agent_message_handler, _web_agent_session_binder _web_agent_message_handler = message_handler _web_agent_session_binder = session_binder def reset_web_agent_message_runtime() -> None: """ 清除 WebAgent 消息运行端口，避免 lifespan 重启复用旧对象。 """ global _web_agent_message_handler, _web_agent_session_binder _web_agent_message_handler = None _web_agent_session_binder = None def _handle_web_agent_message(**kwargs: Any) -> object: """ 调用已装配的传统消息入口。 """ if _web_agent_message_handler is None: raise RuntimeError("WebAgent 传统消息入口尚未配置") return _web_agent_message_handler(**kwargs) def bind_web_agent_user_session(user_id: str, session_id: str) -> None: """ 把 Web 用户绑定到服务端 Agent 会话。 """ if _web_agent_session_binder is None: raise RuntimeError("WebAgent 会话绑定端口尚未配置") _web_agent_session_binder(user_id, session_id) class WebAgentUpload(Protocol): """ WebAgent 上传源所需的最小异步读取合同。 """ async def read(self, size: int = -1) -> bytes: """ 读取下一段上传内容。 :param size: 最多读取的字节数，-1 表示读取剩余全部内容 :return: 读取的下一段字节，到达文件末尾时返回空字节串 """ ... async def close(self) -> None: """ 关闭上传源。 """ ... class WebAgentUploadTooLargeError(ValueError): """ 上传附件超过 WebAgent 单文件容量限制。 """ @dataclass(frozen=True, slots=True) class WebAgentStreamCommand: """ WebAgent 流式对话用例所需的传输无关输入。 :param text: 提交给 Agent 的用户输入正文 :param display_text: 用户界面展示的正文，None 或空字符串时使用输入正文 :param session_id: 客户端指定的会话标识，None 时由用例解析新会话 :param images: 提供给模型的图片引用列表 :param audio_refs: 上传音频的附件引用列表 :param files: 其他上传附件的描述字典列表 :param choice_selection: 用户对待处理选择请求的提交数据，None 表示普通消息 :param original_message_id: 来源消息标识，供渠道回执或回复关联使用 :param original_chat_id: 来源聊天标识，无外部渠道上下文时为 None :param echo_user: 是否把用户消息写入展示流，默认开启 """ text: str = "" display_text: Optional[str] = None session_id: Optional[str] = None images: list[str] = field(default_factory=list) audio_refs: list[str] = field(default_factory=list) files: list[dict[str, Any]] = field(default_factory=list) choice_selection: Optional[dict[str, Any]] = None original_message_id: Optional[Union[str, int]] = None original_chat_id: Optional[Union[str, int]] = None echo_user: bool = True @dataclass(frozen=True, slots=True) class WebAgentStreamResult: """ WebAgent 流式用例返回的事件迭代器与传输控制标记。 :param events: 按序产生 WebAgent 展示事件的异步迭代器 :param control: 需要传输层识别的特殊控制标记，None 表示普通事件流 """ events: AsyncIterator[dict[str, Any]] control: Optional[str] = None def build_web_agent_display_message( role: str, content: str = "", attachments: Optional[list[dict[str, Any]]] = None, status: str = "done", ) -> dict[str, Any]: """ 构造 WebAgent 前后端共享的展示消息。 """ normalized_content = content or "" return { "id": f"{role}-{uuid.uuid4().hex}", "role": role, "content": normalized_content, "createdAt": int(datetime.now().timestamp() * 1000), "status": status, "thinking": False, "thinking_started_at": None, "thinking_elapsed_ms": 0, "tools": [], "segments": ([{"type": "text", "content": normalized_content}] if normalized_content else []), "attachments": attachments or [], "choices": [], } class WebAgentEventPublisher: """ 合并 WebAgent 文本增量，并通过有界队列向 SSE 消费者提供事件。 """ def __init__(self) -> None: """ 初始化有界事件队列、增量缓冲和单一异步推送任务。 """ self._queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue(maxsize=WEB_AGENT_STREAM_QUEUE_MAX_SIZE) self._pending_events: deque[dict[str, Any]] = deque() self._pending_signal = asyncio.Event() self._pending_delta = "" self._delta_timer: Optional[asyncio.TimerHandle] = None self._disposed = False self._max_depth = 0 self._last_logged_depth = 0 self._pump_task = asyncio.create_task(self._pump()) @property def max_depth(self) -> int: """ 返回本轮发布器观测到的最大积压深度。 """ return self._max_depth def publish(self, event: dict[str, Any]) -> bool: """ 发布事件；返回关闭状态以便受保护投递能准确报告失败。 """ if self._disposed: return False if event.get("type") == "delta": self._pending_delta += str(event.get("content") or "") if len(self._pending_delta) >= WEB_AGENT_STREAM_COALESCE_MAX_CHARS: self._flush_delta() elif self._delta_timer is None: loop = asyncio.get_running_loop() self._delta_timer = loop.call_later( WEB_AGENT_STREAM_COALESCE_SECONDS, self._flush_delta, ) return True self._flush_delta() self._append_event(event) return True async def get(self) -> dict[str, Any]: """ 等待并返回下一条已排序事件。 """ return await self._queue.get() async def aclose(self) -> None: """ 停止发布器并释放等待中的泵任务。 """ if self._disposed: return self._disposed = True self._cancel_delta_timer() self._pending_delta = "" self._pending_events.clear() self._pump_task.cancel() try: await self._pump_task except asyncio.CancelledError: pass def _cancel_delta_timer(self) -> None: """ 取消尚未触发的文本合并计时器。 """ if self._delta_timer is None: return self._delta_timer.cancel() self._delta_timer = None def _flush_delta(self) -> None: """ 把当前文本缓冲转换成一条增量事件。 """ self._cancel_delta_timer() if not self._pending_delta or self._disposed: return content = self._pending_delta self._pending_delta = "" self._append_event({"type": "delta", "content": content}) def _append_event(self, event: dict[str, Any]) -> None: """ 追加待发布事件，相邻文本在出口阻塞时继续合并。 """ if event.get("type") == "delta" and self._pending_events and self._pending_events[-1].get("type") == "delta": self._pending_events[-1]["content"] += str(event.get("content") or "") else: self._pending_events.append(event) self._pending_signal.set() depth = self._queue.qsize() + len(self._pending_events) self._max_depth = max(self._max_depth, depth) if depth >= WEB_AGENT_STREAM_QUEUE_MAX_SIZE // 2 and depth > self._last_logged_depth: self._last_logged_depth = depth logger.debug(f"WebAgent SSE事件积压深度: {depth}") async def _pump(self) -> None: """ 按发布顺序把本地合并结果写入有界出口队列。 """ while True: await self._pending_signal.wait() while self._pending_events: event = self._pending_events.popleft() await self._queue.put(event) self._pending_signal.clear() def build_web_agent_session_id(user: AgentChatPrincipal, session_id: Optional[str]) -> str: """ 构建前端 Agent 会话 ID。 :param user: 当前登录用户 :param session_id: 前端传入的会话标识 :return: 可用于 Agent 记忆隔离的服务端会话 ID """ seed = str(session_id or "").strip() or uuid.uuid4().hex existing_chat = get_configured_agent_chat_service().get_sync(seed) if existing_chat: if not can_access_agent_chat(existing_chat, user): raise PermissionError("会话不存在或无权访问") return seed return _new_web_agent_session_id(user, seed) def _new_web_agent_session_id(user: AgentChatPrincipal, seed: str) -> str: """ 未落库的会话也按用户 ID 分区，上传后可复用同一 ID 而不能占用他人队列。 """ namespace = f"{WEB_AGENT_SESSION_PREFIX}{user.id}:" if seed.startswith(WEB_AGENT_SESSION_PREFIX): suffix = seed.removeprefix(namespace) if seed.startswith(namespace) and len(suffix) == 32 and all(char in "0123456789abcdef" for char in suffix): return seed raise PermissionError("会话不存在或无权访问") digest = hashlib.sha256(f"{user.id}:{seed}".encode("utf-8")).hexdigest() return f"{namespace}{digest[:32]}" async def build_web_agent_session_id_async( user: AgentChatPrincipal, session_id: Optional[str], service: Optional[AgentChatService] = None, ) -> str: """ 异步解析会话并验证已有记录归属，Web 前缀不提供访问授权。 """ seed = str(session_id or "").strip() or uuid.uuid4().hex if service is None: service = get_configured_agent_chat_service() existing_chat = await service.get(seed) if existing_chat: if not can_access_agent_chat(existing_chat, user): raise PermissionError("会话不存在或无权访问") return seed return _new_web_agent_session_id(user, seed) def can_access_agent_chat(chat: Any, user: AgentChatPrincipal) -> bool: """ 判断当前登录用户是否可以访问指定 Agent 会话。 超级用户可查看所有渠道历史；普通用户优先按持久化 ID 匹配，只有无 ID 旧记录使用用户名。 """ if not chat or not user: return False return AgentChatService.can_access(chat, user) async def get_accessible_agent_chat( service: AgentChatService, session_id: str, user: AgentChatPrincipal, ) -> Optional[AgentChatRecord]: """ 读取当前用户可访问的 Agent 会话。 """ return await service.get_accessible(session_id, user) def append_web_agent_text_segment(assistant_message: dict[str, Any], content: str) -> None: """ 将文本增量追加到展示消息，并仅合并相邻文本片段。 :param assistant_message: 当前助手展示消息 :param content: 新增文本 """ if not content: return assistant_message["content"] = str(assistant_message.get("content") or "") + content segments = assistant_message.setdefault("segments", []) if segments and segments[-1].get("type") == "text": segments[-1]["content"] = str(segments[-1].get("content") or "") + content else: segments.append({"type": "text", "content": content}) def build_legacy_web_agent_segments(content: str, tools: list[dict[str, Any]]) -> list[dict[str, Any]]: """ 为未携带有序片段的旧展示消息生成兼容布局。 :param content: 聚合后的助手文本 :param tools: 工具提示列表 :return: 按旧版工具在前、文本在后的顺序生成的片段 """ segments = [{"type": "tool", "toolIndex": index} for index in range(len(tools))] if content: segments.append({"type": "text", "content": content}) return segments def _apply_web_agent_tool_event(event: dict[str, Any], assistant_message: dict[str, Any]) -> None: """ 将工具生命周期事件写入 WebAgent 展示消息及其有序片段。 """ tool_id = str(event.get("tool_id") or event.get("tool_call_id") or "") tool_status = str(event.get("status") or "running") if tool_id and tool_status in {"running", "done", "error"}: matching_tool = next( (tool for tool in assistant_message["tools"] if str(tool.get("id") or "") == tool_id), None, ) if tool_status == "running": if matching_tool is None: tool_index = len(assistant_message["tools"]) assistant_message["tools"].append( { "id": tool_id, "tool_name": str(event.get("tool_name") or ""), "message": str(event.get("message") or "").strip(), "status": "running", } ) assistant_message.setdefault("segments", []).append({"type": "tool", "toolIndex": tool_index}) return if matching_tool is not None: matching_tool["status"] = tool_status return # 旧提示没有调用 ID，只收口上一条旧提示，保留结构化并行工具的实际状态。 for tool in assistant_message["tools"]: if tool.get("status") == "running" and tool.get("tool_name") is None: tool["status"] = "done" tool_index = len(assistant_message["tools"]) assistant_message["tools"].append( { "id": f"tool-{uuid.uuid4().hex}", "message": str(event.get("message") or "").strip(), "status": "running", } ) assistant_message.setdefault("segments", []).append({"type": "tool", "toolIndex": tool_index}) def _apply_web_agent_thinking_event(event: dict[str, Any], assistant_message: dict[str, Any]) -> None: """ 将模型思考状态写入 WebAgent 展示消息，供前端显示耗时。 """ is_thinking = str(event.get("status") or "") == "running" assistant_message["thinking"] = is_thinking if is_thinking: assistant_message["thinking_started_at"] = event.get("started_at") assistant_message["thinking_elapsed_ms"] = 0 else: assistant_message["thinking_elapsed_ms"] = ( event.get("elapsed_ms") or assistant_message.get("thinking_elapsed_ms") or 0 ) def apply_web_agent_display_event(event: dict[str, Any], assistant_message: dict[str, Any]) -> None: """ 将 WebAgent SSE 事件同步应用到服务端展示消息快照。 """ event_type = event.get("type") if event_type == "delta": append_web_agent_text_segment(assistant_message, event.get("content") or "") elif event_type == "tool": _apply_web_agent_tool_event(event, assistant_message) elif event_type == "attachment" and event.get("attachment"): assistant_message["attachments"].append(event["attachment"]) elif event_type == "choice" and event.get("choice"): assistant_message["choices"].append({**event["choice"], "status": "pending"}) elif event_type == "message_update": target_message = event.get("target_message") or {} assistant_message["id"] = target_message.get("id") or assistant_message.get("id") assistant_message["content"] = target_message.get("content") or "" assistant_message["attachments"] = target_message.get("attachments") or [] assistant_message["choices"] = target_message.get("choices") or [] assistant_message["tools"] = target_message.get("tools") or [] target_segments = target_message.get("segments") assistant_message["segments"] = ( target_segments if isinstance(target_segments, list) else build_legacy_web_agent_segments(assistant_message["content"], assistant_message["tools"]) ) assistant_message["status"] = target_message.get("status") or "done" elif event_type == "thinking": _apply_web_agent_thinking_event(event, assistant_message) elif event_type == "error": assistant_message["status"] = "error" if not assistant_message["content"]: append_web_agent_text_segment( assistant_message, event.get("message") or "智能助手响应失败", ) for tool in assistant_message["tools"]: if tool.get("status") == "running": tool["status"] = "error" elif event_type == "done": assistant_message["thinking"] = False if assistant_message.get("status") != "error": assistant_message["status"] = "done" for tool in assistant_message["tools"]: if tool.get("status") == "running": tool["status"] = "done" if tool.get("tool_name") is None else "error" async def save_web_agent_display_snapshot( *, session_id: str, current_user: AgentChatPrincipal, messages: list[dict[str, Any]], client_session_id: Optional[str] = None, service: Optional[AgentChatService] = None, persistence: Optional[AgentChatPersistenceService] = None, append: bool = False, ) -> None: """ 保存 WebAgent 展示快照前重新检查归属，避免入口之后的会话变更造成越权覆盖。 :param append: 后台单轮结果为True，按消息ID追加合并；前端完整快照保留替换合同 """ if service is None: # 直接调用该内部 helper 时没有 FastAPI 依赖注入上下文。 service = get_configured_agent_chat_service() existing_chat = await service.get(session_id) if existing_chat and not can_access_agent_chat(existing_chat, current_user): raise PermissionError("会话不存在或无权访问") if persistence is None: # 直接调用该内部 helper 时没有 FastAPI 依赖注入上下文。 persistence = get_configured_agent_chat_persistence() save_messages = persistence.async_append_display_messages if append else persistence.async_save_display_messages await save_messages( session_id=session_id, user_id=(existing_chat.user_id if existing_chat else str(current_user.id)), username=(existing_chat.username if existing_chat else current_user.name), channel=(existing_chat.channel if existing_chat and existing_chat.channel else NotificationChannel.WebAgent), source=(existing_chat.source if existing_chat and existing_chat.source else WEB_AGENT_SOURCE), original_chat_id=existing_chat.original_chat_id if existing_chat else None, client_session_id=( existing_chat.client_session_id if existing_chat and existing_chat.client_session_id else client_session_id ), messages=messages, ) def sanitize_web_agent_upload_name(filename: Optional[str], mime_type: Optional[str] = None) -> str: """ 规范化 Web Agent 上传文件名，避免路径穿越和空文件名。 :param filename: 浏览器上传的原始文件名 :param mime_type: 浏览器上报的 MIME 类型 :return: 可安全落盘的文件名 """ name = Path(filename or "attachment").name.strip() safe_name = "".join(char for char in name if char.isalnum() or char in (" ", ".", "_", "-")).strip(" .") if not safe_name: safe_name = "attachment" if "." not in safe_name: suffix = mimetypes.guess_extension(mime_type or "") or "" safe_name = f"{safe_name}{suffix}" return safe_name async def get_web_agent_upload_dir( user: AgentChatPrincipal, session_id: Optional[str], service: Optional[AgentChatService] = None, ) -> Path: """ 计算当前 Web Agent 会话的临时附件目录。 :param user: 当前登录用户 :param session_id: 前端会话标识 :return: 已创建的临时附件目录 """ server_session_id = await build_web_agent_session_id_async( user, session_id, service, ) safe_session_id = server_session_id.replace(":", "_") upload_dir = Path(get_api_runtime_config_snapshot().temp_path) / "agent_uploads" / safe_session_id await run_in_threadpool(upload_dir.mkdir, parents=True, exist_ok=True) return upload_dir async def save_web_agent_upload(upload_file: WebAgentUpload, target_path: Path) -> int: """ 分块保存 Web Agent 上传文件，并限制单文件体积。 :param upload_file: FastAPI 上传文件对象 :param target_path: 目标落盘路径 :return: 已写入的字节数 """ size = 0 try: async with aiofiles.open(target_path, "wb") as output: while True: chunk = await upload_file.read(WEB_AGENT_UPLOAD_CHUNK_SIZE) if not chunk: break size += len(chunk) if size > WEB_AGENT_UPLOAD_MAX_BYTES: raise WebAgentUploadTooLargeError("附件超过 32MB，无法发送给智能助手") await output.write(chunk) except Exception: await run_in_threadpool(target_path.unlink, missing_ok=True) raise finally: await upload_file.close() return size def cleanup_web_agent_file_registry() -> None: """ 清理过期或过量的 Web Agent 临时附件引用。 """ now = time.time() expired_ids = [ file_id for file_id, info in _WEB_AGENT_FILE_REGISTRY.items() if now - info.get("created_at", now) > WEB_AGENT_FILE_TTL_SECONDS ] for file_id in expired_ids: _WEB_AGENT_FILE_REGISTRY.pop(file_id, None) overflow = len(_WEB_AGENT_FILE_REGISTRY) - WEB_AGENT_FILE_MAX_ITEMS if overflow <= 0: return sorted_items = sorted( _WEB_AGENT_FILE_REGISTRY.items(), key=lambda item: item[1].get("created_at", 0), ) for file_id, _ in sorted_items[:overflow]: _WEB_AGENT_FILE_REGISTRY.pop(file_id, None) def guess_web_agent_attachment_kind(mime_type: Optional[str], fallback: str = "file") -> str: """ 根据 MIME 类型推断前端附件展示方式。 :param mime_type: 文件 MIME 类型 :param fallback: 无法推断时使用的类型 :return: image、audio 或 file """ if mime_type and mime_type.startswith("image/"): return "image" if mime_type and mime_type.startswith("audio/"): return "audio" return fallback def build_web_agent_url_attachment( url: str, kind: str, name: Optional[str] = None, mime_type: Optional[str] = None, ) -> dict[str, Any]: """ 构建远程或 data URL 附件事件。 :param url: 前端可访问的附件地址 :param kind: 附件展示类型 :param name: 展示名称 :param mime_type: MIME 类型 :return: 前端附件描述 """ return { "kind": kind, "url": url, "download_url": url, "name": name, "mime_type": mime_type, } def build_web_agent_input_attachments( images: list[str], files: list[dict[str, Any]], audio_refs: list[str], ) -> list[dict[str, Any]]: """ 构造 WebAgent 用户输入附件展示记录。 """ attachments: list[dict[str, Any]] = [] for index, image in enumerate(images or [], start=1): attachments.append( { "kind": "image", "url": image, "download_url": image, "name": f"image-{index}", "mime_type": "image/*", } ) for index, file in enumerate(files or [], start=1): ref = file.get("ref") or file.get("url") or file.get("local_path") or "" mime_type = file.get("mime_type") attachments.append( { "kind": guess_web_agent_attachment_kind(mime_type), "url": ref, "download_url": ref, "name": file.get("name") or f"attachment-{index}", "mime_type": mime_type, "size": file.get("size"), "local_path": file.get("local_path"), } ) for index, audio_ref in enumerate(audio_refs or [], start=1): attachments.append( { "kind": "audio", "url": audio_ref, "download_url": audio_ref, "name": f"voice-{index}", "mime_type": "audio/*", } ) return attachments def register_web_agent_file( file_path: Optional[str], file_name: Optional[str] = None, kind: Optional[str] = None, mime_type: Optional[str] = None, *, owner_id: str, ) -> Optional[dict[str, Any]]: """ 注册 Web Agent 本地附件并返回前端可访问的短期下载地址。 :param file_path: 本地文件路径 :param file_name: 前端展示文件名 :param kind: 附件展示类型 :param mime_type: 已知 MIME 类型 :param owner_id: 创建附件的用户 ID :return: 前端附件描述，文件不可访问时返回 None """ normalized_owner_id = str(owner_id).strip() if not file_path or not normalized_owner_id: return None try: resolved_path = Path(file_path).expanduser().resolve(strict=True) except OSError: return None if not resolved_path.is_file(): return None cleanup_web_agent_file_registry() file_id = uuid.uuid4().hex display_name = file_name or resolved_path.name resolved_mime_type = mime_type or mimetypes.guess_type(display_name or str(resolved_path))[0] file_url = f"message/agent/file/{file_id}" _WEB_AGENT_FILE_REGISTRY[file_id] = { "path": resolved_path, "name": display_name, "mime_type": resolved_mime_type or "application/octet-stream", "owner_id": normalized_owner_id, "created_at": time.time(), } return { "kind": kind or guess_web_agent_attachment_kind(resolved_mime_type), "url": file_url, "download_url": file_url, "name": display_name, "mime_type": resolved_mime_type, "size": resolved_path.stat().st_size, } def get_web_agent_audio_mime_type(audio_path: Path) -> Optional[str]: """ 生成浏览器播放更友好的音频 MIME 类型。 :param audio_path: 音频文件路径 :return: 可用于 FileResponse/audio 标签的 MIME 类型 """ suffix = audio_path.suffix.lower() if suffix in {".wav", ".wave"}: return "audio/wav" if suffix == ".mp3": return "audio/mpeg" if suffix in {".m4a", ".mp4"}: return "audio/mp4" if suffix == ".aac": return "audio/aac" return mimetypes.guess_type(audio_path.name)[0] def resolve_web_agent_audio_source_path(voice_path: str) -> Path: """ 解析语音源文件；文件暂时不可用时保留原始路径以维持回退语义。 """ try: return Path(voice_path).expanduser().resolve(strict=True) except OSError: return Path(voice_path) def remove_web_agent_audio_output(path: Path) -> None: """ 清理未完成的转码产物。 """ try: path.unlink() except FileNotFoundError: pass except OSError as err: logger.debug("WebAgent 清理未完成语音转码产物失败: path=%s, error=%s", path, err) async def terminate_web_agent_audio_process( process: Optional[asyncio.subprocess.Process], ) -> None: """ 终止并回收转码进程，避免取消或超时留下孤儿 ffmpeg。 """ if process is None or process.returncode is not None: return try: process.kill() except ProcessLookupError: return try: await process.communicate() except (OSError, ProcessLookupError): pass async def prepare_web_agent_audio_attachment_path_async(voice_path: str) -> Path: """ 异步准备 WebAgent 语音附件，转码等待可取消且有超时。 """ source_path = Path( await run_in_threadpool( resolve_web_agent_audio_source_path, voice_path, ) ) if source_path.suffix.lower() in WEB_AGENT_BROWSER_AUDIO_SUFFIXES: return source_path ffmpeg_path = await run_in_threadpool(shutil.which, "ffmpeg") if not ffmpeg_path: logger.warning("WebAgent 语音转 WAV 跳过：ffmpeg 不可用，path=%s", source_path) return source_path voice_dir = Path(get_api_runtime_config_snapshot().temp_path) / "voice" try: await run_in_threadpool(voice_dir.mkdir, parents=True, exist_ok=True) except OSError as err: logger.warning("WebAgent 语音转 WAV 目录不可用，将回退原文件: path=%s, error=%s", voice_dir, err) return source_path output_path = voice_dir / f"{source_path.stem}_web_{uuid.uuid4().hex[:8]}.wav" process: Optional[asyncio.subprocess.Process] = None async def cleanup_conversion_output() -> None: """ 清理失败或取消的转码进程及临时产物。 """ await terminate_web_agent_audio_process(process) await run_in_threadpool(remove_web_agent_audio_output, output_path) try: process = await asyncio.create_subprocess_exec( ffmpeg_path, "-y", "-i", str(source_path), "-ar", "24000", "-ac", "1", "-f", "wav", str(output_path), stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.PIPE, ) _, stderr = await asyncio.wait_for( process.communicate(), timeout=WEB_AGENT_AUDIO_CONVERSION_TIMEOUT_SECONDS, ) output_exists = await run_in_threadpool(output_path.exists) if process.returncode != 0 or not output_exists: await cleanup_conversion_output() logger.warning( "WebAgent 语音转 WAV 失败，将回退原文件: returncode=%s, stderr=%s", process.returncode, (stderr or b"").decode("utf-8", errors="replace").strip()[:500], ) return source_path return Path(output_path) except asyncio.TimeoutError: await cleanup_conversion_output() logger.warning( "WebAgent 语音转 WAV 超时，将回退原文件: timeout=%ss, path=%s", WEB_AGENT_AUDIO_CONVERSION_TIMEOUT_SECONDS, source_path, ) return source_path except asyncio.CancelledError: await cleanup_conversion_output() logger.warning("WebAgent 语音转 WAV 已取消，path=%s", source_path) raise except OSError as err: await cleanup_conversion_output() logger.warning("WebAgent 语音转 WAV 启动失败，将回退原文件: path=%s, error=%s", source_path, err) return source_path def get_web_agent_registered_file(ref: str, *, owner_id: str) -> Optional[dict[str, Any]]: """ 根据前端附件引用读取当前用户登记的 WebAgent 临时文件。 :param ref: message/agent/file/{file_id} 形式的短期引用 :param owner_id: 请求读取附件的用户 ID :return: 文件登记信息，引用无效、过期或不属于当前用户时返回 None """ normalized_owner_id = str(owner_id).strip() normalized_ref = (ref or "").strip() prefix = "message/agent/file/" if not normalized_owner_id or not normalized_ref.startswith(prefix): return None cleanup_web_agent_file_registry() file_id = normalized_ref[len(prefix) :].split("/", 1)[0] file_info = _WEB_AGENT_FILE_REGISTRY.get(file_id) if not file_info or file_info.get("owner_id") != normalized_owner_id: return None file_path = Path(file_info["path"]) if not file_path.exists() or not file_path.is_file(): _WEB_AGENT_FILE_REGISTRY.pop(file_id, None) return None return file_info def resolve_web_agent_audio_refs( audio_refs: list[str], *, owner_id: str, ) -> list[tuple[str, Path, str]]: """ 在调用协程中解析当前用户的音频引用，返回不依赖登记表的文件快照。 :param audio_refs: 前端提交的临时音频引用 :param owner_id: 请求转写音频的用户 ID :return: 已解析的引用、文件路径和文件名 """ audio_files = [] for audio_ref in audio_refs: file_info = get_web_agent_registered_file(audio_ref, owner_id=owner_id) if not file_info: logger.warning("WebAgent 语音引用不存在或已过期: ref=%s", audio_ref) continue file_path = Path(file_info["path"]) audio_files.append((audio_ref, file_path, file_info.get("name") or file_path.name)) return audio_files def transcribe_web_agent_audio_files( audio_files: list[tuple[str, Path, str]], ) -> Optional[str]: """ 转写 WebAgent 上传的本地录音附件。 文件信息已在调用协程中从短期登记表解析，阻塞文件读取和 provider 调用可在 worker 中执行，避免跨线程访问登记表。 """ if not audio_files: return None if not is_audio_input_available(): logger.warning("WebAgent 音频输入能力未配置或未启用，跳过语音识别") return None transcripts = [] for audio_ref, file_path, file_name in audio_files: try: content = file_path.read_bytes() except OSError as err: logger.warning("WebAgent 语音文件读取失败: ref=%s, error=%s", audio_ref, err) continue transcript = transcribe_audio( content=content, filename=file_name, ) if transcript: transcripts.append(transcript) return "\n".join(transcripts).strip() if transcripts else None async def transcribe_web_agent_audio_input( audio_refs: list[str], *, owner_id: str, ) -> Optional[str]: """ 解析并在线程池中转写当前用户的 WebAgent 音频引用。 :param audio_refs: 前端提交的临时音频引用 :param owner_id: 请求转写音频的用户 ID :return: 合并后的转写文本，没有可转写内容时返回 None """ audio_files = resolve_web_agent_audio_refs(audio_refs, owner_id=owner_id) if not audio_files: return None return await asyncio.to_thread(transcribe_web_agent_audio_files, audio_files) def merge_web_agent_prompt_with_transcript(prompt: str, transcript: Optional[str]) -> str: """ 合并用户输入文本和语音转写文本，避免重复发送相同内容。 """ merged_parts = [] seen_parts = set() for item in (prompt, transcript or ""): normalized = item.strip() if not normalized or normalized in seen_parts: continue seen_parts.add(normalized) merged_parts.append(normalized) return "\n".join(merged_parts).strip() def build_web_agent_choice_event(message: Message) -> Optional[dict[str, Any]]: """ 将带按钮通知转换为 Web Agent 选择卡片事件。 :param message: Agent 工具发出的按钮通知 :return: 选择卡片事件，按钮为空时返回 None """ button_rows = web_agent_events.normalize_web_agent_button_rows(message.buttons) buttons = [button for row in button_rows for button in row] if not buttons: return None choice_id = None parsed = agent_interaction.parse_agent_choice_callback(buttons[0]["callback_data"]) if parsed: choice_id = parsed[0] return { "type": "choice", "choice": { "id": choice_id or uuid.uuid4().hex, "title": message.title, "prompt": message.text or "", "buttons": buttons, "button_rows": button_rows, }, } def resolve_web_agent_choice_payload(callback_data: str, user_id: str) -> Optional[dict[str, Any]]: """ 解析并消费 Web Agent 按钮选择，生成前端反馈与下一条用户消息。 :param callback_data: 前端点击的按钮回调数据 :param user_id: 当前登录用户 ID :return: 可返回给前端的数据，选择无效时返回 None """ parsed = agent_interaction.parse_agent_choice_callback(callback_data) if not parsed: return None request_id, option_index = parsed resolved = agent_interaction.agent_interaction_manager.resolve_web_choice( request_id=request_id, option_index=option_index, user_id=str(user_id), ) if not resolved: return None request, option, already_selected = resolved buttons, button_rows = agent_interaction.build_agent_choice_button_rows(request) selected_description = option.description or option.label return { "already_selected": already_selected, "message": option.value, "session_id": request.session_id, "display_message": selected_description, "choice_selection": { "choice_id": request.request_id, "title": request.title, "prompt": request.prompt, "buttons": buttons, "button_rows": button_rows, "selected_label": option.label, "selected_value": option.value, "selected_description": selected_description, }, "feedback": { "request_id": request.request_id, "title": request.title, "prompt": request.prompt, "selected_label": option.label, "selected_value": option.value, "selected_description": selected_description, "buttons": buttons, "button_rows": button_rows, }, } async def resolve_and_save_web_agent_choice( callback_data: str, current_user: AgentChatPrincipal, service: Optional[AgentChatService] = None, persistence: Optional[AgentChatPersistenceService] = None, ) -> Optional[dict[str, Any]]: """ 校验会话归属并在返回选择反馈前持久化，失败不会派发 Agent。 """ parsed = agent_interaction.parse_agent_choice_callback(callback_data) if parsed is None: return None request = agent_interaction.agent_interaction_manager.get_request(parsed[0], str(current_user.id)) if request is None: return None service = service or get_configured_agent_chat_service() existing = await service.get(request.session_id) if existing is not None and not can_access_agent_chat(existing, current_user): return None result = resolve_web_agent_choice_payload(callback_data, str(current_user.id)) if result is None: return None receipt = build_web_agent_display_message(role="user") receipt["id"] = f"choice-selection-{request.request_id}" receipt["choice_selection"] = result["choice_selection"] try: await save_web_agent_display_snapshot( session_id=request.session_id, current_user=current_user, messages=[receipt], service=service, persistence=persistence, append=True, ) except Exception: agent_interaction.agent_interaction_manager.finish_web_choice(request.request_id, str(current_user.id), persisted=False) raise agent_interaction.agent_interaction_manager.finish_web_choice(request.request_id, str(current_user.id), persisted=True) return result def build_web_agent_message_events( message: Message, *, owner_id: str, prepared_audio_path: Optional[Path] = None, ) -> list[dict[str, Any]]: """ 将 Agent 工具通知转换为当前用户的 Web SSE 事件。 :param message: 工具产生的通知消息 :param owner_id: 当前接收通知的用户 ID :param prepared_audio_path: 已准备好的语音文件路径 :return: 前端可直接应用到当前助手消息的事件列表 """ events = [] choice_event = build_web_agent_choice_event(message) if choice_event: events.append(choice_event) text_parts = [str(item).strip() for item in (message.title, message.text) if str(item or "").strip()] if text_parts and not choice_event: events.append({"type": "delta", "content": "\n\n".join(text_parts)}) if message.image: image_ref = message.image image_path = Path(image_ref).expanduser() attachment = None if not image_ref.startswith(("http://", "https://", "data:", "blob:")): attachment = register_web_agent_file( image_ref, file_name=Path(image_ref).name, kind="image", owner_id=owner_id, ) if not attachment: attachment = build_web_agent_url_attachment( image_ref, kind="image", name=message.title or image_path.name or "image", ) events.append({"type": "attachment", "attachment": attachment}) if message.voice_path: audio_path = prepared_audio_path or Path(message.voice_path) attachment = register_web_agent_file( str(audio_path), file_name=audio_path.name, kind="audio", mime_type=get_web_agent_audio_mime_type(audio_path), owner_id=owner_id, ) if attachment: events.append({"type": "attachment", "attachment": attachment}) if message.file_path: attachment = register_web_agent_file( message.file_path, file_name=message.file_name or Path(message.file_path).name, owner_id=owner_id, ) if attachment: events.append({"type": "attachment", "attachment": attachment}) return events async def build_web_agent_message_events_async( message: Message, *, owner_id: str, ) -> list[dict[str, Any]]: """ 异步构造当前用户的 WebAgent 通知，确保语音转码不占用事件循环。 :param message: 工具产生的通知消息 :param owner_id: 当前接收通知的用户 ID :return: 前端可直接应用到当前助手消息的事件列表 """ prepared_audio_path = None if message.voice_path: prepared_audio_path = await prepare_web_agent_audio_attachment_path_async(message.voice_path) return build_web_agent_message_events( message, owner_id=owner_id, prepared_audio_path=prepared_audio_path, ) def build_web_agent_display_message_from_events( events: list[dict[str, Any]], ) -> dict[str, Any]: """ 将传统消息事件聚合为前端展示消息快照。 :param events: 已转换的 WebAgent SSE 事件列表 :return: 可持久化的助手展示消息 """ message = build_web_agent_display_message( role="assistant", status="streaming", ) for event in events: apply_web_agent_display_event(copy.deepcopy(event), message) apply_web_agent_display_event({"type": "done"}, message) return message def split_web_agent_output(text: str) -> list[dict[str, Any]]: """ 将 Agent 输出拆成普通文本与工具提示事件。 :param text: 本次新增的 Agent 文本 :return: 前端可直接渲染的事件片段 """ if not text: return [] events = [] def append_text(content: str) -> None: """ 将工具汇总行从普通文本中拆出，保留与消息渠道一致的展示文案。 """ if not content: return lines = content.splitlines(keepends=True) buffer = "" for line in lines: stripped_line = line.strip() if stripped_line.startswith("（") and stripped_line.endswith("）"): if buffer: events.append({"type": "delta", "content": buffer}) buffer = "" events.append( { "type": "tool", "message": stripped_line, } ) else: buffer += line if buffer: events.append({"type": "delta", "content": buffer}) marker = "⚙️ => " remaining = text while remaining: marker_index = remaining.find(marker) if marker_index < 0: append_text(remaining) break if marker_index > 0: append_text(remaining[:marker_index]) after_marker = remaining[marker_index + len(marker) :] line_end = after_marker.find("\n") if line_end < 0: message = after_marker.strip() remaining = "" else: message = after_marker[:line_end].strip() remaining = after_marker[line_end:].lstrip("\n") if message: events.append({"type": "tool", "message": f"{marker}{message}"}) return events def is_web_agent_traditional_message(text: str) -> bool: """ 判断用户输入是否应走传统消息命令/交互链路。 :param text: 前端输入文本 :return: 需要交给 MessageChain 时返回 True """ normalized = str(text or "").strip() return not agent_interaction.is_agent_learning_command(normalized) and (normalized.startswith("/") or normalized.startswith("CALLBACK:")) def has_web_agent_traditional_interaction(user_id: str) -> bool: """ 判断当前用户是否存在待继续的传统交互会话。 :param user_id: 当前登录用户 ID :return: 存在传统交互上下文时返回 True """ return bool(has_pending_interaction(user_id)) def build_web_agent_command_items() -> list[dict[str, Any]]: """ 读取当前可用斜杠命令并转换为前端建议列表。 :return: 按分类和命令名排序的命令列表 """ commands = get_commands() or {} items = [] for command, data in commands.items(): if not command.startswith("/"): continue if data.get("show") is False: continue items.append( { "command": command, "description": data.get("description") or "", "category": data.get("category") or "其他", "type": data.get("type") or "", "pid": data.get("pid"), } ) return sorted(items, key=lambda item: (item["category"], item["command"])) def extract_web_agent_slash_command(text: str) -> Optional[str]: """ 从 WebAgent 输入中提取斜杠命令名。 :param text: 前端输入文本 :return: 斜杠命令名，非命令输入返回 None """ normalized = str(text or "").strip() if not normalized.startswith("/") or normalized.startswith("//"): return None command = normalized.split(maxsplit=1)[0].strip() return command or None def get_web_agent_unknown_command_message(text: str) -> Optional[str]: """ 判断 WebAgent 斜杠命令是否不存在。 :param text: 前端输入文本 :return: 命令不存在时返回错误提示，命令存在或非命令时返回 None """ command = extract_web_agent_slash_command(text) if not command: return None if get_command(command): return None return f"命令不存在：{command}" def ensure_web_agent_command_allowed(current_user: AgentChatPrincipal) -> Optional[str]: """ 校验当前 Web 用户是否可以执行传统斜杠命令。 :param current_user: 当前登录用户 :return: 无权限时返回错误提示，允许执行时返回 None """ if getattr(current_user, "is_superuser", False): return None return "只有管理员才有权限执行此命令" async def collect_web_agent_traditional_events( *, text: str, current_user: AgentChatPrincipal, original_message_id: Optional[Union[str, int]] = None, original_chat_id: Optional[Union[str, int]] = None, ) -> list[dict[str, Any]]: """ 执行传统消息链路并收集本次 WebAgent 用户产生的通知事件。 :param text: 需要交给传统消息链路处理的文本 :param current_user: 当前登录用户 :param original_message_id: WebAgent 原助手消息 ID :param original_chat_id: WebAgent 原聊天 ID :return: 可直接发送给前端的 SSE 事件列表 """ message_queue: Queue[Message] = Queue() edit_queue: Queue[dict[str, Any]] = Queue() user_id = str(current_user.id) web_agent_events.attach_web_agent_message_queue(user_id, message_queue) web_agent_events.attach_web_agent_edit_queue(user_id, edit_queue) try: await run_in_threadpool( _handle_web_agent_message, channel=NotificationChannel.WebAgent, source=WEB_AGENT_SOURCE, userid=user_id, username=current_user.name or user_id, text=text, original_message_id=original_message_id, original_chat_id=original_chat_id, ) events = [] deadline = time.monotonic() + WEB_AGENT_TRADITIONAL_MAX_WAIT_SECONDS idle_deadline: Optional[float] = None while time.monotonic() < deadline: now = time.monotonic() drained_edit_event = False while True: try: events.append(edit_queue.get_nowait()) drained_edit_event = True except Empty: break if drained_edit_event: idle_deadline = time.monotonic() + WEB_AGENT_TRADITIONAL_IDLE_TIMEOUT_SECONDS continue wait_until = idle_deadline or deadline timeout = max(0.05, min(0.25, wait_until - now, deadline - now)) try: message = await asyncio.to_thread(message_queue.get, True, timeout) except Empty: if idle_deadline and time.monotonic() >= idle_deadline: break continue if not web_agent_events.is_web_agent_message_for_user(message, user_id): continue events.extend(await build_web_agent_message_events_async(message, owner_id=user_id)) idle_deadline = time.monotonic() + WEB_AGENT_TRADITIONAL_IDLE_TIMEOUT_SECONDS return events finally: web_agent_events.detach_web_agent_message_queue(user_id, message_queue) web_agent_events.detach_web_agent_edit_queue(user_id, edit_queue) def build_web_agent_traditional_callback_payload( callback_data: str, original_message_id: Optional[Union[str, int]] = None, original_chat_id: Optional[Union[str, int]] = None, ) -> dict[str, Any]: """ 构造传统消息链按钮回调的前端执行载荷。 :param callback_data: 前端点击的传统按钮回调数据 :param original_message_id: WebAgent 原助手消息 ID :param original_chat_id: WebAgent 原聊天 ID :return: 前端可继续发送到 /stream 的消息载荷 """ return { "message": f"CALLBACK:{callback_data}", "display_message": callback_data, "traditional": True, "original_message_id": original_message_id, "original_chat_id": original_chat_id, } async def _single_web_agent_event(event: dict[str, Any]) -> AsyncIterator[dict[str, Any]]: """ 把前置校验结果包装成单事件异步流。 """ yield event def _build_traditional_web_agent_stream( *, command: WebAgentStreamCommand, current_user: AgentChatPrincipal, session_id: str, prompt: str, display_prompt: str, service: AgentChatService, persistence: AgentChatPersistenceService, is_disconnected: Callable[[], Awaitable[bool]], ) -> AsyncIterator[dict[str, Any]]: """ 构造传统消息链路的 WebAgent 事件流。 """ user_attachments = build_web_agent_input_attachments( images=command.images, files=command.files, audio_refs=command.audio_refs, ) display_messages = [] if command.echo_user: display_messages.append( build_web_agent_display_message( role="user", content=display_prompt or prompt, attachments=user_attachments, ) ) async def event_generator() -> AsyncIterator[dict[str, Any]]: """ 等待传统消息结果并投影为传输无关事件。 """ yield {"type": "start", "session_id": session_id} collection_task = asyncio.create_task( collect_web_agent_traditional_events( text=prompt, current_user=current_user, original_message_id=command.original_message_id, original_chat_id=command.original_chat_id, ) ) try: while True: try: events = await asyncio.wait_for( asyncio.shield(collection_task), timeout=WEB_AGENT_STREAM_HEARTBEAT_SECONDS, ) break except asyncio.TimeoutError: if await is_disconnected(): collection_task.cancel() return yield {"type": "heartbeat"} except asyncio.CancelledError: return finally: if not collection_task.done(): collection_task.cancel() await asyncio.gather(collection_task, return_exceptions=True) display_messages.append(build_web_agent_display_message_from_events(events)) async def save_display_snapshot() -> None: """ 后台保存传统消息展示快照，不阻塞事件流终态。 """ try: await save_web_agent_display_snapshot( session_id=session_id, current_user=current_user, messages=display_messages, client_session_id=command.session_id or session_id, service=service, persistence=persistence, append=True, ) except Exception as err: logger.error(f"保存WebAgent传统消息快照失败: {str(err)}") create_web_agent_background_task(save_display_snapshot()) await asyncio.sleep(0) for event in events: yield copy.deepcopy(event) if await is_disconnected(): return yield {"type": "done"} return event_generator() async def build_web_agent_stream( command: WebAgentStreamCommand, *, current_user: AgentChatPrincipal, is_disconnected: Callable[[], Awaitable[bool]], protected_transport_supported: bool, service: Optional[AgentChatService] = None, persistence: Optional[AgentChatPersistenceService] = None, ) -> WebAgentStreamResult: """ 编排 WebAgent 会话、传统消息、音频输入、事件投影与展示持久化。 """ prompt = command.text.strip() display_prompt = (command.display_text or command.text).strip() service = service or get_configured_agent_chat_service() persistence = persistence or get_configured_agent_chat_persistence() try: session_id = await build_web_agent_session_id_async( current_user, command.session_id, service, ) except PermissionError: return WebAgentStreamResult( events=_single_web_agent_event({"type": "error", "message": "会话不存在或无权访问"}) ) if command.choice_selection: choice_id = str(command.choice_selection.get("choice_id") or "") claimed = agent_interaction.agent_interaction_manager.get_web_choice(choice_id, str(current_user.id), session_id) if claimed is None: return WebAgentStreamResult(events=_single_web_agent_event({ "type": "error", "message": "该选择已处理或不属于当前会话", })) choice_request, option = claimed canonical = resolve_web_agent_choice_payload( agent_interaction.build_agent_choice_callback(choice_id, choice_request.selected_index or 1), str(current_user.id), ) if canonical is None: return WebAgentStreamResult(events=_single_web_agent_event({"type": "error", "message": "该选择已失效"})) prompt = option.value display_prompt = option.description or option.label command = replace(command, text=prompt, display_text=display_prompt, choice_selection=canonical["choice_selection"]) is_secret_confirmation_candidate = ( prompt in {"确认", "取消"} and not command.images and not command.audio_refs and not command.files ) manager = agent_application.get_running_agent_manager() is_secret_confirmation_control = ( is_secret_confirmation_candidate and manager is not None and manager.matches_secret_confirmation( session_id, str(current_user.id), channel=NotificationChannel.WebAgent.value, source=WEB_AGENT_SOURCE, ) ) if is_secret_confirmation_control and not protected_transport_supported: return WebAgentStreamResult( events=_single_web_agent_event( { "type": "error", "message": "当前客户端不支持安全交付敏感设置，未执行操作。", } ) ) is_traditional_message = not command.choice_selection and not agent_interaction.is_agent_learning_command(prompt) and ( is_web_agent_traditional_message(prompt) or has_web_agent_traditional_interaction(str(current_user.id)) ) if is_traditional_message: denied_message = ensure_web_agent_command_allowed(current_user) if denied_message: return WebAgentStreamResult(events=_single_web_agent_event({"type": "error", "message": denied_message})) unknown_command_message = get_web_agent_unknown_command_message(prompt) if unknown_command_message: return WebAgentStreamResult( events=_single_web_agent_event({"type": "error", "message": unknown_command_message}) ) return WebAgentStreamResult( events=_build_traditional_web_agent_stream( command=command, current_user=current_user, session_id=session_id, prompt=prompt, display_prompt=display_prompt, service=service, persistence=persistence, is_disconnected=is_disconnected, ) ) if not get_api_runtime_config_snapshot().ai_agent_enable: return WebAgentStreamResult( events=_single_web_agent_event({"type": "error", "message": "智能助手未启用，请先在系统设置中开启。"}) ) if manager is None: return WebAgentStreamResult( events=_single_web_agent_event( { "type": "error", "message": "智能助手服务尚未就绪，请稍后重试。", } ) ) from app.application.messaging.webagent.stream import ( WebAgentStreamDependencies, _build_steering_ack_stream, build_agent_web_agent_stream, submit_web_agent_steering, ) transcript = await transcribe_web_agent_audio_input( command.audio_refs, owner_id=str(current_user.id), ) prompt = merge_web_agent_prompt_with_transcript(prompt, transcript) display_prompt = merge_web_agent_prompt_with_transcript(display_prompt, transcript) if not prompt and command.audio_refs and not command.images and not command.files: return WebAgentStreamResult( events=_single_web_agent_event({"type": "error", "message": "语音识别失败，请稍后重试。"}) ) if not prompt and not command.images and not command.files and not command.audio_refs: return WebAgentStreamResult( events=_single_web_agent_event( { "type": "error", "message": "请输入要发送给智能助手的内容或选择附件。", } ) ) # 活动 Agent 运行内的补充输入走同一张图；无活动运行时才创建新一轮。 steering_message = None if command.choice_selection else await submit_web_agent_steering( manager=manager, session_id=session_id, user_id=str(current_user.id), prompt=prompt, images=command.images, files=command.files, audio_refs=command.audio_refs, ) if steering_message is not None: return WebAgentStreamResult( events=_build_steering_ack_stream(session_id=session_id, message=steering_message), control="steering", ) stream_dependencies = WebAgentStreamDependencies( event_publisher_factory=WebAgentEventPublisher, bind_user_session=bind_web_agent_user_session, apply_display_event=apply_web_agent_display_event, build_display_message=build_web_agent_display_message, build_input_attachments=build_web_agent_input_attachments, build_message_events_async=partial( build_web_agent_message_events_async, owner_id=str(current_user.id), ), save_display_snapshot=save_web_agent_display_snapshot, split_output=split_web_agent_output, create_background_task=create_web_agent_background_task, source=WEB_AGENT_SOURCE, heartbeat_seconds=WEB_AGENT_STREAM_HEARTBEAT_SECONDS, ) return WebAgentStreamResult( events=build_agent_web_agent_stream( command=command, current_user=current_user, session_id=session_id, prompt=prompt, display_prompt=display_prompt, has_audio_input=bool(transcript), is_secret_confirmation_control=is_secret_confirmation_control, protected_transport_supported=protected_transport_supported, service=service, persistence=persistence, is_disconnected=is_disconnected, dependencies=stream_dependencies, ), control=("secret-confirmation" if is_secret_confirmation_control else None), )