mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 11:27:43 +08:00
7584d4b9fc
* fix: arbitrate subscription identity and restore workflow state * fix: isolate agent sessions and retain runtime resource ownership * test: preserve legacy empty-owner subscription transactions * test: handle reaped terminal descendants during shutdown polling * fix: align shared atomic configuration method with lifecycle prerequisite * test: cover subagent cancellation and completion lifecycle boundaries --------- Co-authored-by: jxxghp <jxxghp@gmail.com>
498 lines
19 KiB
Python
498 lines
19 KiB
Python
"""
|
||
WebAgent 主运行流和运行中补充消息的展示桥接。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
from dataclasses import dataclass
|
||
from functools import partial
|
||
from typing import Any, AsyncIterator, Awaitable, Callable, Coroutine, Optional, cast
|
||
|
||
from app.agent.steering import SteeringMessage
|
||
from app.application import agent as agent_application
|
||
from app.runtime.log import logger
|
||
from app.runtime.stop import runtime_stop_state
|
||
from app.schemas.message import Message
|
||
from app.schemas.types import NotificationChannel, ReplyMode
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class WebAgentStreamDependencies:
|
||
"""
|
||
注入 WebAgent 流所需的展示、事件和后台任务能力，避免模块相互导入。
|
||
|
||
:param event_publisher_factory: 创建当前事件流使用的发布器
|
||
:param bind_user_session: 接收用户标识和会话标识，维护用户的活动会话绑定
|
||
:param apply_display_event: 把一个流事件合并到可持久化的展示消息对象
|
||
:param build_display_message: 按角色、正文和附件创建展示消息
|
||
:param build_input_attachments: 把图片、音频和文件引用转换为展示附件
|
||
:param build_message_events_async: 把渠道消息异步转换为 WebAgent 展示事件列表
|
||
:param save_display_snapshot: 异步保存当前用户会话的展示消息快照
|
||
:param split_output: 把助手输出文本拆为按序展示的消息片段
|
||
:param create_background_task: 将协程交给宿主管理并返回后台任务
|
||
:param source: 本次会话使用的 WebAgent 来源标识
|
||
:param heartbeat_seconds: 事件流空闲时发送心跳的间隔，单位为秒
|
||
"""
|
||
|
||
event_publisher_factory: Callable[[], Any]
|
||
bind_user_session: Callable[[str, str], None]
|
||
apply_display_event: Callable[[dict[str, Any], dict[str, Any]], None]
|
||
build_display_message: Callable[..., dict[str, Any]]
|
||
build_input_attachments: Callable[..., list[dict[str, Any]]]
|
||
build_message_events_async: Callable[[Message], Awaitable[list[dict[str, Any]]]]
|
||
save_display_snapshot: Callable[..., Awaitable[None]]
|
||
split_output: Callable[[str], list[dict[str, Any]]]
|
||
create_background_task: Callable[
|
||
[Coroutine[Any, Any, object]], asyncio.Task[object]
|
||
]
|
||
source: str
|
||
heartbeat_seconds: float
|
||
|
||
|
||
async def submit_web_agent_steering(
|
||
*,
|
||
manager: Any,
|
||
session_id: str,
|
||
user_id: str,
|
||
prompt: str,
|
||
images: list[str],
|
||
files: list[dict[str, Any]],
|
||
audio_refs: list[str],
|
||
) -> Optional[SteeringMessage]:
|
||
"""
|
||
把活动 WebAgent 的补充输入提交到会话 inbox；无活动运行时返回 None。
|
||
"""
|
||
steering_files = [file for file in files if str(file.get("ref") or "") not in set(audio_refs)]
|
||
steering_files.extend({"ref": audio_ref, "mime_type": "audio/*"} for audio_ref in audio_refs)
|
||
submit = getattr(manager, "submit_steering_message", None)
|
||
if not callable(submit):
|
||
return None
|
||
return cast(
|
||
Optional[SteeringMessage],
|
||
await submit(
|
||
session_id=session_id,
|
||
user_id=user_id,
|
||
message=prompt,
|
||
images=images,
|
||
files=steering_files or None,
|
||
),
|
||
)
|
||
|
||
|
||
def _build_steering_ack_stream(
|
||
*,
|
||
session_id: str,
|
||
message: SteeringMessage,
|
||
) -> AsyncIterator[dict[str, Any]]:
|
||
"""
|
||
为运行中补充消息返回短 ACK 流，不创建第二个助手展示气泡。
|
||
"""
|
||
|
||
async def event_generator() -> AsyncIterator[dict[str, Any]]:
|
||
"""
|
||
按统一 WebAgent SSE 协议报告已排队的消息 ID。
|
||
"""
|
||
yield {"type": "start", "session_id": session_id}
|
||
yield {
|
||
"type": "steering",
|
||
"status": "queued",
|
||
"message_id": message.message_id,
|
||
"content": message.text,
|
||
}
|
||
yield {"type": "done"}
|
||
|
||
return event_generator()
|
||
|
||
|
||
def _publish_web_agent_protected_output(
|
||
event_publisher: Any,
|
||
content: str,
|
||
) -> bool:
|
||
"""
|
||
发布不进入普通展示快照的敏感交互结果。
|
||
"""
|
||
return bool(event_publisher.publish({"type": "interaction-protected", "content": content}))
|
||
|
||
|
||
def _attach_assistant_message_id(
|
||
event: dict[str, Any],
|
||
assistant_message_ref: dict[str, dict[str, Any]],
|
||
) -> dict[str, Any]:
|
||
"""
|
||
为主流展示事件附加产生它的助手消息身份。
|
||
"""
|
||
if event.get("assistant_message_id"):
|
||
return event
|
||
assistant_message_id = assistant_message_ref["message"].get("id")
|
||
if not assistant_message_id:
|
||
return event
|
||
return {**event, "assistant_message_id": assistant_message_id}
|
||
|
||
|
||
def _build_web_agent_steering_callback(
|
||
*,
|
||
display_messages: list[dict[str, Any]],
|
||
assistant_message_ref: dict[str, dict[str, Any]],
|
||
event_publisher: Any,
|
||
build_display_message: Callable[..., dict[str, Any]],
|
||
build_input_attachments: Callable[..., list[dict[str, Any]]],
|
||
) -> Callable[[SteeringMessage, str], None]:
|
||
"""
|
||
构造运行中补充消息的应用回调，并同步展示快照与 SSE 时间线。
|
||
"""
|
||
|
||
def steering_status_callback(message: SteeringMessage, status: str) -> None:
|
||
"""
|
||
把补充消息应用状态投影到当前助手流的真实边界。
|
||
"""
|
||
if status != "applied":
|
||
return
|
||
assistant_message = assistant_message_ref["message"]
|
||
assistant_message["status"] = "done"
|
||
display_message = build_display_message(
|
||
role="user",
|
||
content=message.text,
|
||
attachments=build_input_attachments(
|
||
images=list(message.images),
|
||
files=list(message.files),
|
||
audio_refs=[],
|
||
),
|
||
)
|
||
display_message["steering_message_id"] = message.message_id
|
||
# 应用点是消息流中的真实边界：先收口前一段助手输出，再插入用户消息，
|
||
# 最后创建承接后续工具和文本的新助手气泡。
|
||
# 用对象身份定位当前助手段；字典内容可能暂时相同，不能用值相等误选历史段。
|
||
assistant_index = next(
|
||
(index for index, item in enumerate(display_messages) if item is assistant_message),
|
||
-1,
|
||
)
|
||
if assistant_index < 0:
|
||
logger.warning("WebAgent steering 应用时未找到当前助手展示段: %s", message.message_id)
|
||
return
|
||
display_messages.insert(assistant_index + 1, display_message)
|
||
continuation = build_display_message(role="assistant", status="streaming")
|
||
display_messages.insert(assistant_index + 2, continuation)
|
||
assistant_message_ref["message"] = continuation
|
||
event_publisher.publish(
|
||
{
|
||
"type": "steering",
|
||
"status": "applied",
|
||
"message_id": message.message_id,
|
||
"content": message.text,
|
||
"display_message": display_message,
|
||
"assistant_message_id": assistant_message.get("id"),
|
||
"continuation_message_id": continuation.get("id"),
|
||
}
|
||
)
|
||
|
||
return steering_status_callback
|
||
|
||
|
||
def _build_web_agent_message_callback(
|
||
*,
|
||
assistant_message_ref: dict[str, dict[str, Any]],
|
||
event_publisher: Any,
|
||
build_message_events_async: Callable[[Message], Awaitable[list[dict[str, Any]]]],
|
||
apply_display_event: Callable[[dict[str, Any], dict[str, Any]], None],
|
||
) -> Callable[[Message], Awaitable[None]]:
|
||
"""
|
||
构造 Agent 主动消息的 Web 展示回调。
|
||
"""
|
||
|
||
async def message_callback(message: Message) -> None:
|
||
"""
|
||
接收 Agent 工具主动发送的 Web 通知。
|
||
"""
|
||
for item in await build_message_events_async(message):
|
||
display_event = _attach_assistant_message_id(item, assistant_message_ref)
|
||
apply_display_event(display_event, assistant_message_ref["message"])
|
||
event_publisher.publish(display_event)
|
||
|
||
return message_callback
|
||
|
||
|
||
def _build_web_agent_output_callback(
|
||
*,
|
||
assistant_message_ref: dict[str, dict[str, Any]],
|
||
event_publisher: Any,
|
||
split_output: Callable[[str], list[dict[str, Any]]],
|
||
apply_display_event: Callable[[dict[str, Any], dict[str, Any]], None],
|
||
) -> Callable[[str], None]:
|
||
"""
|
||
构造 Agent 文本增量的 Web 展示回调。
|
||
"""
|
||
|
||
def output_callback(delta: str) -> None:
|
||
"""
|
||
接收 Agent 文本增量并投影为展示事件。
|
||
"""
|
||
for item in split_output(delta):
|
||
display_event = _attach_assistant_message_id(item, assistant_message_ref)
|
||
apply_display_event(display_event, assistant_message_ref["message"])
|
||
event_publisher.publish(display_event)
|
||
|
||
return output_callback
|
||
|
||
|
||
def _build_web_agent_tool_event_callback(
|
||
*,
|
||
assistant_message_ref: dict[str, dict[str, Any]],
|
||
event_publisher: Any,
|
||
apply_display_event: Callable[[dict[str, Any], dict[str, Any]], None],
|
||
) -> Callable[[dict[str, Any]], None]:
|
||
"""
|
||
构造结构化工具生命周期回调，确保快照与 SSE 使用同一事件顺序。
|
||
"""
|
||
|
||
def tool_event_callback(event: dict[str, Any]) -> None:
|
||
"""
|
||
投影工具开始、完成和失败事件。
|
||
"""
|
||
display_event = _attach_assistant_message_id(event, assistant_message_ref)
|
||
apply_display_event(display_event, assistant_message_ref["message"])
|
||
event_publisher.publish(display_event)
|
||
|
||
return tool_event_callback
|
||
|
||
|
||
def _build_web_agent_files(command: Any) -> list[dict[str, Any]]:
|
||
"""
|
||
把 WebAgent 命令中的普通文件和音频引用整理成 Agent 输入。
|
||
"""
|
||
audio_refs = {str(audio_ref) for audio_ref in command.audio_refs}
|
||
files = [file for file in command.files if str(file.get("ref") or "") not in audio_refs]
|
||
files.extend({"ref": audio_ref, "mime_type": "audio/*"} for audio_ref in command.audio_refs)
|
||
return files
|
||
|
||
|
||
def _build_web_agent_event_generator(
|
||
*,
|
||
command: Any,
|
||
current_user: Any,
|
||
session_id: str,
|
||
prompt: str,
|
||
display_messages: list[dict[str, Any]],
|
||
assistant_message_ref: dict[str, dict[str, Any]],
|
||
has_audio_input: bool,
|
||
is_secret_confirmation_control: bool,
|
||
protected_transport_supported: bool,
|
||
service: Any,
|
||
persistence: Any,
|
||
is_disconnected: Callable[[], Awaitable[bool]],
|
||
dependencies: WebAgentStreamDependencies,
|
||
event_publisher: Any,
|
||
output_callback: Callable[[str], None],
|
||
tool_event_callback: Callable[[dict[str, Any]], None],
|
||
message_callback: Callable[[Message], Awaitable[None]],
|
||
protected_output_callback: Callable[[str], bool],
|
||
steering_status_callback: Callable[[SteeringMessage, str], None],
|
||
) -> AsyncIterator[dict[str, Any]]:
|
||
"""
|
||
构造后台 Agent 任务并按断线与终态语义消费展示事件。
|
||
"""
|
||
files = _build_web_agent_files(command)
|
||
|
||
async def run_agent() -> None:
|
||
"""
|
||
后台执行 Agent，并在完成后持久化展示快照。
|
||
"""
|
||
try:
|
||
runtime_manager = agent_application.get_running_agent_manager()
|
||
if runtime_manager is None:
|
||
raise RuntimeError("智能助手服务尚未就绪，请稍后重试。")
|
||
await runtime_manager.process_message(
|
||
session_id=session_id,
|
||
user_id=str(current_user.id),
|
||
message=prompt,
|
||
images=command.images,
|
||
files=files or None,
|
||
has_audio_input=has_audio_input,
|
||
channel=NotificationChannel.WebAgent.value,
|
||
source=dependencies.source,
|
||
username=current_user.name,
|
||
reply_mode=ReplyMode.CAPTURE_ONLY,
|
||
allow_message_tools=True,
|
||
output_callback=output_callback,
|
||
tool_event_callback=tool_event_callback,
|
||
protected_output_callback=(protected_output_callback if protected_transport_supported else None),
|
||
message_callback=message_callback,
|
||
steering_status_callback=steering_status_callback,
|
||
agent_factory=agent_application.get_web_agent_type(),
|
||
wait_for_completion=True,
|
||
)
|
||
except asyncio.CancelledError:
|
||
# 显式停止会话沿用正常终止语义；服务关闭由 manager 的稳定异常分支处理。
|
||
pass
|
||
except Exception as err:
|
||
logger.error(f"Web智能助手执行失败: {str(err)}")
|
||
error_event = _attach_assistant_message_id({
|
||
"type": "error",
|
||
"message": "智能助手执行失败，请稍后重试",
|
||
}, assistant_message_ref)
|
||
dependencies.apply_display_event(error_event, assistant_message_ref["message"])
|
||
event_publisher.publish(error_event)
|
||
finally:
|
||
done_event = _attach_assistant_message_id({"type": "done"}, assistant_message_ref)
|
||
dependencies.apply_display_event(done_event, assistant_message_ref["message"])
|
||
# 终态先进入事件队列，避免展示快照落库延迟前端结束动画。
|
||
event_publisher.publish(done_event)
|
||
if not is_secret_confirmation_control:
|
||
try:
|
||
await dependencies.save_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展示历史失败：{err}")
|
||
|
||
async def event_generator() -> AsyncIterator[dict[str, Any]]:
|
||
"""
|
||
执行后台 Agent 并在客户端断线时保留运行任务。
|
||
"""
|
||
task = dependencies.create_background_task(run_agent())
|
||
disconnected = False
|
||
terminal_sent = False
|
||
try:
|
||
yield {
|
||
"type": "start",
|
||
"session_id": session_id,
|
||
# 让前端把本地占位助手段绑定到服务端展示段，后续 steering
|
||
# 应用事件即可按稳定 ID 定位真实消息边界。
|
||
"assistant_message_id": assistant_message_ref["message"].get("id"),
|
||
}
|
||
while not runtime_stop_state.is_system_stopped:
|
||
if await is_disconnected():
|
||
disconnected = True
|
||
break
|
||
try:
|
||
event = await asyncio.wait_for(
|
||
event_publisher.get(),
|
||
timeout=dependencies.heartbeat_seconds,
|
||
)
|
||
except asyncio.TimeoutError:
|
||
yield {"type": "heartbeat"}
|
||
continue
|
||
if event.get("type") == "done":
|
||
terminal_sent = True
|
||
yield event
|
||
if event.get("type") == "done":
|
||
break
|
||
except asyncio.CancelledError:
|
||
disconnected = True
|
||
return
|
||
finally:
|
||
if not task.done() and not disconnected and not terminal_sent:
|
||
task.cancel()
|
||
try:
|
||
await task
|
||
except asyncio.CancelledError:
|
||
pass
|
||
await event_publisher.aclose()
|
||
# 客户端断线后保留 Agent 继续执行；发布器关闭后拒绝受保护结果。
|
||
|
||
return event_generator()
|
||
|
||
|
||
def build_agent_web_agent_stream(
|
||
*,
|
||
command: Any,
|
||
current_user: Any,
|
||
session_id: str,
|
||
prompt: str,
|
||
display_prompt: str,
|
||
has_audio_input: bool,
|
||
is_secret_confirmation_control: bool,
|
||
protected_transport_supported: bool,
|
||
service: Any,
|
||
persistence: Any,
|
||
is_disconnected: Callable[[], Awaitable[bool]],
|
||
dependencies: WebAgentStreamDependencies,
|
||
) -> AsyncIterator[dict[str, Any]]:
|
||
"""
|
||
构造标准 Agent 执行链路的 WebAgent 事件流。
|
||
"""
|
||
dependencies.bind_user_session(str(current_user.id), session_id)
|
||
event_publisher = dependencies.event_publisher_factory()
|
||
user_attachments = dependencies.build_input_attachments(
|
||
images=command.images,
|
||
files=command.files,
|
||
audio_refs=command.audio_refs,
|
||
)
|
||
display_messages = []
|
||
if command.echo_user and not is_secret_confirmation_control:
|
||
user_display_message = dependencies.build_display_message(
|
||
role="user",
|
||
content=display_prompt or prompt,
|
||
attachments=user_attachments,
|
||
)
|
||
if command.choice_selection:
|
||
user_display_message["choice_selection"] = command.choice_selection
|
||
display_messages.append(user_display_message)
|
||
assistant_display_message = dependencies.build_display_message(
|
||
role="assistant",
|
||
status="streaming",
|
||
)
|
||
display_messages.append(assistant_display_message)
|
||
assistant_message_ref = {"message": assistant_display_message}
|
||
output_callback = _build_web_agent_output_callback(
|
||
assistant_message_ref=assistant_message_ref,
|
||
event_publisher=event_publisher,
|
||
split_output=dependencies.split_output,
|
||
apply_display_event=dependencies.apply_display_event,
|
||
)
|
||
message_callback = _build_web_agent_message_callback(
|
||
assistant_message_ref=assistant_message_ref,
|
||
event_publisher=event_publisher,
|
||
build_message_events_async=dependencies.build_message_events_async,
|
||
apply_display_event=dependencies.apply_display_event,
|
||
)
|
||
tool_event_callback = _build_web_agent_tool_event_callback(
|
||
assistant_message_ref=assistant_message_ref,
|
||
event_publisher=event_publisher,
|
||
apply_display_event=dependencies.apply_display_event,
|
||
)
|
||
protected_output_callback = partial(_publish_web_agent_protected_output, event_publisher)
|
||
steering_status_callback = _build_web_agent_steering_callback(
|
||
display_messages=display_messages,
|
||
assistant_message_ref=assistant_message_ref,
|
||
event_publisher=event_publisher,
|
||
build_display_message=dependencies.build_display_message,
|
||
build_input_attachments=dependencies.build_input_attachments,
|
||
)
|
||
|
||
return _build_web_agent_event_generator(
|
||
command=command,
|
||
current_user=current_user,
|
||
session_id=session_id,
|
||
prompt=prompt,
|
||
display_messages=display_messages,
|
||
assistant_message_ref=assistant_message_ref,
|
||
has_audio_input=has_audio_input,
|
||
is_secret_confirmation_control=is_secret_confirmation_control,
|
||
protected_transport_supported=protected_transport_supported,
|
||
service=service,
|
||
persistence=persistence,
|
||
is_disconnected=is_disconnected,
|
||
dependencies=dependencies,
|
||
event_publisher=event_publisher,
|
||
output_callback=output_callback,
|
||
tool_event_callback=tool_event_callback,
|
||
message_callback=message_callback,
|
||
protected_output_callback=protected_output_callback,
|
||
steering_status_callback=steering_status_callback,
|
||
)
|
||
|
||
|
||
__all__ = [
|
||
"WebAgentStreamDependencies",
|
||
"_build_steering_ack_stream",
|
||
"build_agent_web_agent_stream",
|
||
"submit_web_agent_steering",
|
||
]
|