"""Requirement resolver: the planner agent is the only path. 模型不可用即无法继续 —— 直接抛 ``PLANNER_UNAVAILABLE``，没有任何规则/ 模板兜底（设计 v2：需求确认必须由真智能体完成）。 """ from __future__ import annotations import logging from typing import Any from uuid import uuid4 from app.report_collaboration.contracts.clarifications import ClarificationRecord from app.report_collaboration.requirement.defaults import COORDINATOR_NAME, DEFAULT_TITLE from app.report_collaboration.requirement.intent_router import PrePlanDecision from deerflow.config.app_config import get_app_config from deerflow.persistence.report_collaboration import ( ReportCollaborationStore, ReportCollaborationValidationError, ) from deerflow.persistence.report_collaboration.codec import iso_now logger = logging.getLogger(__name__) # 仅供类型注解；运行时惰性导入避免 requirement ↔ planning 循环依赖。 LlmFactory = Any def _title_from_topic(topic: str, current: str | None) -> str | None: if current and current != DEFAULT_TITLE: return None cleaned = (topic or "").strip() if not cleaned or cleaned == "未指定主题": return None return cleaned[:40] def _state_from_session(record: dict[str, Any]) -> dict[str, Any]: raw = record.get("requirement_json") if isinstance(raw, dict) and ("snapshot" in raw or "clarifications" in raw): return { "snapshot": raw.get("snapshot"), "clarifications": list(raw.get("clarifications") or []), "assumptions": list(raw.get("assumptions") or []), "clarification_round": int(raw.get("clarification_round") or 0), "ready_to_plan": bool(raw.get("ready_to_plan")), } if isinstance(raw, dict) and raw.get("topic"): return { "snapshot": raw, "clarifications": [], "assumptions": [], "clarification_round": 0, "ready_to_plan": False, } return {"snapshot": None, "clarifications": [], "assumptions": [], "clarification_round": 0, "ready_to_plan": False} class RequirementResolver: """Writes requirement revisions and clarification records. Does not enqueue plans.""" def __init__( self, store: ReportCollaborationStore, *, planner_llm_factory: LlmFactory | None = None, config: Any | None = None, ) -> None: self._store = store self._planner_llm_factory = planner_llm_factory self._config = config self._default_llm: Any = None self._default_llm_tried = False def _planner_llm(self) -> Any: """规划师模型：注入优先（测试）；否则按 planner_llm_enabled 懒构建。不可用即抛错。""" if self._planner_llm_factory is not None: llm = self._planner_llm_factory() if llm is None: raise ReportCollaborationValidationError("PLANNER_UNAVAILABLE", "规划师智能体不可用，请检查模型配置") return llm try: from deerflow.config.report_collaboration_config import ReportCollaborationConfig cfg: Any = self._config if cfg is None: cfg = get_app_config().report_collaboration if not ReportCollaborationConfig.model_validate(cfg).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._default_llm_tried: self._default_llm_tried = True try: from app.report_collaboration.planning.planner_agent import build_planner_llm self._default_llm = build_planner_llm() except Exception as exc: logger.warning("rc planner agent llm build failed: %s", exc) self._default_llm = None if self._default_llm is None: raise ReportCollaborationValidationError("PLANNER_UNAVAILABLE", "规划师智能体不可用，请检查模型配置") return self._default_llm async def handle_user_message( self, session_id: str, *, user_message: dict[str, Any], idempotency_key: str, max_clarification_rounds: int | None = None, ) -> PrePlanDecision | None: record = await self._store.get_session_record(session_id) if record is None: return None status = str(record.get("status") or "empty") if status not in {"empty", "clarifying"}: return None messages = await self._store.list_messages(session_id) if any((item.get("metadata") or {}).get("preplan_for") == user_message["message_id"] for item in messages): return None state = _state_from_session(record) cfg_rounds = max_clarification_rounds if cfg_rounds is None: cfg_rounds = int(get_app_config().report_collaboration.max_clarification_rounds) text = str(user_message.get("content") or "") history = [ {"role": "human" if item.get("role") == "human" else "ai", "content": str(item.get("content") or "")} for item in messages if item.get("message_id") != user_message["message_id"] ] decision = await self._agent_decision( text, history=history, state=state, previous_revision=int(record.get("requirement_revision") or 0), max_rounds=cfg_rounds, ) if decision is None: # 模型调用失败/输出不合法 —— 无法继续，直接报错（无兜底）。 raise ReportCollaborationValidationError("PLANNER_UNAVAILABLE", "规划师智能体本轮未返回有效决策，请重试") await self._apply(session_id, record, state, decision, user_message=user_message, idempotency_key=idempotency_key) return decision async def _agent_decision( self, text: str, *, history: list[dict[str, str]], state: dict[str, Any], previous_revision: int, max_rounds: int, ) -> PrePlanDecision | None: llm = self._planner_llm() from app.report_collaboration.planning.planner_agent import PlannerAgent agent = PlannerAgent(llm) turn = await agent.converse( user_text=text, history=history, previous_snapshot=state["snapshot"], previous_revision=previous_revision, clarification_round=int(state["clarification_round"] or 0), max_rounds=max_rounds, ) if turn is None: return None requirement = agent.merge_requirement( turn, previous_snapshot=state["snapshot"], previous_revision=previous_revision, ) return PrePlanDecision( action="ready" if turn.action == "ready" else "clarify", requirement=requirement, questions=[item for item in turn.questions if item and item.strip()][:3], impact=[item for item in turn.impact if item][:3], assumptions=[item for item in turn.assumptions if item], used_defaults=bool(turn.assumptions), reason="planner_agent", confidence=0.9, reply=turn.reply.strip(), ) async def _apply( self, session_id: str, record: dict[str, Any], state: dict[str, Any], decision: PrePlanDecision, *, user_message: dict[str, Any], idempotency_key: str, ) -> None: clarifications = [dict(item) for item in state["clarifications"]] round_count = int(state["clarification_round"] or 0) if decision.action == "clarify": round_count += 1 for item in clarifications: if not item.get("resolved"): item["resolved"] = True clarification = ClarificationRecord( id=f"clr_{uuid4().hex[:12]}", questions=list(decision.questions), impact=list(decision.impact), resolved=False, created_at=iso_now(), ) clarifications.append(clarification.model_dump()) content = decision.reply or "" metadata = { "preplan_for": user_message["message_id"], "clarification": clarification.model_dump(), } ready = False else: for item in clarifications: item["resolved"] = True content = decision.reply or "" metadata = { "preplan_for": user_message["message_id"], "ready_to_plan": True, # 结构化需求快照：前端据此渲染「需求摘要卡」（计划模式 UI）。 "requirement": decision.requirement.model_dump(), } ready = True payload = { "snapshot": decision.requirement.model_dump(), "clarifications": clarifications, "assumptions": list(decision.assumptions), "clarification_round": round_count, "ready_to_plan": ready, } await self._store.save_requirement_state( session_id, requirement_json=payload, requirement_revision=decision.requirement.revision, status="clarifying", title=_title_from_topic(decision.requirement.topic, record.get("title")), ) await self._store.create_message( session_id, role="ai", content=content, idempotency_key=f"{idempotency_key}:preplan", name=COORDINATOR_NAME, metadata=metadata, completed=True, )