314 lines
12 KiB
Python
314 lines
12 KiB
Python
"""PlanController: 2–3 strategy RoleDemand graphs. Models never emit executable graphs."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from pydantic import BaseModel, ConfigDict, Field
|
||
|
||
from app.report_collaboration.contracts.plans import RoleDemand
|
||
from app.report_collaboration.contracts.requirements import ReportRequirementSnapshot
|
||
from app.report_collaboration.requirement.defaults import DEFAULT_ANGLES
|
||
|
||
STRATEGIES: tuple[str, ...] = ("balanced_review", "parallel_depth", "focused_fast")
|
||
|
||
|
||
class RoleSlot(BaseModel):
|
||
"""Controller-local slot. ``demand`` is what a model would emit; ``slot_id`` is server-assigned."""
|
||
|
||
model_config = ConfigDict(extra="forbid")
|
||
|
||
slot_id: str
|
||
demand: RoleDemand
|
||
depends_on: list[str] = Field(default_factory=list)
|
||
|
||
|
||
class StrategyRoleSet(BaseModel):
|
||
model_config = ConfigDict(extra="forbid")
|
||
|
||
strategy: str
|
||
title: str
|
||
summary: str
|
||
rationale: str
|
||
estimated_duration_seconds: int
|
||
estimated_cost_level: str
|
||
recommended: bool = False
|
||
slots: list[RoleSlot] = Field(default_factory=list)
|
||
quality_gate_ids: list[str] = Field(default_factory=list)
|
||
|
||
|
||
def resolve_angles(requirement: ReportRequirementSnapshot, count: int) -> list[str]:
|
||
excluded = {item.strip() for item in requirement.excluded_angles if item and item.strip()}
|
||
source = [item for item in requirement.required_angles if item.strip() and item.strip() not in excluded]
|
||
if not source:
|
||
source = [item for item in DEFAULT_ANGLES if item not in excluded]
|
||
if not source:
|
||
source = ["综合趋势"]
|
||
if count <= 1:
|
||
return source[:1]
|
||
picked = source[:count]
|
||
return picked if picked else source[:1]
|
||
|
||
|
||
def _research_demand(angle: str) -> RoleDemand:
|
||
return RoleDemand(
|
||
role_key="researcher",
|
||
display_name="研究员",
|
||
angle=angle,
|
||
mission=f"检索并整理「{angle}」的事实、来源、日期和摘录，覆盖冲突与缺口",
|
||
depends_on_roles=[],
|
||
output_artifact_type="EvidenceBundle",
|
||
suggested_tools=["web_search", "web_fetch", "knowledge_search"],
|
||
)
|
||
|
||
|
||
class PlanController:
|
||
"""Deterministic strategy templates. A later planner model may only fill RoleDemand."""
|
||
|
||
def propose(self, requirement: ReportRequirementSnapshot, *, max_parallel_tasks: int, max_team_members: int) -> list[StrategyRoleSet]:
|
||
topic = requirement.topic
|
||
focused_angles = resolve_angles(requirement, 1)
|
||
balanced_n = min(2, max(1, max_parallel_tasks))
|
||
parallel_n = min(3, max(1, max_parallel_tasks), max(1, max_team_members - 5))
|
||
balanced_angles = resolve_angles(requirement, balanced_n)
|
||
parallel_angles = resolve_angles(requirement, parallel_n)
|
||
return [
|
||
self._balanced(topic, balanced_angles),
|
||
self._parallel(topic, parallel_angles),
|
||
self._focused(topic, focused_angles),
|
||
]
|
||
|
||
def _balanced(self, topic: str, angles: list[str]) -> StrategyRoleSet:
|
||
slots: list[RoleSlot] = []
|
||
research_ids: list[str] = []
|
||
for index, angle in enumerate(angles):
|
||
slot_id = f"research-{index}"
|
||
research_ids.append(slot_id)
|
||
slots.append(
|
||
RoleSlot(
|
||
slot_id=slot_id,
|
||
demand=_research_demand(angle),
|
||
depends_on=[],
|
||
)
|
||
)
|
||
slots.append(
|
||
RoleSlot(
|
||
slot_id="verify",
|
||
demand=RoleDemand(
|
||
role_key="verifier",
|
||
display_name="核验员",
|
||
mission="交叉核验关键数字口径、时效与冲突来源",
|
||
depends_on_roles=["researcher"],
|
||
output_artifact_type="VerificationResult",
|
||
),
|
||
depends_on=list(research_ids),
|
||
)
|
||
)
|
||
slots.append(
|
||
RoleSlot(
|
||
slot_id="analyze",
|
||
demand=RoleDemand(
|
||
role_key="analyst",
|
||
display_name="分析员",
|
||
mission=f"综合各角度证据，形成关于「{topic}」的判断、假设和局限",
|
||
depends_on_roles=["verifier"],
|
||
output_artifact_type="AngleAnalysis",
|
||
),
|
||
depends_on=["verify"],
|
||
)
|
||
)
|
||
slots.append(
|
||
RoleSlot(
|
||
slot_id="write",
|
||
demand=RoleDemand(
|
||
role_key="writer",
|
||
display_name="撰写员",
|
||
mission="按模板撰写带引用的完整报告",
|
||
depends_on_roles=["analyst"],
|
||
output_artifact_type="ReportVersion",
|
||
),
|
||
depends_on=["analyze"],
|
||
)
|
||
)
|
||
slots.append(
|
||
RoleSlot(
|
||
slot_id="review",
|
||
demand=RoleDemand(
|
||
role_key="reviewer",
|
||
display_name="审稿员",
|
||
mission="独立审稿：结论与证据一致、章节完整",
|
||
depends_on_roles=["writer"],
|
||
output_artifact_type="ReviewDecision",
|
||
),
|
||
depends_on=["write"],
|
||
)
|
||
)
|
||
return StrategyRoleSet(
|
||
strategy="balanced_review",
|
||
title="均衡协作 · 研究核验写作审稿全链路",
|
||
summary=f"{len(angles)} 条研究线 + 交叉核验 + 分析 + 写作 + 独立审稿，质量与成本均衡。",
|
||
rationale="在覆盖主要角度的同时控制团队规模，适合大多数正式报告。",
|
||
estimated_duration_seconds=1500,
|
||
estimated_cost_level="medium",
|
||
recommended=True,
|
||
slots=slots,
|
||
quality_gate_ids=["cite_required", "independent_review"],
|
||
)
|
||
|
||
def _parallel(self, topic: str, angles: list[str]) -> StrategyRoleSet:
|
||
slots: list[RoleSlot] = []
|
||
research_ids: list[str] = []
|
||
for index, angle in enumerate(angles):
|
||
slot_id = f"research-{index}"
|
||
research_ids.append(slot_id)
|
||
slots.append(RoleSlot(slot_id=slot_id, demand=_research_demand(angle), depends_on=[]))
|
||
mid = max(1, len(research_ids) // 2)
|
||
first, second = research_ids[:mid], research_ids[mid:] or research_ids[:1]
|
||
slots.append(
|
||
RoleSlot(
|
||
slot_id="verify-a",
|
||
demand=RoleDemand(
|
||
role_key="verifier",
|
||
display_name="核验员",
|
||
angle="主证据链",
|
||
mission="核验前半研究线的口径、时效与冲突",
|
||
depends_on_roles=["researcher"],
|
||
output_artifact_type="VerificationResult",
|
||
),
|
||
depends_on=list(first),
|
||
)
|
||
)
|
||
slots.append(
|
||
RoleSlot(
|
||
slot_id="verify-b",
|
||
demand=RoleDemand(
|
||
role_key="verifier",
|
||
display_name="核验员",
|
||
angle="交叉证据链",
|
||
mission="二次核验其余研究线与高风险结论",
|
||
depends_on_roles=["researcher"],
|
||
output_artifact_type="VerificationResult",
|
||
),
|
||
depends_on=list(second),
|
||
)
|
||
)
|
||
slots.append(
|
||
RoleSlot(
|
||
slot_id="analyze",
|
||
demand=RoleDemand(
|
||
role_key="analyst",
|
||
display_name="分析员",
|
||
mission=f"分角度深入分析「{topic}」，并给出 12–24 个月展望",
|
||
depends_on_roles=["verifier"],
|
||
output_artifact_type="AngleAnalysis",
|
||
),
|
||
depends_on=["verify-a", "verify-b"],
|
||
)
|
||
)
|
||
slots.append(
|
||
RoleSlot(
|
||
slot_id="write",
|
||
demand=RoleDemand(
|
||
role_key="writer",
|
||
display_name="撰写员",
|
||
mission="撰写完整报告",
|
||
depends_on_roles=["analyst"],
|
||
output_artifact_type="ReportVersion",
|
||
),
|
||
depends_on=["analyze"],
|
||
)
|
||
)
|
||
slots.append(
|
||
RoleSlot(
|
||
slot_id="review",
|
||
demand=RoleDemand(
|
||
role_key="reviewer",
|
||
display_name="审稿员",
|
||
mission="高风险场景独立审稿",
|
||
depends_on_roles=["writer"],
|
||
output_artifact_type="ReviewDecision",
|
||
),
|
||
depends_on=["write"],
|
||
)
|
||
)
|
||
return StrategyRoleSet(
|
||
strategy="parallel_depth",
|
||
title="深度并行 · 多角度双重核验",
|
||
summary=f"{len(angles)} 条并行研究 + 双重核验 + 分析 + 写作 + 审稿，覆盖更广、成本更高。",
|
||
rationale="面向高风险决策：角度覆盖与证据强度优先，接受更高成本与时长。",
|
||
estimated_duration_seconds=2400,
|
||
estimated_cost_level="high",
|
||
slots=slots,
|
||
quality_gate_ids=["cite_required", "dual_verification", "independent_review", "coverage_matrix"],
|
||
)
|
||
|
||
def _focused(self, topic: str, angles: list[str]) -> StrategyRoleSet:
|
||
angle = angles[0]
|
||
slots = [
|
||
RoleSlot(
|
||
slot_id="research-0",
|
||
demand=RoleDemand(
|
||
role_key="researcher",
|
||
display_name="研究员",
|
||
angle=angle,
|
||
mission=f"按指定角度「{angle}」检索证据，范围收窄",
|
||
depends_on_roles=[],
|
||
output_artifact_type="EvidenceBundle",
|
||
suggested_tools=["web_search", "web_fetch", "knowledge_search"],
|
||
),
|
||
depends_on=[],
|
||
),
|
||
RoleSlot(
|
||
slot_id="verify",
|
||
demand=RoleDemand(
|
||
role_key="verifier",
|
||
display_name="核验员",
|
||
mission="核验关键事实与数字来源",
|
||
depends_on_roles=["researcher"],
|
||
output_artifact_type="VerificationResult",
|
||
),
|
||
depends_on=["research-0"],
|
||
),
|
||
RoleSlot(
|
||
slot_id="analyze",
|
||
demand=RoleDemand(
|
||
role_key="analyst",
|
||
display_name="分析员",
|
||
mission=f"形成关于「{topic}」的聚焦结论",
|
||
depends_on_roles=["verifier"],
|
||
output_artifact_type="AngleAnalysis",
|
||
),
|
||
depends_on=["verify"],
|
||
),
|
||
RoleSlot(
|
||
slot_id="write",
|
||
demand=RoleDemand(
|
||
role_key="writer",
|
||
display_name="撰写员",
|
||
mission="快速产出带引用的报告",
|
||
depends_on_roles=["analyst"],
|
||
output_artifact_type="ReportVersion",
|
||
),
|
||
depends_on=["analyze"],
|
||
),
|
||
RoleSlot(
|
||
slot_id="review",
|
||
demand=RoleDemand(
|
||
role_key="reviewer",
|
||
display_name="审稿员",
|
||
mission="快速审稿，拦截高危问题",
|
||
depends_on_roles=["writer"],
|
||
output_artifact_type="ReviewDecision",
|
||
),
|
||
depends_on=["write"],
|
||
),
|
||
]
|
||
return StrategyRoleSet(
|
||
strategy="focused_fast",
|
||
title="快速聚焦 · 窄范围短链路",
|
||
summary="单研究线 + 核验 + 分析 + 写作 + 审稿，保留引用核验与审稿，最快出稿。",
|
||
rationale="适合时间紧、角度已知、需要可信结论而非全面覆盖的场景。",
|
||
estimated_duration_seconds=800,
|
||
estimated_cost_level="low",
|
||
slots=slots,
|
||
quality_gate_ids=["cite_required", "independent_review"],
|
||
)
|