469 lines
19 KiB
Python
469 lines
19 KiB
Python
"""Task-analysis coordinator Q&A bridge (任务分析 · 总控问答链路).
|
||
|
||
对方详情页的问答入口：复用多智能体会商第二步（``multi_agent``）的总控派活/
|
||
席位执行后台，把一次问答会话记为一条 ``trigger=manual`` 的 run——过程消息
|
||
（总控用 ``__coordinator__`` 标识）落 ``task_analysis_run_messages``，问答结论
|
||
可经 apply-direction 沉淀为绑定的「执行方向备注」，注入后续每次定时执行。
|
||
|
||
接口：
|
||
- POST /tasks/{task_id}/qa/init 建席位 + 总控 thread，开一条 manual run（会话）
|
||
- POST /tasks/{task_id}/qa/stream 单轮问答：总控派活 → 席位执行 → 总控收口（SSE）
|
||
- POST /qa/{run_id}/apply-direction 问答结论沉淀方向备注并收口 run
|
||
- GET /external/runs/{run_id}/messages 问答历史查询（已有接口，seat 过滤可用）
|
||
|
||
设计口径（与需求定稿一致）：
|
||
- 每次问答会话新建一套席位 thread（跟随会商现状，不跨会话复用）；
|
||
- 任务上下文（名称/简介/总控提示/方向备注）在 init 时注入总控 thread，
|
||
之后每轮 stream 的 message 即用户原始问题；
|
||
- 一轮内：总控派活 → 被派席位**并行**执行（阶段内并发语义）→ 总控综合收口；
|
||
总控澄清 / 空派活（共识）时本轮提前结束。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
from collections.abc import AsyncIterator
|
||
from datetime import UTC, datetime
|
||
from typing import Any
|
||
|
||
from fastapi import APIRouter, HTTPException, Request
|
||
from fastapi.responses import StreamingResponse
|
||
from pydantic import BaseModel, Field
|
||
|
||
from app.gateway.routers._roundtable_seed import ensure_roundtable_functional_agents
|
||
from app.gateway.routers.multi_agent import (
|
||
COORDINATOR_AGENT_ID,
|
||
_append_thread_message,
|
||
_auth_headers,
|
||
_create_thread_direct,
|
||
_get_agents_batch_direct,
|
||
_leader_run,
|
||
_loopback_base,
|
||
_make_loopback_client,
|
||
_special_run,
|
||
)
|
||
from app.gateway.routers.task_analysis import _store, _task_or_404
|
||
from deerflow.persistence.task_analysis.base import TaskAnalysisStore
|
||
from deerflow.runtime.task_analysis.thread_metadata import task_analysis_thread_metadata
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter(prefix="/api/task-analysis", tags=["task-analysis-qa"])
|
||
|
||
COODINATOR_SEAT_KEY = "__coordinator__"
|
||
|
||
|
||
# ── helpers ──────────────────────────────────────────────────────────────
|
||
|
||
|
||
async def _binding_or_404(store: TaskAnalysisStore, task: dict[str, Any]) -> dict[str, Any]:
|
||
binding = await store.get_binding(task["id"])
|
||
if binding is None:
|
||
raise HTTPException(status_code=400, detail="任务未绑定分析方案")
|
||
return binding
|
||
|
||
|
||
def _seat_list(binding: dict[str, Any]) -> list[dict[str, str]]:
|
||
snapshot = binding.get("chain_snapshot") or {}
|
||
seats = snapshot.get("seats") or []
|
||
return [
|
||
{"agent_id": seat["agent_id"], "name": seat.get("name") or seat["agent_id"]}
|
||
for seat in seats
|
||
]
|
||
|
||
|
||
def _task_context_message(task: dict[str, Any], binding: dict[str, Any]) -> str:
|
||
parts = [
|
||
"【任务分析问答】当前任务上下文：",
|
||
f"外部任务：{task.get('name', '')}（任务号 {task.get('source_task_id', '')}）",
|
||
f"任务简介：{task.get('description') or '（无）'}",
|
||
f"协作编排：{binding.get('chainTitle') or binding.get('chain_id')}（同阶段并行 · 阶段间串行）",
|
||
]
|
||
prompt = binding.get("coordinator_prompt")
|
||
if prompt:
|
||
parts.append(f"总控提示：{prompt}")
|
||
notes = binding.get("direction_notes") or []
|
||
if notes:
|
||
parts.append("历次问答沉淀的执行方向备注：\n" + "\n".join(f"- {n}" for n in notes))
|
||
parts.append(
|
||
"用户将就该任务提问。你可以通过 agent_orchestration 派活给编排中的席位收集信息，"
|
||
"再基于席位交付回答；回答请贴合任务上下文。"
|
||
)
|
||
return "\n".join(parts)
|
||
|
||
|
||
def _sse_frame(payload: dict[str, Any]) -> str:
|
||
return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
|
||
|
||
|
||
async def _log_message(store: TaskAnalysisStore, run_id: str, seat: str, kind: str, content: str) -> None:
|
||
try:
|
||
await store.append_run_messages(
|
||
run_id, [{"seat_agent_id": seat, "stage_index": 0, "event_type": kind, "content": content}]
|
||
)
|
||
except Exception:
|
||
logger.exception("task-analysis qa message persist failed (run %s)", run_id)
|
||
|
||
|
||
# ── qa/init ──────────────────────────────────────────────────────────────
|
||
|
||
|
||
class QaInitResponse(BaseModel):
|
||
run_id: str
|
||
coordinator: str
|
||
thread_ids: dict[str, str]
|
||
seats: list[dict[str, str]]
|
||
|
||
|
||
@router.post("/tasks/{task_id}/qa/init", response_model=QaInitResponse)
|
||
async def qa_init(request: Request, task_id: str) -> QaInitResponse:
|
||
"""初始化问答会话：建席位/总控 thread + 注入任务上下文 + 开 manual run。"""
|
||
store = _store(request)
|
||
task = await _task_or_404(store, task_id=task_id)
|
||
binding = await _binding_or_404(store, task)
|
||
seats = _seat_list(binding)
|
||
if not seats:
|
||
raise HTTPException(status_code=400, detail="绑定链没有可调度席位")
|
||
|
||
ensure_roundtable_functional_agents()
|
||
agent_ids = [seat["agent_id"] for seat in seats]
|
||
|
||
# 席位 meta（名称/描述，用于总控清单注入与 leader seat_meta）。
|
||
agent_meta = await _get_agents_batch_direct(request, agent_ids)
|
||
|
||
# 并行建席位 thread + 总控 thread。
|
||
seat_threads = await _gather_thread_ids(request, task, agent_ids)
|
||
coord_thread = await _create_thread_direct(
|
||
request,
|
||
metadata=task_analysis_thread_metadata(
|
||
task,
|
||
thread_type="task-analysis-qa",
|
||
role="coordinator",
|
||
),
|
||
)
|
||
thread_ids: dict[str, str] = {**seat_threads, COORDINATOR_AGENT_ID: coord_thread}
|
||
|
||
# 席位清单 + 任务上下文注入总控 thread（两条 human 消息）。
|
||
seat_lines = [
|
||
f"- agent_name: {seat_id}（{agent_meta.get(seat_id, {}).get('name') or seat['name']}）："
|
||
f"{agent_meta.get(seat_id, {}).get('description') or ''}"
|
||
for seat in seats
|
||
for seat_id in [seat["agent_id"]]
|
||
]
|
||
base = _loopback_base(request)
|
||
headers = _auth_headers(request)
|
||
async with _make_loopback_client(request) as client:
|
||
await _append_thread_message(
|
||
client, base, headers, coord_thread,
|
||
"【圆桌系统初始化】本次问答可调度的席位清单如下,请严格按 agent_name 通过 "
|
||
"agent_orchestration 技能派活,不要派给不在清单内的 agent:\n" + "\n".join(seat_lines),
|
||
)
|
||
await _append_thread_message(client, base, headers, coord_thread, _task_context_message(task, binding))
|
||
|
||
# 问答会话 = 一条 manual run（stages 快照为空列表：问答不做阶段编排回放）。
|
||
run = await store.create_run_once(
|
||
{
|
||
"binding_id": binding["id"],
|
||
"external_task_id": task["id"],
|
||
"trigger": "manual",
|
||
"scheduled_for": datetime.now(UTC).replace(microsecond=0),
|
||
"status": "running",
|
||
"stages_snapshot": [],
|
||
"seat_names_snapshot": {seat["agent_id"]: seat["name"] for seat in seats},
|
||
"seat_results": {},
|
||
}
|
||
)
|
||
# 数据交付已去掉时间轴（期）维度：问答会话不再开新期，提交统一合并进
|
||
# 任务唯一的当前数据（首次提交时由 submit 兜底开期）。
|
||
await _log_message(store, run["id"], COODINATOR_SEAT_KEY, "input", _task_context_message(task, binding))
|
||
return QaInitResponse(
|
||
run_id=run["id"],
|
||
coordinator=COORDINATOR_AGENT_ID,
|
||
thread_ids=thread_ids,
|
||
seats=seats,
|
||
)
|
||
|
||
|
||
async def _gather_thread_ids(
|
||
request: Request,
|
||
task: dict[str, Any],
|
||
agent_ids: list[str],
|
||
) -> dict[str, str]:
|
||
import asyncio
|
||
|
||
async def _one(agent_id: str) -> tuple[str, str]:
|
||
thread_id = await _create_thread_direct(
|
||
request,
|
||
metadata=task_analysis_thread_metadata(
|
||
task,
|
||
thread_type="task-analysis-qa",
|
||
role="seat",
|
||
seat_id=agent_id,
|
||
),
|
||
)
|
||
return agent_id, thread_id
|
||
|
||
pairs = await asyncio.gather(*[_one(agent_id) for agent_id in agent_ids])
|
||
return dict(pairs)
|
||
|
||
|
||
# ── qa/stream：一轮问答（总控派活 → 席位执行 → 收口） ─────────────────────
|
||
|
||
|
||
class QaStreamRequest(BaseModel):
|
||
run_id: str
|
||
message: str = Field(min_length=1, max_length=8192)
|
||
# qa/init 返回的 thread 映射（每轮 stream 传回；缺失则要求重新 init）。
|
||
thread_ids: dict[str, str] | None = None
|
||
|
||
|
||
@router.post("/tasks/{task_id}/qa/stream")
|
||
async def qa_stream(request: Request, task_id: str, body: QaStreamRequest) -> StreamingResponse:
|
||
"""单轮问答编排（SSE）：透传总控/席位运行事件，消息落 run_messages。
|
||
|
||
事件帧（``data: {...}``）：
|
||
- ``{"event": "leader", "status": ...}`` 总控终态（派活表 / 共识 / 澄清）
|
||
- ``{"event": "seat_start", "agent": id}`` 某席位开始执行
|
||
- ``{"event": "seat", "agent": id, ...}`` 席位终态（content = 交付）
|
||
- ``{"event": "round_done", ...}`` 本轮收口（consensus=True 表示总控已给出结论）
|
||
"""
|
||
store = _store(request)
|
||
task = await _task_or_404(store, task_id=task_id)
|
||
binding = await _binding_or_404(store, task)
|
||
run = await store.get_run(body.run_id)
|
||
if run is None or run["external_task_id"] != task["id"]:
|
||
raise HTTPException(status_code=404, detail="问答会话不存在")
|
||
if run["status"] not in {"running", "pending"}:
|
||
raise HTTPException(status_code=409, detail="该问答会话已收口，请重新发起")
|
||
|
||
seats = _seat_list(binding)
|
||
seat_meta = {
|
||
seat["agent_id"]: {"name": seat["name"], "description": ""}
|
||
for seat in seats
|
||
}
|
||
# init 重建的 thread_ids 由调用方持有；run 行不落 thread 映射（敏感且体积大），
|
||
# 调用方需在每轮 stream 里传回 —— 兼容两种方式：body.thread_ids 或重新 init。
|
||
thread_ids: dict[str, str] = dict(body.thread_ids or {})
|
||
missing = [seat["agent_id"] for seat in seats if seat["agent_id"] not in thread_ids]
|
||
if missing or COORDINATOR_AGENT_ID not in thread_ids:
|
||
raise HTTPException(status_code=409, detail="会话 thread 映射缺失，请重新调用 qa/init")
|
||
|
||
async def _generate() -> AsyncIterator[str]:
|
||
base = _loopback_base(request)
|
||
headers = _auth_headers(request)
|
||
async with _make_loopback_client(request) as client:
|
||
await _log_message(store, body.run_id, COODINATOR_SEAT_KEY, "input", body.message)
|
||
|
||
# 1) 总控单轮。
|
||
leader_final: dict[str, Any] | None = None
|
||
async for chunk in _leader_run(
|
||
client,
|
||
base,
|
||
headers,
|
||
thread_ids,
|
||
COORDINATOR_AGENT_ID,
|
||
body.message,
|
||
skill_stop_names=[],
|
||
model_name="",
|
||
seat_meta=seat_meta,
|
||
):
|
||
if isinstance(chunk, dict):
|
||
leader_final = chunk
|
||
else:
|
||
yield chunk
|
||
if leader_final is None:
|
||
yield _sse_frame({"event": "round_done", "consensus": False, "reason": "leader_no_final"})
|
||
return
|
||
|
||
status = leader_final.get("status")
|
||
content = str(leader_final.get("content") or "")
|
||
yield _sse_frame({"event": "leader", **leader_final})
|
||
if content:
|
||
await _log_message(store, body.run_id, COODINATOR_SEAT_KEY, "llm.ai", content)
|
||
|
||
# 澄清 / 共识：本轮结束。澄清 = 等用户补充（不收口 run）；其他字符串终态 = 共识收口。
|
||
if not isinstance(status, list):
|
||
if str(status) == "clarification":
|
||
yield _sse_frame({"event": "round_done", "consensus": False, "kind": "clarification"})
|
||
return
|
||
await _finish_qa_run(store, body.run_id, content or "（总控给出结论）")
|
||
yield _sse_frame({"event": "round_done", "consensus": True, "kind": str(status)})
|
||
return
|
||
|
||
dispatched = [item for item in status if isinstance(item, list) and len(item) >= 2]
|
||
if not dispatched:
|
||
await _finish_qa_run(store, body.run_id, content or "（共识）")
|
||
yield _sse_frame({"event": "round_done", "consensus": True})
|
||
return
|
||
|
||
# 2) 被派活席位并行执行（阶段内并发语义）。
|
||
import asyncio
|
||
|
||
async def _run_one(agent_id: str, task_text: str) -> tuple[str, dict[str, Any] | None]:
|
||
await store.update_run(
|
||
body.run_id,
|
||
{"seat_results": await _patch_seat(store, run, agent_id, "running")},
|
||
)
|
||
final: dict[str, Any] | None = None
|
||
async for chunk in _special_run(
|
||
client,
|
||
base,
|
||
headers,
|
||
thread_ids,
|
||
agent_id,
|
||
task_text,
|
||
skill_stop_names=[],
|
||
model_name="",
|
||
display_name=seat_meta.get(agent_id, {}).get("name", agent_id),
|
||
):
|
||
if isinstance(chunk, dict):
|
||
final = chunk
|
||
# 席位 SSE 行不透传（体量大）；前端通过 seat_start/seat 终态感知。
|
||
return agent_id, final
|
||
|
||
tasks = [
|
||
asyncio.create_task(_run_one(str(item[0]), str(item[1])))
|
||
for item in dispatched
|
||
]
|
||
for coro in asyncio.as_completed(tasks):
|
||
agent_id, final = await coro
|
||
delivery = str((final or {}).get("content") or "")
|
||
yield _sse_frame({
|
||
"event": "seat",
|
||
"agent": agent_id,
|
||
"name": seat_meta.get(agent_id, {}).get("name", agent_id),
|
||
"content": delivery,
|
||
"status": str((final or {}).get("status") or ""),
|
||
})
|
||
if delivery:
|
||
await _log_message(store, body.run_id, agent_id, "llm.ai", delivery)
|
||
fresh = await store.get_run(body.run_id)
|
||
await store.update_run(
|
||
body.run_id,
|
||
{
|
||
"seat_results": await _patch_seat(
|
||
store, fresh or run, agent_id,
|
||
"done" if delivery else "failed",
|
||
delivery=delivery,
|
||
)
|
||
},
|
||
)
|
||
|
||
# 3) 总控收口（综合轮）。
|
||
wrap_up = (
|
||
"所有被派活的席位都已交付（完整交付已注入）。"
|
||
"请基于以上交付回答用户的问题，给出明确结论；如需继续追问某席位可再次派活。"
|
||
)
|
||
synthesis_final: dict[str, Any] | None = None
|
||
async for chunk in _leader_run(
|
||
client,
|
||
base,
|
||
headers,
|
||
thread_ids,
|
||
COORDINATOR_AGENT_ID,
|
||
wrap_up,
|
||
skill_stop_names=[],
|
||
model_name="",
|
||
seat_meta=seat_meta,
|
||
synthesis_mode=True,
|
||
):
|
||
if isinstance(chunk, dict):
|
||
synthesis_final = chunk
|
||
else:
|
||
yield chunk
|
||
final_status = (synthesis_final or {}).get("status")
|
||
final_content = str((synthesis_final or {}).get("content") or "")
|
||
yield _sse_frame({"event": "leader", **(synthesis_final or {})})
|
||
if final_content:
|
||
await _log_message(store, body.run_id, COODINATOR_SEAT_KEY, "llm.ai", final_content)
|
||
consensus = not isinstance(final_status, list) or not final_status
|
||
if consensus:
|
||
await _finish_qa_run(store, body.run_id, final_content or content or "（总控收口）")
|
||
yield _sse_frame({"event": "round_done", "consensus": consensus})
|
||
|
||
return StreamingResponse(_generate(), media_type="text/event-stream")
|
||
|
||
|
||
async def _patch_seat(
|
||
store: TaskAnalysisStore,
|
||
run: dict[str, Any],
|
||
agent_id: str,
|
||
state: str,
|
||
*,
|
||
delivery: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""读-改-写 run.seat_results 的单席位条目（问答轮的席位状态）。"""
|
||
fresh = await store.get_run(run["id"])
|
||
results: dict[str, Any] = dict((fresh or run).get("seat_results") or {})
|
||
existing = results.get(agent_id) or {}
|
||
results[agent_id] = {
|
||
"state": state,
|
||
"duration_ms": existing.get("duration_ms", 0),
|
||
"thread_id": None,
|
||
"seat_name": existing.get("seat_name") or agent_id,
|
||
"delivery": delivery if delivery is not None else existing.get("delivery"),
|
||
"error": None,
|
||
}
|
||
return results
|
||
|
||
|
||
async def _finish_qa_run(store: TaskAnalysisStore, run_id: str, summary: str) -> None:
|
||
await store.update_run(
|
||
run_id,
|
||
{
|
||
"status": "succeeded",
|
||
"finished_at": datetime.now(UTC).isoformat(),
|
||
"error": None,
|
||
"seat_results": await _finalize_seats(store, run_id),
|
||
},
|
||
)
|
||
_ = summary
|
||
|
||
|
||
async def _finalize_seats(store: TaskAnalysisStore, run_id: str) -> dict[str, Any]:
|
||
run = await store.get_run(run_id)
|
||
results: dict[str, Any] = dict((run or {}).get("seat_results") or {})
|
||
for agent_id, item in results.items():
|
||
if item.get("state") == "running":
|
||
results[agent_id] = {**item, "state": "done"}
|
||
return results
|
||
|
||
|
||
# ── apply-direction：问答结论沉淀为方向备注 ───────────────────────────────
|
||
|
||
|
||
class ApplyDirectionRequest(BaseModel):
|
||
note: str = Field(min_length=1, max_length=2048)
|
||
|
||
|
||
@router.post("/qa/{run_id}/apply-direction")
|
||
async def apply_direction(request: Request, run_id: str, body: ApplyDirectionRequest) -> dict:
|
||
"""把本轮问答结论沉淀为绑定的「执行方向备注」并收口问答 run。
|
||
|
||
备注会注入该绑定后续每次定时执行的席位 prompt（见 executor `_build_seat_prompt`）。
|
||
"""
|
||
store = _store(request)
|
||
run = await store.get_run(run_id)
|
||
if run is None:
|
||
raise HTTPException(status_code=404, detail="问答会话不存在")
|
||
binding = await store.get_binding_by_id(run["binding_id"])
|
||
if binding is None:
|
||
raise HTTPException(status_code=404, detail="绑定不存在")
|
||
|
||
notes = [*list(binding.get("direction_notes") or []), body.note.strip()]
|
||
await store.update_binding(binding["id"], {"direction_notes": notes})
|
||
await _log_message(
|
||
store, run_id, COODINATOR_SEAT_KEY, "artifact",
|
||
f"【方向备注已应用】{body.note.strip()}（将注入后续每次定时执行）",
|
||
)
|
||
await store.update_run(
|
||
run_id,
|
||
{
|
||
"status": "succeeded",
|
||
"finished_at": datetime.now(UTC).isoformat(),
|
||
"seat_results": await _finalize_seats(store, run_id),
|
||
},
|
||
)
|
||
return {"applied": True, "direction_notes": notes}
|