|
|
@@ -10,12 +10,15 @@ from langchain_core.language_models.fake_chat_models import FakeMessagesListChat
|
|
|
from langchain_core.messages import AIMessage, ToolMessage
|
|
|
|
|
|
from find_agent_v2.graph import FindAgentRoundGraph
|
|
|
+from find_agent_v2 import tools as find_agent_tools
|
|
|
+from find_agent_v2 import runtime as find_agent_runtime
|
|
|
from find_agent_v2.agent import decide_continued_exploration
|
|
|
from find_agent_v2.runtime import (
|
|
|
DelegateArgs,
|
|
|
FindAgentNodeHost,
|
|
|
_events,
|
|
|
_message_dict,
|
|
|
+ _usage,
|
|
|
)
|
|
|
from find_agent_v2.demand_context import (
|
|
|
V2DemandContext,
|
|
|
@@ -49,11 +52,13 @@ from find_agent_v2.state import (
|
|
|
DemandBrief,
|
|
|
DiscoverySnapshot,
|
|
|
EvaluationBrief,
|
|
|
+ EvidenceAssignment,
|
|
|
ExecutionPlan,
|
|
|
FindAgentState,
|
|
|
NodeRun,
|
|
|
PlanningDecision,
|
|
|
SearchTask,
|
|
|
+ SearchAssignment,
|
|
|
SupervisorDecision,
|
|
|
)
|
|
|
from find_agent_v2.service import _fails_search_share_gate
|
|
|
@@ -63,6 +68,7 @@ from find_agent_v2.tools import (
|
|
|
EVIDENCE_TOOLS,
|
|
|
REPORT_TOOLS,
|
|
|
SEARCH_TOOLS,
|
|
|
+ bound_candidate_tools,
|
|
|
normalize_evaluation_items,
|
|
|
)
|
|
|
|
|
|
@@ -87,6 +93,78 @@ def test_video_tool_error_code_is_observed_as_failed() -> None:
|
|
|
assert _events([message])[0]["error_code"] == "video_understanding_timeout"
|
|
|
|
|
|
|
|
|
+def test_runtime_usage_reads_openrouter_token_usage_cost() -> None:
|
|
|
+ message = AIMessage(
|
|
|
+ content="done",
|
|
|
+ usage_metadata={"input_tokens": 10, "output_tokens": 2, "total_tokens": 12},
|
|
|
+ response_metadata={"token_usage": {"cost": 0.00123}},
|
|
|
+ )
|
|
|
+
|
|
|
+ assert _usage([message]) == {
|
|
|
+ "input_tokens": 10,
|
|
|
+ "output_tokens": 2,
|
|
|
+ "total_tokens": 12,
|
|
|
+ "cost": 0.00123,
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_host_direct_search_and_evidence_paths_use_no_llm(monkeypatch) -> None:
|
|
|
+ calls: list[tuple[str, object]] = []
|
|
|
+
|
|
|
+ async def fake_search(**kwargs):
|
|
|
+ calls.append(("search", kwargs["searches"]))
|
|
|
+ return '{"run_id":"run","searches":[]}'
|
|
|
+
|
|
|
+ async def fake_details(**kwargs):
|
|
|
+ calls.append(("detail", kwargs["candidate_ids"]))
|
|
|
+ return '{"run_id":"run","success_count":2,"failed_count":0}'
|
|
|
+
|
|
|
+ monkeypatch.setattr(find_agent_runtime, "search_videos_v2", fake_search)
|
|
|
+ monkeypatch.setattr(find_agent_runtime, "fetch_candidate_details_v2", fake_details)
|
|
|
+ host = FindAgentNodeHost(observer=NullObserver())
|
|
|
+
|
|
|
+ search_run = await host.run_search_assignment(
|
|
|
+ round_index=1,
|
|
|
+ assignment=SearchAssignment(
|
|
|
+ run_id="run",
|
|
|
+ round_index=1,
|
|
|
+ tasks=[SearchTask(
|
|
|
+ task_id="task-1", keyword="日本间谍", query_reason="测试",
|
|
|
+ )],
|
|
|
+ ),
|
|
|
+ system_prompt="search",
|
|
|
+ user_content="{}",
|
|
|
+ )
|
|
|
+ evidence_run = await host.run_evidence_assignment(
|
|
|
+ round_index=1,
|
|
|
+ assignment=EvidenceAssignment(
|
|
|
+ run_id="run",
|
|
|
+ round_index=1,
|
|
|
+ candidate_ids=[101, 102],
|
|
|
+ evidence_type="detail",
|
|
|
+ candidates=[],
|
|
|
+ ),
|
|
|
+ system_prompt="evidence",
|
|
|
+ user_content="{}",
|
|
|
+ )
|
|
|
+
|
|
|
+ assert calls == [
|
|
|
+ ("search", [{
|
|
|
+ "task_id": "task-1",
|
|
|
+ "keyword": "日本间谍",
|
|
|
+ "query_reason": "测试",
|
|
|
+ "source_type": "mixed",
|
|
|
+ "provider": "internal_keyword",
|
|
|
+ "max_pages": 1,
|
|
|
+ "coverage_targets": [],
|
|
|
+ }]),
|
|
|
+ ("detail", [101, 102]),
|
|
|
+ ]
|
|
|
+ assert (search_run.iterations, evidence_run.iterations) == (0, 0)
|
|
|
+ assert (search_run.tool_calls_made, evidence_run.tool_calls_made) == (1, 1)
|
|
|
+
|
|
|
+
|
|
|
def _evaluation(**overrides):
|
|
|
value = {
|
|
|
"candidate_id": 11,
|
|
|
@@ -449,6 +527,53 @@ def test_stage_tool_allowlists_are_physical_and_isolated() -> None:
|
|
|
assert "query_video_discovery_state" not in all_names
|
|
|
|
|
|
|
|
|
+def test_bound_worker_tools_query_exact_ids_beyond_global_display_limit(monkeypatch) -> None:
|
|
|
+ class Service:
|
|
|
+ def require_run(self, run_id):
|
|
|
+ return {"run_id": run_id, "status": "running"}
|
|
|
+
|
|
|
+ def candidate_inputs(self, run_id, candidate_ids):
|
|
|
+ assert run_id == "large-run"
|
|
|
+ assert candidate_ids == [101, 102]
|
|
|
+ return [
|
|
|
+ {
|
|
|
+ "candidate_id": candidate_id,
|
|
|
+ "aweme_id": f"video-{candidate_id}",
|
|
|
+ "decision_bucket": "pending_evaluation",
|
|
|
+ }
|
|
|
+ for candidate_id in candidate_ids
|
|
|
+ ]
|
|
|
+
|
|
|
+ def get_full_state(self, *_args, **_kwargs):
|
|
|
+ raise AssertionError("Worker 不应通过全局展示投影查询自己的分片")
|
|
|
+
|
|
|
+ def evaluate(self, run_id, items):
|
|
|
+ assert run_id == "large-run"
|
|
|
+ return [
|
|
|
+ {"candidate_id": item["candidate_id"], "decision_bucket": "primary"}
|
|
|
+ for item in items
|
|
|
+ ]
|
|
|
+
|
|
|
+ monkeypatch.setattr(find_agent_tools, "get_find_agent_v2_service", lambda: Service())
|
|
|
+ query_tool = bound_candidate_tools(
|
|
|
+ (EVIDENCE_TOOLS[2],), run_id="large-run", candidate_ids=[101, 102],
|
|
|
+ )[0]
|
|
|
+ query_result = json.loads(query_tool(run_id="large-run", limit=100))
|
|
|
+ assert [item["candidate_id"] for item in query_result["candidates"]] == [101, 102]
|
|
|
+
|
|
|
+ evaluation_tool = bound_candidate_tools(
|
|
|
+ (EVALUATION_TOOLS[1],), run_id="large-run", candidate_ids=[101, 102],
|
|
|
+ )[0]
|
|
|
+ evaluation_result = json.loads(evaluation_tool(
|
|
|
+ run_id="large-run",
|
|
|
+ items=[
|
|
|
+ _evaluation(candidate_id=101),
|
|
|
+ _evaluation(candidate_id=102),
|
|
|
+ ],
|
|
|
+ ))
|
|
|
+ assert [item["candidate_id"] for item in evaluation_result["updated"]] == [101, 102]
|
|
|
+
|
|
|
+
|
|
|
def test_common_prompt_points_to_v2_tables_and_tools() -> None:
|
|
|
assert "find_agent_v2_run" in COMMON_RULES
|
|
|
assert "video_discovery_run" not in COMMON_RULES
|
|
|
@@ -507,9 +632,10 @@ def test_exploration_decision_uses_volume_and_pass_rate(
|
|
|
|
|
|
|
|
|
def test_video_understanding_only_allows_candidates_passing_all_hard_gates(monkeypatch) -> None:
|
|
|
- graph = FindAgentRoundGraph(service=_FakeService(pending_after_search=0), runner=object())
|
|
|
- monkeypatch.setattr(graph, "_full_state", lambda _state: {
|
|
|
- "run": {"rule_config": build_rule_snapshot()},
|
|
|
+ service = _FakeService(pending_after_search=0)
|
|
|
+ graph = FindAgentRoundGraph(service=service, runner=object())
|
|
|
+ monkeypatch.setattr(service, "require_run", lambda _run_id: {
|
|
|
+ "run_id": "run", "rule_config": build_rule_snapshot(),
|
|
|
})
|
|
|
common = {
|
|
|
"video_url": "https://example.test/video.mp4",
|
|
|
@@ -537,26 +663,121 @@ class _FakeService:
|
|
|
self.pending_after_search = pending_after_search
|
|
|
self.stage = "start"
|
|
|
self.updates: list[dict] = []
|
|
|
+ self.rejected_ids: set[int] = set()
|
|
|
|
|
|
- def get_full_state(self, run_id: str, **_kwargs):
|
|
|
+ def require_run(self, run_id: str):
|
|
|
snapshot = self.snapshot(run_id)
|
|
|
- evidence_status = "pending" if self.stage in {"start", "searched"} else "success"
|
|
|
return {
|
|
|
- "run": {"run_id": run_id},
|
|
|
- "searches": [],
|
|
|
- "candidates": [
|
|
|
- {
|
|
|
- "candidate_id": index,
|
|
|
- "decision_bucket": "pending_evaluation",
|
|
|
- "detail_status": evidence_status,
|
|
|
- "portrait_status": evidence_status,
|
|
|
- "publish_at": build_rule_snapshot()["current_datetime"],
|
|
|
- "duration_seconds": 60,
|
|
|
- "share_count": 2000,
|
|
|
- "content_50_plus_ratio": 0.30,
|
|
|
- }
|
|
|
- for index in range(1, snapshot.pending_count + 1)
|
|
|
- ],
|
|
|
+ "run_id": run_id,
|
|
|
+ "status": "running",
|
|
|
+ "outcome_status": None,
|
|
|
+ "current_round": 1,
|
|
|
+ "search_count": snapshot.search_count,
|
|
|
+ "candidate_count": snapshot.candidate_count,
|
|
|
+ "valid_primary_count": snapshot.valid_primary_count,
|
|
|
+ "rule_config": build_rule_snapshot(),
|
|
|
+ }
|
|
|
+
|
|
|
+ def get_search_summaries(self, _run_id: str):
|
|
|
+ if self.stage == "start":
|
|
|
+ return []
|
|
|
+ return [{
|
|
|
+ "search_id": 1,
|
|
|
+ "round_index": 1,
|
|
|
+ "keyword": "测试需求",
|
|
|
+ "query_reason": "建立候选池",
|
|
|
+ "provider": "internal_keyword",
|
|
|
+ "page_no": 1,
|
|
|
+ "has_more": False,
|
|
|
+ "next_cursor": None,
|
|
|
+ "status": "success",
|
|
|
+ "result_count": self.pending_after_search,
|
|
|
+ }]
|
|
|
+
|
|
|
+ def _candidate_rows(self):
|
|
|
+ if self.stage == "start":
|
|
|
+ return []
|
|
|
+ evidence_status = "pending" if self.stage == "searched" else "success"
|
|
|
+ return [{
|
|
|
+ "candidate_id": index,
|
|
|
+ "decision_bucket": (
|
|
|
+ "rejected"
|
|
|
+ if self.stage == "evaluated" or index in self.rejected_ids
|
|
|
+ else "pending_evaluation"
|
|
|
+ ),
|
|
|
+ "detail_status": evidence_status,
|
|
|
+ "portrait_status": evidence_status,
|
|
|
+ "publish_at": build_rule_snapshot()["current_datetime"],
|
|
|
+ "duration_seconds": 60,
|
|
|
+ "share_count": 2000,
|
|
|
+ "content_50_plus_ratio": 0.30,
|
|
|
+ } for index in range(1, self.pending_after_search + 1)]
|
|
|
+
|
|
|
+ def get_full_state(self, run_id: str, **_kwargs):
|
|
|
+ return {
|
|
|
+ "run": self.require_run(run_id),
|
|
|
+ "searches": self.get_search_summaries(run_id),
|
|
|
+ "candidates": self._candidate_rows(),
|
|
|
+ }
|
|
|
+
|
|
|
+ def candidate_inputs(self, _run_id: str, candidate_ids: list[int]):
|
|
|
+ allowed = {int(value) for value in candidate_ids}
|
|
|
+ return [
|
|
|
+ item for item in self._candidate_rows()
|
|
|
+ if int(item["candidate_id"]) in allowed
|
|
|
+ ]
|
|
|
+
|
|
|
+ def get_candidate_progress(self, _run_id: str):
|
|
|
+ candidates = self._candidate_rows()
|
|
|
+ pending = [
|
|
|
+ item for item in candidates
|
|
|
+ if item["decision_bucket"] == "pending_evaluation"
|
|
|
+ ]
|
|
|
+ return {
|
|
|
+ "total_count": len(candidates),
|
|
|
+ "pending_count": len(pending),
|
|
|
+ "primary_count": 0,
|
|
|
+ "rejected_count": len(candidates) - len(pending),
|
|
|
+ "detail_pending_count": sum(item["detail_status"] == "pending" for item in pending),
|
|
|
+ "detail_success_count": sum(item["detail_status"] == "success" for item in pending),
|
|
|
+ "detail_failed_count": 0,
|
|
|
+ "portrait_pending_count": sum(item["portrait_status"] == "pending" for item in pending),
|
|
|
+ "portrait_success_count": sum(item["portrait_status"] == "success" for item in pending),
|
|
|
+ "portrait_failed_count": 0,
|
|
|
+ "evidence_completed_count": sum(
|
|
|
+ item["detail_status"] != "pending" and item["portrait_status"] != "pending"
|
|
|
+ for item in pending
|
|
|
+ ),
|
|
|
+ "evidence_success_count": sum(
|
|
|
+ item["detail_status"] == "success" and item["portrait_status"] == "success"
|
|
|
+ for item in pending
|
|
|
+ ),
|
|
|
+ }
|
|
|
+
|
|
|
+ def list_pending_evidence_ids(self, _run_id: str, evidence_type: str, *, limit: int):
|
|
|
+ key = f"{evidence_type}_status"
|
|
|
+ return [
|
|
|
+ int(item["candidate_id"]) for item in self._candidate_rows()
|
|
|
+ if item["decision_bucket"] == "pending_evaluation" and item[key] == "pending"
|
|
|
+ ][:limit]
|
|
|
+
|
|
|
+ def list_ready_evaluation_ids(self, _run_id: str, *, limit: int):
|
|
|
+ return [
|
|
|
+ int(item["candidate_id"]) for item in self._candidate_rows()
|
|
|
+ if item["decision_bucket"] == "pending_evaluation"
|
|
|
+ and item["detail_status"] != "pending"
|
|
|
+ and item["portrait_status"] != "pending"
|
|
|
+ ][:limit]
|
|
|
+
|
|
|
+ def count_pending_candidates(self, run_id: str) -> int:
|
|
|
+ return int(self.get_candidate_progress(run_id)["pending_count"])
|
|
|
+
|
|
|
+ def get_report_state(self, run_id: str):
|
|
|
+ return {
|
|
|
+ "run": self.require_run(run_id),
|
|
|
+ "summary": self.get_candidate_progress(run_id),
|
|
|
+ "primary_candidates": [],
|
|
|
+ "rejection_reason_distribution": {},
|
|
|
}
|
|
|
|
|
|
def snapshot(self, _run_id: str) -> DiscoverySnapshot:
|
|
|
@@ -579,7 +800,9 @@ class _FakeService:
|
|
|
return 0
|
|
|
|
|
|
def reject_failed_gates(self, _run_id: str, failures) -> list[int]:
|
|
|
- return [candidate_id for candidate_id, _gate in failures]
|
|
|
+ rejected = [candidate_id for candidate_id, _gate in failures]
|
|
|
+ self.rejected_ids.update(rejected)
|
|
|
+ return rejected
|
|
|
|
|
|
|
|
|
class _FakeRunner:
|
|
|
@@ -680,39 +903,82 @@ class _WrongEvidenceProposalRunner(_FakeRunner):
|
|
|
return run
|
|
|
|
|
|
|
|
|
+class _OptimizedRunner(_FakeRunner):
|
|
|
+ def __init__(self, service: _FakeService) -> None:
|
|
|
+ super().__init__(service)
|
|
|
+ self.host_calls: list[str] = []
|
|
|
+
|
|
|
+ async def run_search_assignment(self, *, round_index, user_content, **_kwargs):
|
|
|
+ self.host_calls.append("search")
|
|
|
+ self.inputs.append(("search", user_content))
|
|
|
+ self.service.stage = "searched"
|
|
|
+ return NodeRun("search", round_index, "host search", 0, 1)
|
|
|
+
|
|
|
+ async def run_evidence_assignment(
|
|
|
+ self, *, round_index, user_content, **_kwargs,
|
|
|
+ ):
|
|
|
+ self.host_calls.append("evidence")
|
|
|
+ self.inputs.append(("evidence", user_content))
|
|
|
+ self.service.stage = "evidenced"
|
|
|
+ return NodeRun("evidence", round_index, "host evidence", 0, 1)
|
|
|
+
|
|
|
+ async def run_host_supervision(
|
|
|
+ self, *, round_index, decision, output_enricher=None, **_kwargs,
|
|
|
+ ):
|
|
|
+ self.host_calls.append(f"supervisor:{decision.next_action}")
|
|
|
+ if output_enricher is not None:
|
|
|
+ self.supervisor_visual_outputs.append(output_enricher(decision))
|
|
|
+ return NodeRun("supervisor", round_index, decision.model_dump_json(), 0, 0), decision
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_host_executes_deterministic_stages_without_redundant_llm_calls() -> None:
|
|
|
+ service = _FakeService(pending_after_search=2)
|
|
|
+ runner = _OptimizedRunner(service)
|
|
|
+ graph = FindAgentRoundGraph(service=service, runner=runner)
|
|
|
+
|
|
|
+ result = await graph.invoke(FindAgentState(
|
|
|
+ run_id="optimized-run", user_input="task", round_index=1,
|
|
|
+ ))
|
|
|
+
|
|
|
+ assert runner.host_calls == [
|
|
|
+ "search",
|
|
|
+ "supervisor:evidence",
|
|
|
+ "evidence",
|
|
|
+ "evidence",
|
|
|
+ "supervisor:evaluator",
|
|
|
+ ]
|
|
|
+ assert [node for node, _tools in runner.calls] == [
|
|
|
+ "supervisor", "evaluator", "supervisor",
|
|
|
+ ]
|
|
|
+ assert result.snapshot is not None and result.snapshot.pending_count == 0
|
|
|
+
|
|
|
+
|
|
|
def test_supervisor_progress_counts_missing_evidence_only_for_pending_candidates() -> None:
|
|
|
class ProgressService:
|
|
|
@staticmethod
|
|
|
- def get_full_state(_run_id: str, **_kwargs):
|
|
|
+ def require_run(_run_id: str):
|
|
|
+ return {}
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def get_search_summaries(_run_id: str):
|
|
|
+ return []
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def get_candidate_progress(_run_id: str):
|
|
|
return {
|
|
|
- "run": {},
|
|
|
- "searches": [],
|
|
|
- "candidates": [
|
|
|
- {
|
|
|
- "candidate_id": 1,
|
|
|
- "decision_bucket": "pending_evaluation",
|
|
|
- "detail_status": "success",
|
|
|
- "portrait_status": "pending",
|
|
|
- },
|
|
|
- {
|
|
|
- "candidate_id": 2,
|
|
|
- "decision_bucket": "rejected",
|
|
|
- "detail_status": "pending",
|
|
|
- "portrait_status": "pending",
|
|
|
- },
|
|
|
- {
|
|
|
- "candidate_id": 3,
|
|
|
- "decision_bucket": "pending_evaluation",
|
|
|
- "detail_status": "failed",
|
|
|
- "portrait_status": "success",
|
|
|
- },
|
|
|
- {
|
|
|
- "candidate_id": 4,
|
|
|
- "decision_bucket": "pending_evaluation",
|
|
|
- "detail_status": "success",
|
|
|
- "portrait_status": "success",
|
|
|
- },
|
|
|
- ],
|
|
|
+ "total_count": 4,
|
|
|
+ "pending_count": 3,
|
|
|
+ "primary_count": 0,
|
|
|
+ "rejected_count": 1,
|
|
|
+ "detail_pending_count": 0,
|
|
|
+ "detail_success_count": 2,
|
|
|
+ "detail_failed_count": 1,
|
|
|
+ "portrait_pending_count": 1,
|
|
|
+ "portrait_success_count": 2,
|
|
|
+ "portrait_failed_count": 0,
|
|
|
+ "evidence_completed_count": 2,
|
|
|
+ "evidence_success_count": 1,
|
|
|
}
|
|
|
|
|
|
graph = FindAgentRoundGraph(service=ProgressService(), runner=object())
|
|
|
@@ -845,28 +1111,34 @@ async def test_evaluator_model_never_receives_hard_gate_failures() -> None:
|
|
|
service.stage = "evidenced"
|
|
|
rejected: list[int] = []
|
|
|
|
|
|
- def state(_run_id: str, **_kwargs):
|
|
|
- return {
|
|
|
- "run": {"run_id": "r", "rule_config": build_rule_snapshot()},
|
|
|
- "searches": [],
|
|
|
- "candidates": [{
|
|
|
- "candidate_id": 1,
|
|
|
- "decision_bucket": "pending_evaluation",
|
|
|
- "detail_status": "success",
|
|
|
- "portrait_status": "success",
|
|
|
- "publish_at": build_rule_snapshot()["current_datetime"],
|
|
|
- "duration_seconds": 10,
|
|
|
- "share_count": 2000,
|
|
|
- "content_50_plus_ratio": 0.30,
|
|
|
- "video_url": "https://example.test/video.mp4",
|
|
|
- }],
|
|
|
- }
|
|
|
-
|
|
|
- service.get_full_state = state # type: ignore[method-assign]
|
|
|
- service.reject_failed_gates = ( # type: ignore[method-assign]
|
|
|
- lambda _run_id, failures: rejected.extend(candidate_id for candidate_id, _ in failures)
|
|
|
- or [candidate_id for candidate_id, _ in failures]
|
|
|
+ bucket = ["pending_evaluation"]
|
|
|
+
|
|
|
+ def candidate_inputs(_run_id: str, candidate_ids: list[int]):
|
|
|
+ if 1 not in candidate_ids:
|
|
|
+ return []
|
|
|
+ return [{
|
|
|
+ "candidate_id": 1,
|
|
|
+ "decision_bucket": bucket[0],
|
|
|
+ "detail_status": "success",
|
|
|
+ "portrait_status": "success",
|
|
|
+ "publish_at": build_rule_snapshot()["current_datetime"],
|
|
|
+ "duration_seconds": 10,
|
|
|
+ "share_count": 2000,
|
|
|
+ "content_50_plus_ratio": 0.30,
|
|
|
+ "video_url": "https://example.test/video.mp4",
|
|
|
+ }]
|
|
|
+
|
|
|
+ def reject(_run_id: str, failures):
|
|
|
+ ids = [candidate_id for candidate_id, _ in failures]
|
|
|
+ rejected.extend(ids)
|
|
|
+ bucket[0] = "rejected"
|
|
|
+ return ids
|
|
|
+
|
|
|
+ service.candidate_inputs = candidate_inputs # type: ignore[method-assign]
|
|
|
+ service.list_ready_evaluation_ids = ( # type: ignore[method-assign]
|
|
|
+ lambda _run_id, limit: [1] if bucket[0] == "pending_evaluation" else []
|
|
|
)
|
|
|
+ service.reject_failed_gates = reject # type: ignore[method-assign]
|
|
|
runner = _FakeRunner(service)
|
|
|
graph = FindAgentRoundGraph(service=service, runner=runner)
|
|
|
|
|
|
@@ -880,39 +1152,67 @@ async def test_evaluator_model_never_receives_hard_gate_failures() -> None:
|
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
-async def test_round_graph_reenters_evaluator_until_pending_queue_is_empty() -> None:
|
|
|
- service = _FakeService(pending_after_search=3)
|
|
|
+async def test_evaluator_drains_database_batches_until_pending_queue_is_empty() -> None:
|
|
|
+ class BatchedService(_FakeService):
|
|
|
+ def __init__(self) -> None:
|
|
|
+ super().__init__(pending_after_search=3)
|
|
|
+ self.remaining_ids = {1, 2, 3}
|
|
|
+
|
|
|
+ def _candidate_rows(self):
|
|
|
+ if self.stage == "start":
|
|
|
+ return []
|
|
|
+ evidence_status = "pending" if self.stage == "searched" else "success"
|
|
|
+ return [{
|
|
|
+ "candidate_id": index,
|
|
|
+ "decision_bucket": (
|
|
|
+ "pending_evaluation" if index in self.remaining_ids else "rejected"
|
|
|
+ ),
|
|
|
+ "detail_status": evidence_status,
|
|
|
+ "portrait_status": evidence_status,
|
|
|
+ "publish_at": build_rule_snapshot()["current_datetime"],
|
|
|
+ "duration_seconds": 60,
|
|
|
+ "share_count": 2000,
|
|
|
+ "content_50_plus_ratio": 0.30,
|
|
|
+ } for index in range(1, 4)]
|
|
|
+
|
|
|
+ def list_ready_evaluation_ids(self, _run_id: str, *, limit: int):
|
|
|
+ del limit
|
|
|
+ return sorted(self.remaining_ids)[:1]
|
|
|
+
|
|
|
+ def snapshot(self, _run_id: str) -> DiscoverySnapshot:
|
|
|
+ if self.stage == "start":
|
|
|
+ return DiscoverySnapshot("running", 0, 0, 0, 0, 0, 0)
|
|
|
+ return DiscoverySnapshot(
|
|
|
+ "running", 1, 3, len(self.remaining_ids), 0, 0,
|
|
|
+ 3 - len(self.remaining_ids),
|
|
|
+ )
|
|
|
+
|
|
|
+ service = BatchedService()
|
|
|
|
|
|
class BatchedRunner(_FakeRunner):
|
|
|
def __init__(self, fake_service: _FakeService) -> None:
|
|
|
super().__init__(fake_service)
|
|
|
- self.remaining = 3
|
|
|
-
|
|
|
- async def run_node(self, *, node, round_index, tools=(), **kwargs) -> NodeRun:
|
|
|
- result = await super().run_node(
|
|
|
- node=node, round_index=round_index, tools=tools, **kwargs,
|
|
|
- )
|
|
|
+ async def run_node(
|
|
|
+ self, *, node, round_index, tools=(), user_content="", **kwargs,
|
|
|
+ ) -> NodeRun:
|
|
|
if node == "evaluator":
|
|
|
- self.remaining -= 1
|
|
|
- self.service.stage = "evaluated" if self.remaining == 0 else "batched"
|
|
|
- return result
|
|
|
-
|
|
|
- runner = BatchedRunner(service)
|
|
|
- original_snapshot = service.snapshot
|
|
|
-
|
|
|
- def batched_snapshot(run_id: str) -> DiscoverySnapshot:
|
|
|
- if service.stage == "batched":
|
|
|
- base = original_snapshot(run_id)
|
|
|
- return replace(
|
|
|
- base,
|
|
|
- search_count=1,
|
|
|
- candidate_count=3,
|
|
|
- pending_count=runner.remaining,
|
|
|
- rejected_count=3 - runner.remaining,
|
|
|
+ self.calls.append((node, _names(tools)))
|
|
|
+ self.inputs.append((node, user_content))
|
|
|
+ candidate_ids = json.loads(user_content)["candidate_ids"]
|
|
|
+ self.service.remaining_ids.difference_update(candidate_ids)
|
|
|
+ self.service.stage = (
|
|
|
+ "evaluated" if not self.service.remaining_ids else "batched"
|
|
|
+ )
|
|
|
+ return NodeRun(node, round_index, "", 1, 0)
|
|
|
+ return await super().run_node(
|
|
|
+ node=node,
|
|
|
+ round_index=round_index,
|
|
|
+ tools=tools,
|
|
|
+ user_content=user_content,
|
|
|
+ **kwargs,
|
|
|
)
|
|
|
- return original_snapshot(run_id)
|
|
|
|
|
|
- service.snapshot = batched_snapshot # type: ignore[method-assign]
|
|
|
+ runner = BatchedRunner(service)
|
|
|
graph = FindAgentRoundGraph(service=service, runner=runner)
|
|
|
|
|
|
result = await graph.invoke(
|
|
|
@@ -948,7 +1248,7 @@ def test_supervisor_can_choose_an_extra_search_within_budget() -> None:
|
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
-async def test_round_graph_retries_one_stagnant_evaluator_response() -> None:
|
|
|
+async def test_round_graph_rejects_incomplete_evaluator_batch() -> None:
|
|
|
service = _FakeService(pending_after_search=2)
|
|
|
|
|
|
class RetryRunner(_FakeRunner):
|
|
|
@@ -958,8 +1258,6 @@ async def test_round_graph_retries_one_stagnant_evaluator_response() -> None:
|
|
|
if node == "evaluator":
|
|
|
self.calls.append((node, _names(tools)))
|
|
|
self.evaluator_calls += 1
|
|
|
- if self.evaluator_calls == 2:
|
|
|
- self.service.stage = "evaluated"
|
|
|
return NodeRun(node, round_index, "", 1, 0)
|
|
|
return await super().run_node(
|
|
|
node=node, round_index=round_index, tools=tools, **kwargs,
|
|
|
@@ -968,9 +1266,9 @@ async def test_round_graph_retries_one_stagnant_evaluator_response() -> None:
|
|
|
runner = RetryRunner(service)
|
|
|
graph = FindAgentRoundGraph(service=service, runner=runner)
|
|
|
|
|
|
- result = await graph.invoke(
|
|
|
- FindAgentState(run_id="retry-run", user_input="task", round_index=1),
|
|
|
- )
|
|
|
+ with pytest.raises(RuntimeError, match="评估分批未完整消费"):
|
|
|
+ await graph.invoke(
|
|
|
+ FindAgentState(run_id="retry-run", user_input="task", round_index=1),
|
|
|
+ )
|
|
|
|
|
|
- assert runner.evaluator_calls == 2
|
|
|
- assert result.snapshot is not None and result.snapshot.pending_count == 0
|
|
|
+ assert runner.evaluator_calls == 1
|
xueyiming