wehub-resource-sync
354 lines
14 KiB
Python
354 lines
14 KiB
Python
"""Subagent run orchestration service.
|
|
|
|
This module owns parent/child agent-thread relationships. It decides whether a
|
|
task starts a new child thread or continues an existing one, records the
|
|
``SubagentThread`` relation and builds the subagent-only runtime payload.
|
|
|
|
It deliberately delegates durable run mechanics to ``agent_run_service``:
|
|
request id idempotency, active-run conflict checks, input message persistence,
|
|
AgentRun row creation and queue enqueueing all stay in the shared AgentRun
|
|
lifecycle boundary.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
import yuxi.services.agent_run_service as agent_run_service
|
|
from fastapi import HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from yuxi.repositories.agent_run_repository import AgentRunRepository
|
|
from yuxi.repositories.conversation_repository import ConversationRepository
|
|
from yuxi.repositories.subagent_thread_repository import SubagentThreadRepository
|
|
from yuxi.services.input_message_service import AgentRunInputMessage
|
|
from yuxi.storage.postgres.models_business import Agent, AgentRun, SubagentThread
|
|
from yuxi.utils.datetime_utils import utc_isoformat
|
|
from yuxi.utils.hash_utils import hash_id, subagent_child_thread_id
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SubagentStartResult:
|
|
run: AgentRun
|
|
created: bool
|
|
continuing: bool
|
|
relation: SubagentThread
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SubagentRunBusy(Exception):
|
|
thread_id: str
|
|
active_run_id: str | None
|
|
active_run_status: str | None
|
|
message: str | None
|
|
|
|
def to_payload(self) -> dict:
|
|
return {
|
|
"status": "busy",
|
|
"thread_id": self.thread_id,
|
|
"active_run_id": self.active_run_id,
|
|
"active_run_status": self.active_run_status,
|
|
"message": self.message,
|
|
}
|
|
|
|
|
|
def subagent_run_urls(run_id: str) -> dict[str, str]:
|
|
"""生成子智能体 run 对外暴露的事件流和结果查询 URL。"""
|
|
return {
|
|
"events_url": f"/api/agent/runs/{run_id}/events",
|
|
"result_url": f"/api/agent/runs/{run_id}/result",
|
|
}
|
|
|
|
|
|
def serialize_subagent_run_state(run: AgentRun) -> dict:
|
|
"""序列化给父智能体状态使用的子智能体 run 摘要。
|
|
|
|
任务描述不在此冗余存储：其唯一来源是父对话里 `task` 工具调用的入参，
|
|
前端面板按 tool_call_id 回填展示。
|
|
"""
|
|
payload = run.input_payload
|
|
if not isinstance(payload, dict):
|
|
raise ValueError("subagent run 缺少 input_payload")
|
|
runtime = payload.get("runtime")
|
|
if not isinstance(runtime, dict):
|
|
raise ValueError("subagent run 缺少 runtime")
|
|
tool_call_id = str(runtime.get("tool_call_id") or "").strip()
|
|
if not tool_call_id:
|
|
raise ValueError("subagent run 缺少 tool_call_id")
|
|
|
|
state = {
|
|
"id": tool_call_id,
|
|
"run_id": run.id,
|
|
"subagent_slug": run.agent_slug,
|
|
"subagent_name": runtime.get("subagent_name"),
|
|
"child_thread_id": run.conversation_thread_id,
|
|
"status": run.status,
|
|
"created_at": utc_isoformat(run.created_at) if run.created_at else None,
|
|
"completed_at": utc_isoformat(run.finished_at) if run.finished_at else None,
|
|
"error": run.error_message,
|
|
**subagent_run_urls(run.id),
|
|
}
|
|
return {key: value for key, value in state.items() if value is not None}
|
|
|
|
|
|
class SubagentRunService:
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
self.run_repo = AgentRunRepository(db)
|
|
self.conv_repo = ConversationRepository(db)
|
|
self.thread_repo = SubagentThreadRepository(db)
|
|
|
|
async def start(
|
|
self,
|
|
*,
|
|
uid: str,
|
|
created_by_run_id: str,
|
|
agent_item: Agent,
|
|
input_message: AgentRunInputMessage,
|
|
tool_call_id: str,
|
|
requested_thread_id: str | None = None,
|
|
file_thread_id: str | None = None,
|
|
model_spec: str | None = None,
|
|
) -> SubagentStartResult:
|
|
"""启动或继续一个后台子智能体 run，并在新建时入队 worker。"""
|
|
|
|
creator_run = await self.run_repo.get_run_for_user(created_by_run_id, uid)
|
|
if not creator_run:
|
|
raise ValueError("父运行任务不存在")
|
|
if getattr(creator_run, "run_type", None) == "subagent":
|
|
raise ValueError("子智能体不能创建子智能体")
|
|
|
|
child_thread_id = str(requested_thread_id or "").strip()
|
|
continuing = bool(child_thread_id) # 表示是继续运行已有子智能体线程，而非新建子线程
|
|
if not child_thread_id:
|
|
child_thread_id = subagent_child_thread_id(
|
|
creator_run.conversation_thread_id,
|
|
agent_item.slug,
|
|
tool_call_id,
|
|
)
|
|
|
|
# 1. 确保子线程有对应 conversation，必要时创建 subagent 对话
|
|
# 2. 确保父子线程关系存在，必要时创建 SubagentThread 记录；relation 是后台子 run 的线程归属来源
|
|
relation = await self._ensure_thread_relation(
|
|
child_thread_id=child_thread_id,
|
|
uid=uid,
|
|
agent_item=agent_item,
|
|
creator_run=creator_run,
|
|
continuing=continuing,
|
|
)
|
|
|
|
# 创建数据库记录
|
|
request_id = hash_id("req:", f"{creator_run.id}:{child_thread_id}:{tool_call_id}")
|
|
try:
|
|
run, created = await self._create_run_record(
|
|
input_message=input_message,
|
|
request_id=request_id,
|
|
current_uid=uid,
|
|
model_spec=model_spec,
|
|
creator_run=creator_run,
|
|
relation=relation,
|
|
tool_call_id=tool_call_id,
|
|
file_thread_id=file_thread_id,
|
|
)
|
|
except HTTPException as exc:
|
|
detail = exc.detail
|
|
if exc.status_code == 409 and isinstance(detail, dict) and detail.get("code") == "run_busy":
|
|
raise SubagentRunBusy(
|
|
thread_id=str(detail.get("thread_id") or child_thread_id),
|
|
active_run_id=detail.get("active_run_id"),
|
|
active_run_status=detail.get("active_run_status"),
|
|
message=detail.get("message"),
|
|
) from exc
|
|
raise ValueError(detail if isinstance(detail, str) else json.dumps(detail, ensure_ascii=False)) from exc
|
|
|
|
# 创建成功后入队 worker 执行；幂等命中已有 run 时不重复入队。
|
|
if created:
|
|
await self.db.commit()
|
|
await agent_run_service.enqueue_agent_run(run.id)
|
|
|
|
return SubagentStartResult(
|
|
run=run,
|
|
created=created,
|
|
continuing=continuing,
|
|
relation=relation,
|
|
)
|
|
|
|
async def get_run_for_creator(self, *, uid: str, created_by_run_id: str, run_id: str) -> AgentRun:
|
|
"""在父 run 作用域内读取子智能体 run，防止工具访问其它对话的子任务。"""
|
|
run = await self.run_repo.get_subagent_run_for_creator(
|
|
uid=uid,
|
|
created_by_run_id=created_by_run_id,
|
|
run_id=run_id,
|
|
)
|
|
if not run:
|
|
raise ValueError("子智能体运行不存在或不属于当前父运行")
|
|
return run
|
|
|
|
async def _create_run_record(
|
|
self,
|
|
*,
|
|
input_message: AgentRunInputMessage,
|
|
request_id: str,
|
|
current_uid: str,
|
|
model_spec: str | None,
|
|
creator_run: AgentRun,
|
|
relation: SubagentThread,
|
|
tool_call_id: str,
|
|
file_thread_id: str | None,
|
|
) -> tuple[Any, bool]:
|
|
"""创建后台子智能体 run，并把规范化输入消息保存为该 run 的输入。"""
|
|
if not input_message.content:
|
|
raise HTTPException(status_code=422, detail="input_message 不能为空")
|
|
|
|
scope = await agent_run_service.prepare_agent_run_creation_scope(
|
|
agent_slug=relation.subagent_slug,
|
|
conversation_thread_id=relation.child_thread_id,
|
|
request_id=request_id,
|
|
current_uid=current_uid,
|
|
db=self.db,
|
|
run_type="subagent",
|
|
agent_kind="subagent",
|
|
created_by_run_id=creator_run.id,
|
|
subagent_thread_relation_id=relation.id,
|
|
)
|
|
if relation.child_conversation_id != scope.conversation.id:
|
|
raise HTTPException(status_code=409, detail="subagent thread relation 与本次运行不匹配")
|
|
if scope.existing_run:
|
|
return scope.existing_run, False
|
|
|
|
if creator_run.conversation_id != relation.parent_conversation_id:
|
|
raise HTTPException(status_code=409, detail="subagent thread relation 与本次运行不匹配")
|
|
|
|
resolved_model_spec = agent_run_service.resolve_agent_run_model_spec(
|
|
model_spec,
|
|
scope.agent_item,
|
|
scope.agent_backend,
|
|
)
|
|
runtime_payload = {
|
|
"tool_call_id": tool_call_id,
|
|
"subagent_name": scope.agent_item.name,
|
|
"parent_thread_id": creator_run.conversation_thread_id,
|
|
"file_thread_id": file_thread_id or creator_run.conversation_thread_id,
|
|
"skills_thread_id": relation.child_thread_id,
|
|
}
|
|
input_payload = {
|
|
"model_spec": resolved_model_spec,
|
|
"runtime": {key: value for key, value in runtime_payload.items() if value is not None},
|
|
}
|
|
subagent_input_message = input_message.with_metadata(
|
|
{
|
|
"request_id": request_id,
|
|
"source": "subagent",
|
|
"raw_message": input_message.raw_message(),
|
|
}
|
|
)
|
|
persisted_input_message = await agent_run_service.create_agent_run_input_message(
|
|
db=self.db,
|
|
conversation_id=scope.conversation.id,
|
|
request_id=request_id,
|
|
input_message=subagent_input_message,
|
|
)
|
|
return await agent_run_service.persist_agent_run_record(
|
|
agent_slug=relation.subagent_slug,
|
|
conversation_thread_id=relation.child_thread_id,
|
|
current_uid=current_uid,
|
|
db=self.db,
|
|
request_id=request_id,
|
|
conversation_id=scope.conversation.id,
|
|
run_type="subagent",
|
|
input_payload=input_payload,
|
|
persisted_input_message=persisted_input_message,
|
|
created_by_run_id=creator_run.id,
|
|
subagent_thread_relation_id=relation.id,
|
|
)
|
|
|
|
async def _ensure_child_conversation(
|
|
self,
|
|
*,
|
|
child_thread_id: str,
|
|
uid: str,
|
|
agent_item: Agent,
|
|
creator_run: AgentRun,
|
|
):
|
|
"""确保子线程有对应 conversation；新线程会创建标记为 subagent 的对话。"""
|
|
conversation = await self.conv_repo.get_conversation_by_thread_id(child_thread_id)
|
|
if conversation:
|
|
if conversation.uid != str(uid) or conversation.status == "deleted":
|
|
raise ValueError("子智能体线程不存在")
|
|
if conversation.status != "subagent":
|
|
raise ValueError(f"子智能体线程 {child_thread_id} 已被普通对话占用")
|
|
if conversation.agent_id != agent_item.slug:
|
|
raise ValueError(f"子智能体线程 {child_thread_id} 属于智能体 {conversation.agent_id}")
|
|
return conversation
|
|
|
|
conversation = await self.conv_repo.add_conversation(
|
|
uid=uid,
|
|
agent_id=agent_item.slug,
|
|
title=f"SubAgent: {agent_item.name}",
|
|
thread_id=child_thread_id,
|
|
metadata={
|
|
"source": "subagent",
|
|
"parent_thread_id": creator_run.conversation_thread_id,
|
|
"created_by_run_id": creator_run.id,
|
|
"parent_conversation_id": creator_run.conversation_id,
|
|
"subagent_slug": agent_item.slug,
|
|
},
|
|
)
|
|
conversation.status = "subagent"
|
|
await self.db.flush()
|
|
return conversation
|
|
|
|
def _validate_thread_relation(
|
|
self,
|
|
relation: SubagentThread,
|
|
*,
|
|
child_thread_id: str,
|
|
agent_item: Agent,
|
|
creator_run: AgentRun,
|
|
) -> None:
|
|
"""校验已有子线程关系仍属于当前父对话和子智能体。"""
|
|
if relation.parent_conversation_id != creator_run.conversation_id:
|
|
raise ValueError(f"子智能体线程 {child_thread_id}：线程不属于当前对话")
|
|
if relation.subagent_slug != agent_item.slug:
|
|
raise ValueError(f"子智能体线程 {child_thread_id} 属于子智能体 {relation.subagent_slug or '未知'}")
|
|
|
|
async def _ensure_thread_relation(
|
|
self,
|
|
*,
|
|
child_thread_id: str,
|
|
uid: str,
|
|
agent_item: Agent,
|
|
creator_run: AgentRun,
|
|
continuing: bool,
|
|
) -> SubagentThread:
|
|
"""读取或创建父子线程关系；relation 是后台子 run 的线程归属来源。"""
|
|
existing = await self.thread_repo.get_by_child_thread_for_user(child_thread_id, uid)
|
|
if existing:
|
|
self._validate_thread_relation(
|
|
existing,
|
|
child_thread_id=child_thread_id,
|
|
agent_item=agent_item,
|
|
creator_run=creator_run,
|
|
)
|
|
return existing
|
|
if continuing:
|
|
raise ValueError(f"无法继续子智能体线程 {child_thread_id}：当前对话中没有找到对应的运行记录")
|
|
if creator_run.conversation_id is None:
|
|
raise ValueError("父运行任务缺少 conversation_id，无法创建子智能体线程关系")
|
|
|
|
child_conversation = await self._ensure_child_conversation(
|
|
child_thread_id=child_thread_id,
|
|
uid=uid,
|
|
agent_item=agent_item,
|
|
creator_run=creator_run,
|
|
)
|
|
return await self.thread_repo.create(
|
|
uid=uid,
|
|
parent_conversation_id=creator_run.conversation_id,
|
|
child_conversation_id=child_conversation.id,
|
|
child_thread_id=child_thread_id,
|
|
subagent_slug=agent_item.slug,
|
|
created_by_run_id=creator_run.id,
|
|
)
|