256 lines
12 KiB
Python
256 lines
12 KiB
Python
"""RC-BE-006: agent catalog, selector, and candidate plans."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from app.report_collaboration.contracts.requirements import ReportRequirementSnapshot
|
|
from app.report_collaboration.planning.assembler import AssembledPlan, NodeBinding, PlanAssembler
|
|
from app.report_collaboration.planning.catalog import project_agent_catalog
|
|
from app.report_collaboration.planning.controller import PlanController
|
|
from app.report_collaboration.planning.selector import AgentSelector
|
|
from app.report_collaboration.planning.service import PlanProposalService
|
|
from app.report_collaboration.planning.validator import validate_assembled_plan
|
|
from deerflow.config.report_collaboration_config import ReportCollaborationConfig
|
|
from deerflow.persistence.report_collaboration import MemoryReportCollaborationStore, ReportCollaborationValidationError
|
|
|
|
|
|
class _AgentStore:
|
|
def __init__(self, rows: list[dict]) -> None:
|
|
self._rows = rows
|
|
|
|
async def list_visible(self, user_id: str) -> list[dict]:
|
|
return list(self._rows)
|
|
|
|
async def get_extras_for(self, agent_ids: list[str]) -> dict[str, dict]:
|
|
return {row["id"]: {"skills": ["web_search"], "tool_groups": ["web"], "model": "demo"} for row in self._rows if row["id"] in agent_ids}
|
|
|
|
|
|
def _requirement() -> ReportRequirementSnapshot:
|
|
return ReportRequirementSnapshot(
|
|
topic="新能源汽车主要品牌市场趋势",
|
|
audience="企业战略决策人员",
|
|
required_angles=["市场份额", "价格策略", "海外进展", "政策环境"],
|
|
excluded_angles=[],
|
|
revision=1,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_catalog_strips_secrets_and_keeps_builtin_roles() -> None:
|
|
store = _AgentStore(
|
|
[
|
|
{
|
|
"id": "ag-research",
|
|
"name": "行业研究员",
|
|
"description": "负责检索与研究",
|
|
"soul": "SECRET_SOUL",
|
|
"api_key": "sk-secret",
|
|
"base_url": "https://hidden.example",
|
|
}
|
|
]
|
|
)
|
|
catalog = await project_agent_catalog(user_id="u1", config=ReportCollaborationConfig(), agent_store=store)
|
|
dumped = [item.model_dump() for item in catalog.agents]
|
|
blob = str(dumped)
|
|
assert "SECRET_SOUL" not in blob
|
|
assert "sk-secret" not in blob
|
|
assert "hidden.example" not in blob
|
|
assert any(item.agent_id == "ag-research" for item in catalog.agents)
|
|
assert any(item.agent_id == "rc.role.reviewer" for item in catalog.agents)
|
|
assert "web_search" in catalog.allowed_tool_names
|
|
|
|
|
|
def test_selector_prefers_visible_user_agent() -> None:
|
|
async def _run() -> None:
|
|
catalog = await project_agent_catalog(
|
|
user_id="u1",
|
|
config=ReportCollaborationConfig(),
|
|
agent_store=_AgentStore([{"id": "ag-research", "name": "行业研究员", "description": "检索研究"}]),
|
|
)
|
|
agent, reason = AgentSelector().pick("researcher", catalog.agents)
|
|
assert agent.agent_id == "ag-research"
|
|
assert "用户可见" in reason
|
|
|
|
import asyncio
|
|
|
|
asyncio.run(_run())
|
|
|
|
|
|
def test_three_strategies_differ_and_stay_acyclic() -> None:
|
|
import asyncio
|
|
|
|
async def _run() -> None:
|
|
cfg = ReportCollaborationConfig()
|
|
catalog = await project_agent_catalog(user_id="u1", config=cfg, agent_store=None)
|
|
requirement = _requirement()
|
|
strategies = PlanController().propose(requirement, max_parallel_tasks=cfg.max_parallel_tasks, max_team_members=cfg.max_team_members)
|
|
assert [item.strategy for item in strategies] == ["balanced_review", "parallel_depth", "focused_fast"]
|
|
assembler = PlanAssembler()
|
|
assembled = [
|
|
assembler.assemble(item, requirement=requirement, catalog=catalog, config=cfg, proposal_group_id="g1")
|
|
for item in strategies
|
|
]
|
|
node_counts = [len(item.candidate.nodes) for item in assembled]
|
|
costs = [item.candidate.estimated_cost_level for item in assembled]
|
|
gates = [tuple(gate.id for gate in item.candidate.quality_gates) for item in assembled]
|
|
assert node_counts[1] > node_counts[0] > node_counts[2]
|
|
assert costs == ["medium", "high", "low"]
|
|
assert "dual_verification" in gates[1]
|
|
assert "dual_verification" not in gates[0]
|
|
assert assembled[0].candidate.recommended is True
|
|
group_ids = {item.candidate.proposal_group_id for item in assembled}
|
|
assert len(group_ids) == 1
|
|
for item in assembled:
|
|
validation = validate_assembled_plan(item, catalog=catalog, config=cfg)
|
|
assert validation.ok, validation.errors
|
|
assert all(node.output_artifact_type for node in item.candidate.nodes)
|
|
assert all(node.acceptance_criteria for node in item.candidate.nodes)
|
|
assert all(binding.agent_id.startswith("rc.role.") for binding in item.bindings)
|
|
assert all(tool in cfg.allowed_tools for node in item.candidate.nodes for tool in node.allowed_tools)
|
|
|
|
asyncio.run(_run())
|
|
|
|
|
|
def test_validator_rejects_unknown_agent_id() -> None:
|
|
import asyncio
|
|
|
|
async def _run() -> None:
|
|
cfg = ReportCollaborationConfig()
|
|
catalog = await project_agent_catalog(user_id="u1", config=cfg, agent_store=None)
|
|
requirement = _requirement()
|
|
strategy = PlanController().propose(requirement, max_parallel_tasks=4, max_team_members=8)[2]
|
|
assembled = PlanAssembler().assemble(strategy, requirement=requirement, catalog=catalog, config=cfg, proposal_group_id="g1")
|
|
assembled.bindings[0] = NodeBinding(
|
|
node_id=assembled.bindings[0].node_id,
|
|
agent_id="not-in-catalog",
|
|
agent_name="幽灵",
|
|
role_key="researcher",
|
|
reason="hallucinated",
|
|
)
|
|
validation = validate_assembled_plan(assembled, catalog=catalog, config=cfg)
|
|
assert validation.ok is False
|
|
assert any("目录外" in item for item in validation.errors)
|
|
|
|
asyncio.run(_run())
|
|
|
|
|
|
class _DesignFakeLLM:
|
|
"""规划师工作流设计假模型：返回单份合法定制方案。"""
|
|
|
|
_payload = (
|
|
'{"reply": "我按需求设计了两条研究线汇写再审的定制工作流。",'
|
|
'"strategy": "custom", "title": "定制方案", "summary": "两线并行汇写再审",'
|
|
'"rationale": "贴合需求", "estimated_duration_seconds": 1200,'
|
|
'"estimated_cost_level": "medium", "quality_gate_ids": ["cite_required"],'
|
|
'"nodes": ['
|
|
'{"slot_id": "research-a", "agent_id": "ag-research", "role_key": "researcher", "display_name": "研究节点", "angle": "市场", "mission": "检索市场证据", "depends_on": [], "output_artifact_type": "EvidenceBundle"},'
|
|
'{"slot_id": "write-main", "agent_id": "ag-writer", "role_key": "writer", "display_name": "撰写节点", "mission": "成稿", "depends_on": ["research-a"], "output_artifact_type": "ReportSectionDraft"}'
|
|
']}'
|
|
)
|
|
|
|
async def ainvoke(self, messages):
|
|
from types import SimpleNamespace
|
|
|
|
return SimpleNamespace(content=self._payload)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_service_writes_plans_without_starting_a_run() -> None:
|
|
store = MemoryReportCollaborationStore()
|
|
session = await store.create_session(owner_id="u1", title="", idempotency_key="s1")
|
|
await store.save_requirement_state(
|
|
session["id"],
|
|
requirement_json={"snapshot": _requirement().model_dump(), "clarifications": [], "assumptions": [], "clarification_round": 1, "ready_to_plan": True},
|
|
requirement_revision=1,
|
|
status="clarifying",
|
|
)
|
|
command = await store.enqueue_plan_request(session["id"], idempotency_key="ik-plan", expected_revision=1)
|
|
agent_store = _AgentStore(
|
|
[
|
|
{"id": "ag-research", "name": "新能源汽车情报员", "description": "检索并分析新能源汽车市场信息"},
|
|
{"id": "ag-writer", "name": "战略报告撰稿人", "description": "把研究成果写成面向管理层的报告"},
|
|
]
|
|
)
|
|
saved = await PlanProposalService(
|
|
store,
|
|
agent_store=agent_store,
|
|
llm_factory=lambda: _DesignFakeLLM(),
|
|
).fulfill(
|
|
session["id"], user_id="u1", command_id=command["command_id"], idempotency_key="ik-plan"
|
|
)
|
|
assert len(saved) == 3
|
|
snapshot = await store.get_snapshot(session["id"])
|
|
assert snapshot["session"]["status"] == "proposal_ready"
|
|
assert snapshot["session"]["selected_plan_id"] is None
|
|
assert snapshot["run"] is None
|
|
assert snapshot["plans"][0]["strategy"] == "custom"
|
|
custom_nodes = snapshot["plans"][0]["nodes"]
|
|
assert [item["agent_id"] for item in custom_nodes] == ["ag-research", "ag-writer"]
|
|
assert [item["role_display_name"] for item in custom_nodes] == ["新能源汽车情报员", "战略报告撰稿人"]
|
|
assert all(item["requirement_revision"] == 1 for item in snapshot["plans"])
|
|
assert snapshot["session"]["active_run_id"] is None
|
|
ai = [item for item in snapshot["messages"] if item["role"] == "ai"][-1]
|
|
assert ai["metadata"]["plan_source"] == "planner_agent"
|
|
assert "定制工作流" in ai["content"]
|
|
|
|
selected = await store.select_plan(session["id"], saved[0]["id"], idempotency_key="ik-select", expected_revision=1)
|
|
after_select = await store.get_snapshot(session["id"])
|
|
assert selected["selected_plan_id"] == saved[0]["id"]
|
|
assert after_select["run"] is None
|
|
assert after_select["session"]["active_run_id"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_service_raises_when_planner_unavailable() -> None:
|
|
store = MemoryReportCollaborationStore()
|
|
session = await store.create_session(owner_id="u1", title="", idempotency_key="s1b")
|
|
await store.save_requirement_state(
|
|
session["id"],
|
|
requirement_json={"snapshot": _requirement().model_dump(), "clarifications": [], "assumptions": [], "clarification_round": 1, "ready_to_plan": True},
|
|
requirement_revision=1,
|
|
status="clarifying",
|
|
)
|
|
command = await store.enqueue_plan_request(session["id"], idempotency_key="ik-plan2", expected_revision=1)
|
|
with pytest.raises(ReportCollaborationValidationError) as exc:
|
|
await PlanProposalService(store, llm_factory=lambda: None).fulfill(
|
|
session["id"], user_id="u1", command_id=command["command_id"], idempotency_key="ik-plan2"
|
|
)
|
|
assert exc.value.code == "PLANNER_UNAVAILABLE"
|
|
|
|
|
|
def test_cycle_is_rejected() -> None:
|
|
cfg = ReportCollaborationConfig()
|
|
candidate_nodes_cycle = AssembledPlan.model_validate(
|
|
{
|
|
"candidate": {
|
|
"id": "p1",
|
|
"proposal_group_id": "g",
|
|
"title": "环",
|
|
"strategy": "focused_fast",
|
|
"summary": "s",
|
|
"rationale": "r",
|
|
"estimated_duration_seconds": 10,
|
|
"estimated_cost_level": "low",
|
|
"requirement_revision": 1,
|
|
"roles": [],
|
|
"nodes": [
|
|
{"id": "a", "label": "A", "depends_on": ["b"], "output_artifact_type": "EvidenceBundle", "acceptance_criteria": ["x"]},
|
|
{"id": "b", "label": "B", "depends_on": ["a"], "output_artifact_type": "VerificationResult", "acceptance_criteria": ["x"]},
|
|
],
|
|
"edges": [{"source": "b", "target": "a"}, {"source": "a", "target": "b"}],
|
|
"quality_gates": [{"id": "cite_required", "label": "来源"}],
|
|
"validation": {"ok": True, "errors": []},
|
|
"revision": 1,
|
|
"status": "proposed",
|
|
},
|
|
"bindings": [],
|
|
}
|
|
)
|
|
from app.report_collaboration.planning.catalog import AgentCatalogProjection
|
|
|
|
catalog = AgentCatalogProjection(agents=[], tools=[], allowed_tool_names=list(cfg.allowed_tools))
|
|
validation = validate_assembled_plan(candidate_nodes_cycle, catalog=catalog, config=cfg)
|
|
assert validation.ok is False
|
|
assert any("循环" in item for item in validation.errors)
|