268 lines
12 KiB
Python
268 lines
12 KiB
Python
"""Fulfill a durable plan_request: catalog → select → assemble → validate → persist.
|
||
|
||
Does not start a run. P1.2 起优先由规划师 LLM 生成定制策略，模板为回退。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from typing import Any
|
||
from uuid import uuid4
|
||
|
||
from app.report_collaboration.contracts.requirements import ReportRequirementSnapshot
|
||
from app.report_collaboration.planning.assembler import PlanAssembler
|
||
from app.report_collaboration.planning.catalog import project_agent_catalog
|
||
from app.report_collaboration.planning.controller import PlanController
|
||
from app.report_collaboration.planning.planner_agent import LlmFactory
|
||
from app.report_collaboration.planning.validator import apply_validation, validate_assembled_plan
|
||
from app.report_collaboration.requirement.defaults import DEFAULT_TITLE
|
||
from deerflow.config.app_config import get_app_config
|
||
from deerflow.config.report_collaboration_config import ReportCollaborationConfig
|
||
from deerflow.persistence.report_collaboration import (
|
||
ReportCollaborationConflictError,
|
||
ReportCollaborationStore,
|
||
ReportCollaborationValidationError,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_BUSY = frozenset({"running", "awaiting_input", "reviewing", "completed", "failed", "cancelled"})
|
||
|
||
|
||
def requirement_from_session(record: dict[str, Any]) -> ReportRequirementSnapshot:
|
||
raw = record.get("requirement_json")
|
||
snapshot: dict[str, Any] | None = None
|
||
if isinstance(raw, dict) and isinstance(raw.get("snapshot"), dict):
|
||
snapshot = dict(raw["snapshot"])
|
||
elif isinstance(raw, dict) and raw.get("topic"):
|
||
snapshot = dict(raw)
|
||
if snapshot is None or not str(snapshot.get("topic") or "").strip() or snapshot.get("topic") == "未指定主题":
|
||
title = str(record.get("title") or "").strip()
|
||
if title and title not in {DEFAULT_TITLE, ""}:
|
||
snapshot = {"topic": title, "required_angles": [], "excluded_angles": [], "revision": max(1, int(record.get("requirement_revision") or 1))}
|
||
else:
|
||
raise ReportCollaborationValidationError("REQUIREMENT_INCOMPLETE", "请先说明报告主题，再生成协作方案")
|
||
snapshot.setdefault("required_angles", [])
|
||
snapshot.setdefault("excluded_angles", [])
|
||
snapshot["revision"] = max(1, int(snapshot.get("revision") or record.get("requirement_revision") or 1))
|
||
return ReportRequirementSnapshot.model_validate(snapshot)
|
||
|
||
|
||
class PlanProposalService:
|
||
def __init__(
|
||
self,
|
||
store: ReportCollaborationStore,
|
||
*,
|
||
agent_store: Any | None = None,
|
||
config: ReportCollaborationConfig | None = None,
|
||
llm_factory: LlmFactory | None = None,
|
||
) -> None:
|
||
self._store = store
|
||
self._agent_store = agent_store
|
||
self._config = config
|
||
self._assembler = PlanAssembler()
|
||
self._llm_factory = llm_factory
|
||
self._llm: Any | None = None
|
||
self._llm_tried = False
|
||
|
||
def _cfg(self) -> ReportCollaborationConfig:
|
||
return self._config or get_app_config().report_collaboration
|
||
|
||
async def fulfill(
|
||
self,
|
||
session_id: str,
|
||
*,
|
||
user_id: str,
|
||
command_id: str,
|
||
idempotency_key: str,
|
||
adjustment_text: str | None = None,
|
||
precomputed_design: tuple[list[Any], str] | None = None,
|
||
planner_reasoning: str | None = None,
|
||
planner_steps: list[dict[str, Any]] | None = None,
|
||
) -> list[dict[str, Any]]:
|
||
record = await self._store.get_session_record(session_id)
|
||
if record is None:
|
||
raise ReportCollaborationValidationError("NOT_FOUND", "会话不存在")
|
||
if record.get("selected_plan_id"):
|
||
raise ReportCollaborationConflictError("PLAN_ALREADY_SELECTED", "会话已选定方案，无法重新规划", current_revision=int(record.get("requirement_revision") or 0))
|
||
if record.get("status") in _BUSY:
|
||
raise ReportCollaborationConflictError("SESSION_BUSY", "当前会话状态不允许生成方案", current_revision=int(record.get("requirement_revision") or 0))
|
||
|
||
try:
|
||
requirement_state = record.get("requirement_json")
|
||
if not isinstance(requirement_state, dict) or requirement_state.get("ready_to_plan") is not True:
|
||
raise ReportCollaborationValidationError(
|
||
"REQUIREMENT_NOT_READY",
|
||
"规划师尚未确认完整意图，请先继续回答澄清问题",
|
||
)
|
||
return await self._build(
|
||
session_id,
|
||
user_id=user_id,
|
||
command_id=command_id,
|
||
idempotency_key=idempotency_key,
|
||
record=record,
|
||
adjustment_text=adjustment_text,
|
||
precomputed_design=precomputed_design,
|
||
planner_reasoning=planner_reasoning,
|
||
planner_steps=planner_steps,
|
||
)
|
||
except Exception as exc:
|
||
await self.fail_command(session_id, command_id, exc)
|
||
raise
|
||
|
||
async def fail_command(self, session_id: str, command_id: str, exc: Exception) -> None:
|
||
try:
|
||
await self._store.update_command_status(command_id, status="failed", error=str(exc)[:500])
|
||
record = await self._store.get_session_record(session_id)
|
||
if record is None or record.get("status") != "planning":
|
||
return
|
||
plans = await self._store.list_plans(session_id)
|
||
if any(item.get("status") in {"proposed", "selected"} for item in plans):
|
||
restore = "proposal_ready"
|
||
elif int(record.get("requirement_revision") or 0) > 0:
|
||
restore = "clarifying"
|
||
else:
|
||
restore = "empty"
|
||
await self._store.save_requirement_state(
|
||
session_id,
|
||
requirement_json=record.get("requirement_json") or {},
|
||
requirement_revision=int(record.get("requirement_revision") or 0),
|
||
status=restore,
|
||
)
|
||
except Exception:
|
||
return
|
||
|
||
async def _build(
|
||
self,
|
||
session_id: str,
|
||
*,
|
||
user_id: str,
|
||
command_id: str,
|
||
idempotency_key: str,
|
||
record: dict[str, Any],
|
||
adjustment_text: str | None,
|
||
precomputed_design: tuple[list[Any], str] | None,
|
||
planner_reasoning: str | None,
|
||
planner_steps: list[dict[str, Any]] | None,
|
||
) -> list[dict[str, Any]]:
|
||
requirement = requirement_from_session(record)
|
||
cfg = self._cfg()
|
||
catalog = await project_agent_catalog(user_id=user_id, config=cfg, agent_store=self._agent_store)
|
||
group_id = f"rpg_{uuid4().hex[:12]}"
|
||
|
||
# 规划师智能体带着完整对话、当前候选和真实智能体目录设计候选工作流。
|
||
# 确定性策略仅用于补足候选数量，最终仍会绑定真实目录中的智能体。
|
||
from app.report_collaboration.planning.planner_agent import PlannerAgent
|
||
|
||
history = [
|
||
{"role": "human" if item.get("role") == "human" else "ai", "content": str(item.get("content") or "")}
|
||
for item in await self._store.list_messages(session_id)
|
||
]
|
||
current_plans = [
|
||
item
|
||
for item in await self._store.list_plans(session_id)
|
||
if item.get("status") in {"proposed", "selected"}
|
||
]
|
||
designed = precomputed_design
|
||
if designed is None:
|
||
designed = await PlannerAgent(self.planner_llm()).design_workflows(
|
||
requirement=requirement,
|
||
history=history,
|
||
catalog=catalog,
|
||
current_plans=current_plans,
|
||
adjustment_text=adjustment_text,
|
||
max_team_members=cfg.max_team_members,
|
||
)
|
||
if designed is None:
|
||
raise ReportCollaborationValidationError(
|
||
"PLANNER_UNAVAILABLE", "规划师智能体未能生成有效工作流，请重试或换个说法再试"
|
||
)
|
||
strategies: list[Any] = list(designed[0])
|
||
planner_reply: str | None = designed[1]
|
||
if len(strategies) < 2:
|
||
existing = {item.strategy for item in strategies}
|
||
for fallback in PlanController().propose(
|
||
requirement,
|
||
max_parallel_tasks=cfg.max_parallel_tasks,
|
||
max_team_members=cfg.max_team_members,
|
||
):
|
||
if fallback.strategy in existing:
|
||
continue
|
||
fallback.recommended = not strategies
|
||
strategies.append(fallback)
|
||
existing.add(fallback.strategy)
|
||
if len(strategies) >= 3:
|
||
break
|
||
|
||
assembled_list = [
|
||
self._assembler.assemble(
|
||
strategy,
|
||
requirement=requirement,
|
||
catalog=catalog,
|
||
config=cfg,
|
||
proposal_group_id=group_id,
|
||
revision=1,
|
||
)
|
||
for strategy in strategies
|
||
]
|
||
persisted: list[dict[str, Any]] = []
|
||
for assembled in assembled_list:
|
||
validation = validate_assembled_plan(assembled, catalog=catalog, config=cfg)
|
||
apply_validation(assembled, validation)
|
||
if not validation.ok:
|
||
raise ReportCollaborationValidationError("PLAN_INVALID", "；".join(validation.errors))
|
||
persisted.append(assembled.candidate)
|
||
|
||
await self._store.supersede_proposed_plans(session_id)
|
||
saved: list[dict[str, Any]] = []
|
||
for candidate in persisted:
|
||
saved.append(await self._store.insert_plan(session_id, candidate.model_dump()))
|
||
await self._store.update_command_status(command_id, status="completed")
|
||
content = planner_reply or "候选工作流已规划完成，你可以选择一个方案，也可以继续通过对话调整智能体结构。"
|
||
await self._store.create_message(
|
||
session_id,
|
||
role="ai",
|
||
content=content,
|
||
idempotency_key=f"{idempotency_key}:plans",
|
||
name="工作流规划师",
|
||
metadata={
|
||
"plan_request_command_id": command_id,
|
||
"proposal_group_id": group_id,
|
||
"plan_source": "planner_agent",
|
||
**({"reasoning_content": planner_reasoning} if planner_reasoning else {}),
|
||
},
|
||
tool_calls=planner_steps or [],
|
||
completed=True,
|
||
)
|
||
return saved
|
||
|
||
def planner_llm(self) -> Any:
|
||
"""规划师模型：注入优先（测试）；否则按 planner_llm_enabled 懒构建。不可用即抛错。"""
|
||
|
||
if self._llm_factory is not None:
|
||
llm = self._llm_factory()
|
||
if llm is None:
|
||
raise ReportCollaborationValidationError("PLANNER_UNAVAILABLE", "规划师智能体不可用，请检查模型配置")
|
||
return llm
|
||
try:
|
||
if not (self._config or get_app_config().report_collaboration).planner_llm_enabled:
|
||
raise ReportCollaborationValidationError(
|
||
"PLANNER_UNAVAILABLE", "规划师智能体未启用（report_collaboration.planner_llm_enabled）"
|
||
)
|
||
except ReportCollaborationValidationError:
|
||
raise
|
||
except Exception as exc:
|
||
raise ReportCollaborationValidationError("PLANNER_UNAVAILABLE", f"规划师智能体配置不可读：{exc}") from None
|
||
if not self._llm_tried:
|
||
self._llm_tried = True
|
||
try:
|
||
from app.report_collaboration.planning.planner_agent import build_planner_llm
|
||
|
||
self._llm = build_planner_llm()
|
||
except Exception as exc:
|
||
logger.warning("rc planner agent llm build failed: %s", exc)
|
||
self._llm = None
|
||
if self._llm is None:
|
||
raise ReportCollaborationValidationError("PLANNER_UNAVAILABLE", "规划师智能体不可用，请检查模型配置")
|
||
return self._llm
|