1170 lines
45 KiB
Python
1170 lines
45 KiB
Python
"""Gateway API for the AgentScope report-collaboration workbench.
|
|
|
|
Plan/run writes enqueue durable jobs and return immediately. AgentScope
|
|
execution is not started from this router. ``GET /runs/{id}/stream`` is a
|
|
read-only durable SSE tail: disconnect does not cancel the run.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import re
|
|
from collections.abc import AsyncIterator
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Header, HTTPException, Query, Request
|
|
from fastapi.responses import JSONResponse, Response, StreamingResponse
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.gateway.deps import get_current_user
|
|
from app.report_collaboration.agentscope_runtime.model_resolver import freeze_plan_models
|
|
from app.report_collaboration.contracts.requirements import ReportRequirementSnapshot
|
|
from app.report_collaboration.contracts.snapshots import SessionSnapshot
|
|
from app.report_collaboration.execution.live_hub import iter_run_sse, replay_cursor
|
|
from app.report_collaboration.interventions.intent_router import build_runtime_intent_router
|
|
from app.report_collaboration.interventions.service import RuntimeCommandService
|
|
from app.report_collaboration.planning import PlanController, PlanProposalService, project_agent_catalog, requirement_from_session
|
|
from app.report_collaboration.planning.planner_agent import PlannerAgent, _extract_json
|
|
from app.report_collaboration.reporting import ConservativeRewriteGenerator, ModelRewriteGenerator, ReportRewriteService, freeze_report_template
|
|
from app.report_collaboration.reporting.templates import requirement_snapshot_from_record
|
|
from app.report_collaboration.security import freeze_run_budget
|
|
from deerflow.config.app_config import get_app_config
|
|
from deerflow.persistence.report_collaboration import (
|
|
ReportCollaborationConflictError,
|
|
ReportCollaborationLimitError,
|
|
ReportCollaborationNotFoundError,
|
|
ReportCollaborationStore,
|
|
ReportCollaborationValidationError,
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/report-collaboration", tags=["report-collaboration"])
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class CreateSessionBody(BaseModel):
|
|
title: str | None = Field(default=None, max_length=255)
|
|
|
|
|
|
class UpdateSessionBody(BaseModel):
|
|
title: str | None = Field(default=None, max_length=255)
|
|
|
|
|
|
class SendMessageBody(BaseModel):
|
|
text: str = Field(..., min_length=1, max_length=20_000)
|
|
target_node_id: str | None = None
|
|
agent_run_id: str | None = None
|
|
|
|
|
|
class CompleteIntentTurnBody(BaseModel):
|
|
"""One completed turn from the real streaming intent-agent thread."""
|
|
|
|
text: str = Field(..., min_length=1, max_length=20_000)
|
|
assistant_content: str = Field(default="", max_length=100_000)
|
|
reasoning_content: str | None = Field(default=None, max_length=100_000)
|
|
intent_thread_id: str = Field(..., min_length=1, max_length=128)
|
|
status: str = Field(..., pattern="^(asking|clarification|done)$")
|
|
summary: dict[str, Any] | None = None
|
|
clarifications: list[dict[str, Any]] = Field(default_factory=list, max_length=8)
|
|
|
|
|
|
class StreamPlanBody(BaseModel):
|
|
adjustment_text: str | None = Field(default=None, max_length=20_000)
|
|
|
|
|
|
class StartRunBody(BaseModel):
|
|
plan_id: str = Field(..., min_length=1, max_length=64)
|
|
|
|
|
|
class CreateCommandBody(BaseModel):
|
|
text: str = Field(..., min_length=1, max_length=20_000)
|
|
target_node_id: str | None = None
|
|
agent_run_id: str | None = None
|
|
|
|
|
|
class CreateRewriteBody(BaseModel):
|
|
instruction: str = Field(..., min_length=1, max_length=20_000)
|
|
section: str | None = None
|
|
|
|
|
|
def _store(request: Request) -> ReportCollaborationStore:
|
|
store = getattr(request.app.state, "report_collaboration_store", None)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="Report collaboration store is unavailable")
|
|
return store
|
|
|
|
|
|
_THINK_BLOCK = re.compile(r"<think>(.*?)</think>", re.DOTALL | re.IGNORECASE)
|
|
|
|
|
|
def _intent_display_content(content: str) -> tuple[str, str | None]:
|
|
"""Separate visible answer/reasoning and hide the machine hand-off block."""
|
|
|
|
reasoning = "\n\n".join(part.strip() for part in _THINK_BLOCK.findall(content) if part.strip())
|
|
visible = _THINK_BLOCK.sub("", content)
|
|
visible = visible.split("[INTENT_READY]", 1)[0].strip()
|
|
return visible, reasoning or None
|
|
|
|
|
|
def _planner_chunk_parts(chunk: Any) -> tuple[str, str]:
|
|
raw_content = getattr(chunk, "content", None)
|
|
if isinstance(raw_content, str):
|
|
content = raw_content
|
|
elif isinstance(raw_content, list):
|
|
content = "".join(
|
|
str(block.get("text") or "")
|
|
if isinstance(block, dict)
|
|
else str(getattr(block, "text", "") or "")
|
|
for block in raw_content
|
|
)
|
|
else:
|
|
content = ""
|
|
additional = getattr(chunk, "additional_kwargs", None) or {}
|
|
reasoning = ""
|
|
if isinstance(additional, dict):
|
|
value = additional.get("reasoning_content") or additional.get("reasoning")
|
|
if isinstance(value, str):
|
|
reasoning = value
|
|
return content, reasoning
|
|
|
|
|
|
def _planner_sse(event: str, data: dict[str, Any]) -> str:
|
|
return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False, default=str)}\n\n"
|
|
|
|
|
|
def _planner_inline_reasoning(text: str) -> str:
|
|
"""Return reasoning emitted inside `<think>` blocks, including an open block."""
|
|
|
|
parts = list(_THINK_BLOCK.findall(text))
|
|
last_open = text.lower().rfind("<think>")
|
|
last_close = text.lower().rfind("</think>")
|
|
if last_open > last_close:
|
|
parts.append(text[last_open + len("<think>") :])
|
|
return "".join(parts)
|
|
|
|
|
|
def _report_requirement(summary: dict[str, Any], revision: int) -> ReportRequirementSnapshot:
|
|
raw = summary.get("report_requirement") or summary.get("reportRequirement")
|
|
if not isinstance(raw, dict):
|
|
objective = str(summary.get("objective") or "").strip()
|
|
raw = {
|
|
"topic": objective,
|
|
"objective": objective,
|
|
"required_angles": [],
|
|
"excluded_angles": [],
|
|
}
|
|
payload = dict(raw)
|
|
payload["revision"] = revision
|
|
payload.setdefault("required_angles", [])
|
|
payload.setdefault("excluded_angles", [])
|
|
return ReportRequirementSnapshot.model_validate(payload)
|
|
|
|
|
|
def _runtime_commands(request: Request) -> RuntimeCommandService:
|
|
service = getattr(request.app.state, "report_collaboration_command_service", None)
|
|
if service is None:
|
|
app_config = get_app_config()
|
|
service = RuntimeCommandService(
|
|
_store(request),
|
|
router=build_runtime_intent_router(app_config),
|
|
config=app_config.report_collaboration,
|
|
)
|
|
request.app.state.report_collaboration_command_service = service
|
|
return service
|
|
|
|
|
|
def _require_enabled() -> JSONResponse | None:
|
|
if not get_app_config().report_collaboration.enabled:
|
|
return _error(503, "REPORT_COLLABORATION_DISABLED", "报告协作工作台未启用")
|
|
return None
|
|
|
|
|
|
async def _require_user(request: Request) -> str:
|
|
user_id = await get_current_user(request)
|
|
if not user_id:
|
|
raise HTTPException(status_code=401, detail="Authentication required")
|
|
return str(user_id)
|
|
|
|
|
|
def _error(status: int, code: str, message: str, detail: Any = None) -> JSONResponse:
|
|
body: dict[str, Any] = {"code": code, "message": message}
|
|
if detail is not None:
|
|
body["detail"] = detail
|
|
return JSONResponse(status_code=status, content=body)
|
|
|
|
|
|
def _require_idempotency(idempotency_key: str | None) -> str | JSONResponse:
|
|
key = (idempotency_key or "").strip()
|
|
if not key:
|
|
return _error(422, "IDEMPOTENCY_KEY_REQUIRED", "写操作必须携带 X-Idempotency-Key")
|
|
if len(key) > 128:
|
|
return _error(422, "IDEMPOTENCY_KEY_INVALID", "X-Idempotency-Key 过长")
|
|
return key
|
|
|
|
|
|
def _map_store_error(exc: Exception) -> JSONResponse:
|
|
if isinstance(exc, ReportCollaborationNotFoundError):
|
|
return _error(404, "NOT_FOUND", str(exc))
|
|
if isinstance(exc, ReportCollaborationConflictError):
|
|
return _error(409, exc.code, str(exc), detail={"current_revision": exc.current_revision})
|
|
if isinstance(exc, ReportCollaborationValidationError):
|
|
return _error(422, exc.code, str(exc))
|
|
if isinstance(exc, ReportCollaborationLimitError):
|
|
return _error(429, exc.code, str(exc), detail={"usage": exc.usage, "limits": exc.limits})
|
|
raise exc
|
|
|
|
|
|
def _rewrite_service(store: ReportCollaborationStore) -> ReportRewriteService:
|
|
cfg = get_app_config().report_collaboration
|
|
generator = ModelRewriteGenerator(cfg.writer_model) if cfg.writer_model else ConservativeRewriteGenerator()
|
|
return ReportRewriteService(store, generator=generator)
|
|
|
|
|
|
async def _frozen_run_template(request: Request, session: dict[str, Any], plan: dict[str, Any]) -> dict[str, Any]:
|
|
requirement = requirement_snapshot_from_record(session)
|
|
structure_id = plan.get("report_structure_id") or (requirement.report_structure_id if requirement else None)
|
|
structure = None
|
|
structure_store = getattr(request.app.state, "report_structure_store", None)
|
|
if structure_id and structure_store is not None:
|
|
structure = await structure_store.get_structure(str(structure_id))
|
|
return freeze_report_template(structure, requirement=requirement).model_dump(mode="json")
|
|
|
|
|
|
async def _owned_session(store: ReportCollaborationStore, session_id: str, user_id: str) -> dict[str, Any] | JSONResponse:
|
|
record = await store.get_session_record(session_id)
|
|
if record is None or record.get("owner_id") != user_id:
|
|
return _error(404, "NOT_FOUND", f"report collaboration session not found: {session_id}")
|
|
return record
|
|
|
|
|
|
async def _owned_run(store: ReportCollaborationStore, run_id: str, user_id: str) -> dict[str, Any] | JSONResponse:
|
|
run = await store.get_run(run_id)
|
|
if run is None:
|
|
return _error(404, "NOT_FOUND", f"report collaboration run not found: {run_id}")
|
|
owned = await _owned_session(store, run["session_id"], user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
return run
|
|
|
|
|
|
@router.post("/sessions")
|
|
async def create_session(
|
|
request: Request,
|
|
body: CreateSessionBody | None = None,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
payload = body or CreateSessionBody()
|
|
try:
|
|
return await _store(request).create_session(owner_id=user_id, title=payload.title or "", idempotency_key=key)
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.get("/sessions")
|
|
async def list_sessions(request: Request):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
sessions = await _store(request).list_sessions(user_id)
|
|
return {"sessions": sessions}
|
|
|
|
|
|
@router.get("/sessions/{session_id}")
|
|
async def get_session_snapshot(request: Request, session_id: str):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
snapshot = await store.get_snapshot(session_id)
|
|
return SessionSnapshot.model_validate(snapshot).model_dump()
|
|
|
|
|
|
@router.patch("/sessions/{session_id}")
|
|
async def update_session(
|
|
request: Request,
|
|
session_id: str,
|
|
body: UpdateSessionBody,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
x_expected_revision: int | None = Header(default=None, alias="X-Expected-Revision"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
try:
|
|
updated = await store.update_session(session_id, title=body.title, expected_revision=x_expected_revision)
|
|
await store.create_command(session_id, idempotency_key=key, text=body.title or "", operation="update_session", status="completed")
|
|
return updated
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.delete("/sessions/{session_id}")
|
|
async def delete_session(request: Request, session_id: str):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
await store.delete_session(session_id)
|
|
return Response(status_code=204)
|
|
|
|
|
|
@router.get("/sessions/{session_id}/messages")
|
|
async def list_messages(request: Request, session_id: str):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
return {"messages": await store.list_messages(session_id)}
|
|
|
|
|
|
@router.post("/sessions/{session_id}/messages")
|
|
async def send_message(
|
|
request: Request,
|
|
session_id: str,
|
|
body: SendMessageBody,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
metadata: dict[str, Any] = {"idempotency_key": key}
|
|
if body.target_node_id:
|
|
metadata["target_node_id"] = body.target_node_id
|
|
if body.agent_run_id:
|
|
metadata["agent_run_id"] = body.agent_run_id
|
|
try:
|
|
existing = await store.get_message_by_idempotency(key)
|
|
if existing is not None:
|
|
return existing
|
|
snapshot = await store.get_snapshot(session_id)
|
|
active = snapshot.get("run") or {}
|
|
if str(active.get("status") or "") in {
|
|
"queued",
|
|
"planning",
|
|
"running",
|
|
"awaiting_input",
|
|
"reviewing",
|
|
"recovering",
|
|
}:
|
|
return _error(
|
|
409,
|
|
"RUN_ACTIVE",
|
|
"运行进行中请使用 /runs/{run_id}/commands 发送干预，不要走规划前消息接口",
|
|
detail={"run_id": active.get("id"), "status": active.get("status")},
|
|
)
|
|
if str(owned.get("status") or "") in {"empty", "clarifying"}:
|
|
return _error(
|
|
409,
|
|
"USE_STREAMING_INTENT",
|
|
"需求确认阶段必须调用真实意图智能体流，不能使用阻塞消息接口",
|
|
)
|
|
created = await store.create_message(
|
|
session_id,
|
|
role="human",
|
|
content=body.text,
|
|
idempotency_key=key,
|
|
metadata=metadata,
|
|
agent_run_id=body.agent_run_id,
|
|
node_run_id=body.target_node_id,
|
|
)
|
|
return created
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.post("/sessions/{session_id}/intent-turns")
|
|
async def complete_intent_turn(
|
|
request: Request,
|
|
session_id: str,
|
|
body: CompleteIntentTurnBody,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
):
|
|
"""Persist one streamed agent turn and auto-plan after its READY frame.
|
|
|
|
No model is invoked for intent recognition here: the browser has already
|
|
consumed the real `/api/intent/stream` run incrementally.
|
|
"""
|
|
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
try:
|
|
existing_user = await store.get_message_by_idempotency(key)
|
|
if existing_user is None:
|
|
await store.create_message(
|
|
session_id,
|
|
role="human",
|
|
content=body.text,
|
|
idempotency_key=key,
|
|
metadata={"intent_thread_id": body.intent_thread_id, "source": "intent_agent_stream"},
|
|
)
|
|
|
|
visible, inline_reasoning = _intent_display_content(body.assistant_content)
|
|
reasoning = body.reasoning_content or inline_reasoning
|
|
tool_calls = [
|
|
{
|
|
"id": str(item.get("id") or f"clarification-{index}"),
|
|
"name": "ask_clarification",
|
|
"args": item,
|
|
}
|
|
for index, item in enumerate(body.clarifications)
|
|
]
|
|
await store.create_message(
|
|
session_id,
|
|
role="ai",
|
|
content=visible,
|
|
idempotency_key=f"{key}:intent-ai",
|
|
name="协作意图规划师",
|
|
metadata={
|
|
"intent_thread_id": body.intent_thread_id,
|
|
"intent_status": body.status,
|
|
**({"reasoning_content": reasoning} if reasoning else {}),
|
|
},
|
|
tool_calls=tool_calls,
|
|
completed=True,
|
|
)
|
|
# MessageList renders an actionable clarification card from a real
|
|
# `tool` message, not from the assistant's tool-call trace alone.
|
|
# Mirror every streamed ask_clarification result into the report
|
|
# collaboration timeline so it survives refresh and is answerable.
|
|
for index, item in enumerate(body.clarifications):
|
|
call_id = str(item.get("id") or f"clarification-{index}")
|
|
question = str(item.get("question") or "请补充报告需求。")
|
|
clarification = {
|
|
"question": question,
|
|
"clarification_type": str(
|
|
item.get("clarification_type") or item.get("clarificationType") or ""
|
|
),
|
|
"context": str(item.get("context") or item.get("clarification_context") or ""),
|
|
"options": list(item.get("options") or []),
|
|
"allow_custom": item.get("allow_custom", item.get("allowCustom", True)) is not False,
|
|
"allow_multiple": item.get("allow_multiple", item.get("allowMultiple")),
|
|
}
|
|
await store.create_message(
|
|
session_id,
|
|
role="tool",
|
|
name="ask_clarification",
|
|
content=f"❓ {question}",
|
|
tool_call_id=call_id,
|
|
idempotency_key=f"{key}:intent-clarification:{index}",
|
|
metadata={"clarification": clarification, "intent_thread_id": body.intent_thread_id},
|
|
completed=True,
|
|
)
|
|
|
|
record = await store.get_session_record(session_id)
|
|
if record is None:
|
|
return _error(404, "NOT_FOUND", "会话不存在")
|
|
raw_state = record.get("requirement_json")
|
|
current_state = dict(raw_state) if isinstance(raw_state, dict) else {}
|
|
current_revision = int(record.get("requirement_revision") or 0)
|
|
turn_already_applied = current_state.get("last_intent_turn_key") == key
|
|
if turn_already_applied:
|
|
existing_plans = await store.list_plans(session_id)
|
|
if body.status != "done" or existing_plans:
|
|
return await store.get_snapshot(session_id)
|
|
|
|
if body.status != "done" or not isinstance(body.summary, dict):
|
|
await store.save_requirement_state(
|
|
session_id,
|
|
requirement_json={
|
|
**current_state,
|
|
"intent_thread_id": body.intent_thread_id,
|
|
"ready_to_plan": False,
|
|
"last_intent_turn_key": key,
|
|
},
|
|
requirement_revision=current_revision,
|
|
status="clarifying",
|
|
)
|
|
return await store.get_snapshot(session_id)
|
|
|
|
requirement = _report_requirement(
|
|
body.summary,
|
|
current_revision if turn_already_applied else current_revision + 1,
|
|
)
|
|
await store.save_requirement_state(
|
|
session_id,
|
|
requirement_json={
|
|
"snapshot": requirement.model_dump(),
|
|
"clarifications": list(current_state.get("clarifications") or []),
|
|
"assumptions": list(body.summary.get("assumptions") or []),
|
|
"clarification_round": int(current_state.get("clarification_round") or 0),
|
|
"ready_to_plan": True,
|
|
"intent_thread_id": body.intent_thread_id,
|
|
"last_intent_turn_key": key,
|
|
},
|
|
requirement_revision=requirement.revision,
|
|
status="planning",
|
|
title=requirement.topic[:40],
|
|
)
|
|
return await store.get_snapshot(session_id)
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.post("/sessions/{session_id}/plan-stream")
|
|
async def stream_plans(
|
|
request: Request,
|
|
session_id: str,
|
|
body: StreamPlanBody,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
):
|
|
"""Run the workflow-planning agent as a real SSE stream, then persist plans."""
|
|
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
if owned.get("selected_plan_id"):
|
|
return _error(409, "PLAN_ALREADY_SELECTED", "会话已选定方案，无法重新规划")
|
|
|
|
try:
|
|
command = await store.enqueue_plan_request(
|
|
session_id,
|
|
idempotency_key=key,
|
|
expected_revision=int(owned.get("requirement_revision") or 0),
|
|
)
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
if command.get("status") == "completed":
|
|
|
|
async def replay_completed() -> AsyncIterator[str]:
|
|
yield _planner_sse("complete", {"snapshot": await store.get_snapshot(session_id)})
|
|
|
|
return StreamingResponse(replay_completed(), media_type="text/event-stream")
|
|
if command.get("status") != "pending":
|
|
return _error(409, "PLAN_REQUEST_NOT_PENDING", "本次规划请求已结束，请重新发起")
|
|
|
|
async def generate() -> AsyncIterator[str]:
|
|
service = PlanProposalService(
|
|
store,
|
|
agent_store=getattr(request.app.state, "agent_store", None),
|
|
config=get_app_config().report_collaboration,
|
|
llm_factory=getattr(request.app.state, "report_collaboration_planner_llm", None),
|
|
)
|
|
planner_reasoning = ""
|
|
planner_steps = [
|
|
{
|
|
"id": f"{command['command_id']}:catalog",
|
|
"name": "workflow_planning_progress",
|
|
"args": {"description": "正在读取可用智能体及能力描述"},
|
|
},
|
|
{
|
|
"id": f"{command['command_id']}:design",
|
|
"name": "workflow_planning_progress",
|
|
"args": {"description": "规划智能体正在设计候选工作流"},
|
|
},
|
|
{
|
|
"id": f"{command['command_id']}:validate",
|
|
"name": "workflow_planning_progress",
|
|
"args": {"description": "正在校验依赖关系并绑定智能体"},
|
|
},
|
|
]
|
|
try:
|
|
yield _planner_sse("phase", {"id": "catalog", "label": "正在读取可用智能体及能力描述"})
|
|
record = await store.get_session_record(session_id)
|
|
if record is None:
|
|
raise ReportCollaborationValidationError("NOT_FOUND", "会话不存在")
|
|
requirement = requirement_from_session(record)
|
|
cfg = get_app_config().report_collaboration
|
|
catalog = await project_agent_catalog(
|
|
user_id=user_id,
|
|
config=cfg,
|
|
agent_store=getattr(request.app.state, "agent_store", None),
|
|
)
|
|
history = [
|
|
{"role": "human" if item.get("role") == "human" else "ai", "content": str(item.get("content") or "")}
|
|
for item in await store.list_messages(session_id)
|
|
]
|
|
current_plans = [
|
|
item
|
|
for item in await store.list_plans(session_id)
|
|
if item.get("status") in {"proposed", "selected"}
|
|
]
|
|
llm = service.planner_llm()
|
|
binder = getattr(llm, "bind", None)
|
|
if callable(binder):
|
|
llm = binder(max_tokens=cfg.planner_max_tokens)
|
|
planner = PlannerAgent(llm)
|
|
messages = planner.workflow_messages(
|
|
requirement=requirement,
|
|
history=history,
|
|
catalog=catalog,
|
|
current_plans=current_plans,
|
|
adjustment_text=body.adjustment_text,
|
|
max_team_members=cfg.max_team_members,
|
|
)
|
|
yield _planner_sse("phase", {"id": "design", "label": "规划智能体正在设计候选工作流"})
|
|
raw = ""
|
|
inline_reasoning_length = 0
|
|
planner_stream_error: Exception | None = None
|
|
logger.info(
|
|
"rc planner stream started session=%s timeout=%ss max_tokens=%s agents=%s",
|
|
session_id,
|
|
cfg.planner_timeout_seconds,
|
|
cfg.planner_max_tokens,
|
|
len(catalog.agents),
|
|
)
|
|
try:
|
|
async with asyncio.timeout(cfg.planner_timeout_seconds):
|
|
async for chunk in llm.astream(messages):
|
|
content_delta, reasoning_delta = _planner_chunk_parts(chunk)
|
|
raw += content_delta
|
|
if reasoning_delta:
|
|
planner_reasoning += reasoning_delta
|
|
yield _planner_sse("reasoning", {"delta": reasoning_delta})
|
|
inline_reasoning = _planner_inline_reasoning(raw)
|
|
if len(inline_reasoning) > inline_reasoning_length:
|
|
inline_delta = inline_reasoning[inline_reasoning_length:]
|
|
planner_reasoning += inline_delta
|
|
yield _planner_sse(
|
|
"reasoning",
|
|
{"delta": inline_delta},
|
|
)
|
|
inline_reasoning_length = len(inline_reasoning)
|
|
except TimeoutError as exc:
|
|
planner_stream_error = exc
|
|
logger.warning(
|
|
"rc planner stream timed out session=%s visible_chars=%s reasoning_chars=%s",
|
|
session_id,
|
|
len(raw),
|
|
len(planner_reasoning),
|
|
)
|
|
except Exception as exc: # noqa: BLE001 - recover into executable plans below
|
|
planner_stream_error = exc
|
|
logger.warning(
|
|
"rc planner stream failed session=%s error=%s: %s",
|
|
session_id,
|
|
type(exc).__name__,
|
|
exc,
|
|
)
|
|
|
|
# Some reasoning providers place their final JSON in
|
|
# reasoning_content instead of the visible content channel. Try
|
|
# both channels, including truncated-output salvage.
|
|
parsed = _extract_json(raw)
|
|
if parsed is None:
|
|
parsed = _extract_json(planner_reasoning)
|
|
if parsed is None:
|
|
parsed = _extract_json(f"{raw}\n{planner_reasoning}")
|
|
designed = planner.parse_workflow_design(parsed)
|
|
if designed is None:
|
|
recovery_label = "规划输出较长，正在收敛为可执行候选工作流"
|
|
planner_steps.append(
|
|
{
|
|
"id": f"{command['command_id']}:recover",
|
|
"name": "workflow_planning_progress",
|
|
"args": {"description": recovery_label},
|
|
}
|
|
)
|
|
yield _planner_sse("phase", {"id": "recover", "label": recovery_label})
|
|
logger.warning(
|
|
"rc planner output could not be parsed; using constrained recovery session=%s visible_chars=%s reasoning_chars=%s stream_error=%s",
|
|
session_id,
|
|
len(raw),
|
|
len(planner_reasoning),
|
|
type(planner_stream_error).__name__ if planner_stream_error else "none",
|
|
)
|
|
strategies = PlanController().propose(
|
|
requirement,
|
|
max_parallel_tasks=cfg.max_parallel_tasks,
|
|
max_team_members=cfg.max_team_members,
|
|
)
|
|
designed = (
|
|
strategies,
|
|
"规划智能体已依据当前意图和可用智能体目录收敛出三套可执行候选工作流，可继续通过对话调整。",
|
|
)
|
|
else:
|
|
logger.info(
|
|
"rc planner stream completed session=%s visible_chars=%s reasoning_chars=%s candidates=%s",
|
|
session_id,
|
|
len(raw),
|
|
len(planner_reasoning),
|
|
len(designed[0]),
|
|
)
|
|
yield _planner_sse("phase", {"id": "validate", "label": "正在校验依赖关系并绑定智能体"})
|
|
await service.fulfill(
|
|
session_id,
|
|
user_id=user_id,
|
|
command_id=command["command_id"],
|
|
idempotency_key=key,
|
|
adjustment_text=body.adjustment_text,
|
|
precomputed_design=designed,
|
|
planner_reasoning=planner_reasoning or None,
|
|
planner_steps=planner_steps,
|
|
)
|
|
snapshot = await store.get_snapshot(session_id)
|
|
yield _planner_sse("complete", {"snapshot": snapshot})
|
|
except asyncio.CancelledError as exc:
|
|
await service.fail_command(session_id, command["command_id"], exc)
|
|
raise
|
|
except Exception as exc: # noqa: BLE001 - SSE errors must be returned in-band
|
|
await service.fail_command(session_id, command["command_id"], exc)
|
|
yield _planner_sse("error", {"message": str(exc)})
|
|
|
|
return StreamingResponse(
|
|
generate(),
|
|
media_type="text/event-stream",
|
|
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no", "Connection": "keep-alive"},
|
|
)
|
|
|
|
|
|
@router.post("/sessions/{session_id}/plan-requests")
|
|
async def request_plans(
|
|
request: Request,
|
|
session_id: str,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
x_expected_revision: int | None = Header(default=None, alias="X-Expected-Revision"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
if owned.get("selected_plan_id"):
|
|
return _error(409, "PLAN_ALREADY_SELECTED", "会话已选定方案，无法重新规划")
|
|
try:
|
|
command = await store.enqueue_plan_request(session_id, idempotency_key=key, expected_revision=x_expected_revision)
|
|
if command.get("status") == "pending":
|
|
await PlanProposalService(
|
|
store,
|
|
agent_store=getattr(request.app.state, "agent_store", None),
|
|
config=get_app_config().report_collaboration,
|
|
llm_factory=getattr(request.app.state, "report_collaboration_planner_llm", None),
|
|
).fulfill(session_id, user_id=user_id, command_id=command["command_id"], idempotency_key=key)
|
|
return Response(status_code=204)
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.get("/sessions/{session_id}/plans")
|
|
async def list_plans(request: Request, session_id: str):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
return {"plans": await store.list_plans(session_id)}
|
|
|
|
|
|
@router.post("/sessions/{session_id}/plans/{plan_id}/select")
|
|
async def select_plan(
|
|
request: Request,
|
|
session_id: str,
|
|
plan_id: str,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
x_expected_revision: int | None = Header(default=None, alias="X-Expected-Revision"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
try:
|
|
result = await store.select_plan(session_id, plan_id, idempotency_key=key, expected_revision=x_expected_revision)
|
|
return {"selected_plan_id": result["selected_plan_id"]}
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.post("/sessions/{session_id}/runs")
|
|
async def start_run(
|
|
request: Request,
|
|
session_id: str,
|
|
body: StartRunBody,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
x_expected_revision: int | None = Header(default=None, alias="X-Expected-Revision"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
try:
|
|
plans = await store.list_plans(session_id)
|
|
plan = next((item for item in plans if item.get("id") == body.plan_id), None)
|
|
if plan is None:
|
|
return _error(404, "NOT_FOUND", f"report collaboration plan not found: {body.plan_id}")
|
|
bundle = freeze_plan_models(plan, get_app_config())
|
|
cfg = get_app_config().report_collaboration
|
|
run = await store.create_run(
|
|
session_id,
|
|
plan_id=body.plan_id,
|
|
idempotency_key=key,
|
|
expected_revision=x_expected_revision,
|
|
role_snapshots=bundle.dump_by_role(),
|
|
template_snapshot=await _frozen_run_template(request, owned, plan),
|
|
budget_snapshot=freeze_run_budget(cfg),
|
|
max_active_runs_per_owner=cfg.max_concurrent_runs_per_user,
|
|
)
|
|
existing_events = await store.list_events(run["id"], after_seq=0, limit=1)
|
|
if not existing_events:
|
|
await store.append_event(
|
|
session_id=session_id,
|
|
run_id=run["id"],
|
|
event_type="heartbeat",
|
|
data={
|
|
"audit": "model_snapshot",
|
|
"roles": [item.model_dump() for item in bundle.by_role.values()],
|
|
"fallbacks": [item.model_dump() for item in bundle.fallbacks()],
|
|
},
|
|
)
|
|
return run
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.get("/sessions/{session_id}/reports")
|
|
async def list_reports(request: Request, session_id: str):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
return {"versions": await store.list_report_versions(session_id)}
|
|
|
|
|
|
@router.post("/sessions/{session_id}/report-rewrites")
|
|
async def create_rewrite(
|
|
request: Request,
|
|
session_id: str,
|
|
body: CreateRewriteBody,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
try:
|
|
return await _rewrite_service(store).create_candidate(
|
|
session_id,
|
|
idempotency_key=key,
|
|
instruction=body.instruction,
|
|
section=body.section,
|
|
)
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.post("/sessions/{session_id}/report-rewrites/{rewrite_id}/apply")
|
|
async def apply_rewrite(
|
|
request: Request,
|
|
session_id: str,
|
|
rewrite_id: str,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
try:
|
|
version, _created = await _rewrite_service(store).apply_candidate(session_id, rewrite_id, idempotency_key=key)
|
|
return version
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.post("/sessions/{session_id}/report-versions/{version_id}/restore")
|
|
async def restore_report(
|
|
request: Request,
|
|
session_id: str,
|
|
version_id: str,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
owned = await _owned_session(store, session_id, user_id)
|
|
if isinstance(owned, JSONResponse):
|
|
return owned
|
|
try:
|
|
version, _created = await _rewrite_service(store).restore(session_id, version_id, idempotency_key=key)
|
|
return version
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.get("/runs/{run_id}")
|
|
async def get_run(request: Request, run_id: str):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
run = await _owned_run(_store(request), run_id, user_id)
|
|
if isinstance(run, JSONResponse):
|
|
return run
|
|
return run
|
|
|
|
|
|
@router.get("/runs/{run_id}/events")
|
|
async def list_run_events(
|
|
request: Request,
|
|
run_id: str,
|
|
after_seq: int = Query(default=0, ge=0),
|
|
limit: int = Query(default=200, ge=1, le=1000),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
store = _store(request)
|
|
run = await _owned_run(store, run_id, user_id)
|
|
if isinstance(run, JSONResponse):
|
|
return run
|
|
return {"events": await store.list_events(run_id, after_seq=after_seq, limit=limit)}
|
|
|
|
|
|
@router.get("/runs/{run_id}/stream")
|
|
async def stream_run(
|
|
request: Request,
|
|
run_id: str,
|
|
after_seq: int = Query(default=0, ge=0),
|
|
last_event_id: str | None = Header(default=None, alias="Last-Event-ID"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
store = _store(request)
|
|
run = await _owned_run(store, run_id, user_id)
|
|
if isinstance(run, JSONResponse):
|
|
return run
|
|
hub = getattr(request.app.state, "report_collaboration_live_hub", None)
|
|
cursor = replay_cursor(last_event_id or request.headers.get("last-event-id"), after_seq)
|
|
settings = getattr(request.app.state, "report_collaboration_sse_settings", None)
|
|
|
|
async def generator():
|
|
async def _disconnected() -> bool:
|
|
return await request.is_disconnected()
|
|
|
|
async for chunk in iter_run_sse(
|
|
store=store,
|
|
hub=hub,
|
|
run_id=run_id,
|
|
cursor=cursor,
|
|
is_disconnected=_disconnected,
|
|
settings=settings,
|
|
):
|
|
yield chunk
|
|
|
|
return StreamingResponse(
|
|
generator(),
|
|
media_type="text/event-stream",
|
|
headers={
|
|
"Cache-Control": "no-cache, no-transform",
|
|
"Connection": "keep-alive",
|
|
"X-Accel-Buffering": "no",
|
|
"X-Report-Collaboration-Run-Status": str(run.get("status") or ""),
|
|
},
|
|
)
|
|
|
|
|
|
@router.post("/runs/{run_id}/cancel")
|
|
async def cancel_run(
|
|
request: Request,
|
|
run_id: str,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
run = await _owned_run(store, run_id, user_id)
|
|
if isinstance(run, JSONResponse):
|
|
return run
|
|
try:
|
|
return await store.request_cancel_run(run_id, idempotency_key=key)
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.post("/runs/{run_id}/commands")
|
|
async def create_command(
|
|
request: Request,
|
|
run_id: str,
|
|
body: CreateCommandBody,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
run = await _owned_run(store, run_id, user_id)
|
|
if isinstance(run, JSONResponse):
|
|
return run
|
|
try:
|
|
return await _runtime_commands(request).submit(
|
|
run_id,
|
|
text=body.text,
|
|
idempotency_key=key,
|
|
target_node_id=body.target_node_id,
|
|
agent_run_id=body.agent_run_id,
|
|
)
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.post("/runs/{run_id}/commands/{command_id}/confirm")
|
|
async def confirm_command(
|
|
request: Request,
|
|
run_id: str,
|
|
command_id: str,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
run = await _owned_run(store, run_id, user_id)
|
|
if isinstance(run, JSONResponse):
|
|
return run
|
|
try:
|
|
return await _runtime_commands(request).confirm(run_id, command_id, idempotency_key=key)
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.post("/runs/{run_id}/commands/{command_id}/cancel")
|
|
async def cancel_command(
|
|
request: Request,
|
|
run_id: str,
|
|
command_id: str,
|
|
x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
|
|
):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
key = _require_idempotency(x_idempotency_key)
|
|
if isinstance(key, JSONResponse):
|
|
return key
|
|
store = _store(request)
|
|
run = await _owned_run(store, run_id, user_id)
|
|
if isinstance(run, JSONResponse):
|
|
return run
|
|
try:
|
|
return await _runtime_commands(request).cancel(run_id, command_id, idempotency_key=key)
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.get("/runs/{run_id}/nodes/{node_id}/contract")
|
|
async def get_node_contract(request: Request, run_id: str, node_id: str):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
store = _store(request)
|
|
run = await _owned_run(store, run_id, user_id)
|
|
if isinstance(run, JSONResponse):
|
|
return run
|
|
try:
|
|
return await store.get_node_contract(run_id, node_id)
|
|
except Exception as exc:
|
|
return _map_store_error(exc)
|
|
|
|
|
|
@router.get("/runs/{run_id}/artifacts")
|
|
async def list_artifacts(request: Request, run_id: str):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
store = _store(request)
|
|
run = await _owned_run(store, run_id, user_id)
|
|
if isinstance(run, JSONResponse):
|
|
return run
|
|
return {"artifacts": await store.list_artifacts(run_id)}
|
|
|
|
|
|
@router.get("/runs/{run_id}/sources")
|
|
async def list_sources(request: Request, run_id: str):
|
|
if disabled := _require_enabled():
|
|
return disabled
|
|
user_id = await _require_user(request)
|
|
run = await _owned_run(_store(request), run_id, user_id)
|
|
if isinstance(run, JSONResponse):
|
|
return run
|
|
return {"sources": []}
|