修复发现的问题

find_agent_v2/graph.py

@@ -20,6 +20,7 @@ from find_agent_v2.tools import (
     SEARCH_TOOLS,
     ToolFn,
     bound_candidate_tools,
+    normalize_evaluation_items,
 )
 
 
@@ -295,6 +296,30 @@ class FindAgentRoundGraph:
         async def run_shard(index: int, items: list[dict[str, Any]]) -> NodeRun:
             candidate_ids = [int(item["candidate_id"]) for item in items]
             async with semaphore:
+                structured_runner = getattr(self.runner, "run_evaluation", None)
+                if callable(structured_runner):
+                    run, proposed = await structured_runner(
+                        round_index=state["round_index"],
+                        system_prompt=EVALUATOR_PROMPT,
+                        user_content=(
+                            f"你只负责当前分片 candidate_ids={candidate_ids}。"
+                            "必须为这些 candidate_id 各输出一次结构化评估。\n\n"
+                            + self._shard_context(state, items)
+                        ),
+                        slots=self._shard_slots(state, items),
+                        branch_key=f"step-{state.get('supervisor_step', 0)}-shard-{index}",
+                    )
+                    normalized = normalize_evaluation_items(
+                        proposed, allowed_candidates=items,
+                    )
+                    updated = self.service.evaluate(state["run_id"], normalized)
+                    updated_ids = {int(item["candidate_id"]) for item in updated}
+                    if updated_ids != set(candidate_ids):
+                        raise RuntimeError(
+                            "评估写入结果不完整："
+                            f"expected={candidate_ids}, updated={sorted(updated_ids)}"
+                        )
+                    return run
                 return await self.runner.run_node(
                     node="evaluator",
                     round_index=state["round_index"],

find_agent_v2/prompts.py

@@ -73,10 +73,11 @@ EVALUATOR_PROMPT = COMMON_RULES + """
 
 # 当前模块：评估与分池
 
-宿主会给你一个互斥的 pending_evaluation 候选分片。使用 `query_pending_candidates_v2` 只读当前
-分片，根据已保存证据给出 R/E/S/V，并批量调用 `evaluate_candidates_v2`。分片内所有候选必须
-进入 primary 或 rejected；更新后再次查询确认。不得访问分片外候选。
-不得搜索、补证或修改运行终态。
+宿主会给你一个互斥的 pending_evaluation 候选分片。你只负责根据输入中已经保存的证据，输出
+严格结构化的 R/E/S/V 和分池判断；数据库写入、分片校验与完成确认由宿主负责。
+每一项必须填写输入中的数据库 `candidate_id`（整数），不得拿 `aweme_id` 代替。分片内每个候选
+必须恰好输出一次并进入 primary 或 rejected，不得遗漏、重复或访问分片外候选。
+不得调用工具、搜索、补证或修改运行终态。
 """
 
 REPORT_PROMPT = COMMON_RULES + """

find_agent_v2/runtime.py

@@ -22,6 +22,7 @@ from pydantic import BaseModel, Field
 from find_agent_v2.observability import InputSlot, ObagentObserver
 from find_agent_v2.state import NodeRun
 from find_agent_v2.tools import ToolFn
+from find_agent_v2.tools import EvaluationBatch
 from supply_agent.config import Settings, get_settings
 
 
@@ -282,6 +283,65 @@ class FindAgentNodeHost:
                 tool_calls_made=int(process["tool_calls_made"]),
             )
 
+    async def run_evaluation(
+        self,
+        *,
+        round_index: int,
+        system_prompt: str,
+        user_content: str,
+        slots: tuple[InputSlot, ...] = (),
+        branch_key: str = "",
+    ) -> tuple[NodeRun, list[dict[str, Any]]]:
+        """Get one structured decision batch; the graph owns validation and persistence."""
+        model_name = self.models_by_role.get("evaluator", self.default_model)
+        agent = create_agent(
+            model=self._model("evaluator"),
+            tools=[],
+            system_prompt=system_prompt,
+            middleware=self._middleware("evaluator"),
+            response_format=EvaluationBatch,
+            name="find_agent_v2_evaluator",
+        )
+        with self.observer.node(node="evaluator", branch_key=branch_key) as observation:
+            actual_user_content = observation.declare(
+                fallback=user_content,
+                system_prompt=system_prompt,
+                slots=slots,
+                tools=(),
+                model=model_name,
+            )
+            result = await agent.ainvoke(
+                {"messages": [{"role": "user", "content": actual_user_content}]},
+                config={"recursion_limit": 24},
+            )
+            messages: list[BaseMessage] = list(result.get("messages") or [])
+            usage = _usage(messages)
+            for key in ("input_tokens", "output_tokens", "total_tokens"):
+                self.usage[key] = int(self.usage[key]) + int(usage[key])
+            self.usage["cost"] = round(float(self.usage["cost"]) + float(usage["cost"]), 8)
+            structured = result.get("structured_response")
+            batch = structured if isinstance(structured, EvaluationBatch) else EvaluationBatch.model_validate(structured)
+            items = [item.model_dump(exclude_none=True) for item in batch.items]
+            process = {
+                "messages": [_message_dict(message) for message in messages],
+                "events": _events(messages),
+                "usage": usage,
+                "iterations": sum(isinstance(message, AIMessage) for message in messages),
+                "tool_calls_made": 0,
+                "structured_items": items,
+            }
+            observation.record_react(output=process, ok=True)
+            observation.set_output({"结构化评估": items, **process}, ok=True)
+            last_ai = next((message for message in reversed(messages) if isinstance(message, AIMessage)), None)
+            run = NodeRun(
+                node="evaluator",
+                round_index=round_index,
+                content=str(last_ai.content if last_ai is not None else ""),
+                iterations=int(process["iterations"]),
+                tool_calls_made=0,
+            )
+            return run, items
+
 
 def normalize_models(
     *,

find_agent_v2/tools.py

@@ -6,6 +6,8 @@ import json
 from collections.abc import Callable, Iterable
 from typing import Any
 
+from pydantic import BaseModel, Field, model_validator
+
 from find_agent_v2.providers import (
     fetch_details,
     fetch_portraits,
@@ -19,6 +21,79 @@ from supply_agent.tools.registry import ToolRegistry
 ToolFn = Callable[..., Any]
 
 
+class CandidateEvaluation(BaseModel):
+    """One candidate decision. Prefer candidate_id; aweme_id is a safe fallback."""
+
+    candidate_id: int | None = Field(
+        default=None,
+        description="数据库候选编号，即输入中的 candidate_id；不要填写 aweme_id。",
+    )
+    aweme_id: str | None = Field(
+        default=None,
+        description="抖音视频号；仅在无法填写 candidate_id 时作为兼容定位字段。",
+    )
+    relevance_score: float = Field(ge=0, le=1, description="R 相关性评分，0~1。")
+    elder_score: float = Field(ge=0, le=1, description="E 中老年适配评分，0~1。")
+    share_score: float = Field(ge=0, le=1, description="S 传播力评分，0~1。")
+    value_score: float = Field(ge=0, le=1, description="V 综合价值评分，0~1。")
+    decision_bucket: str = Field(description="只能是 primary 或 rejected。")
+    decision_reason: str = Field(min_length=1, description="基于真实证据的中文判断理由。")
+    reject_reason_code: str | None = None
+
+    @model_validator(mode="after")
+    def validate_identity_and_bucket(self):
+        if self.candidate_id is None and not str(self.aweme_id or "").strip():
+            raise ValueError("candidate_id 和 aweme_id 至少提供一个")
+        if self.decision_bucket not in {"primary", "rejected"}:
+            raise ValueError("decision_bucket 只能是 primary/rejected")
+        return self
+
+
+class EvaluationBatch(BaseModel):
+    """Structured evaluator output consumed and persisted by the host."""
+
+    items: list[CandidateEvaluation] = Field(
+        min_length=1,
+        description="当前 Worker 分片内全部候选的评估结果，每个候选恰好一条。",
+    )
+
+
+def normalize_evaluation_items(
+    items: list[dict[str, Any] | CandidateEvaluation],
+    *,
+    allowed_candidates: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    """Validate a complete worker shard and resolve aweme_id without widening access."""
+    allowed_ids = {int(item["candidate_id"]) for item in allowed_candidates}
+    aweme_to_id = {
+        str(item.get("aweme_id") or ""): int(item["candidate_id"])
+        for item in allowed_candidates
+    }
+    normalized: list[dict[str, Any]] = []
+    for raw in items:
+        value = raw if isinstance(raw, CandidateEvaluation) else CandidateEvaluation.model_validate(raw)
+        candidate_id = value.candidate_id
+        if candidate_id is None:
+            candidate_id = aweme_to_id.get(str(value.aweme_id or ""))
+        if candidate_id not in allowed_ids:
+            raise ValueError(
+                "评估项不属于当前 Worker 分片；"
+                f"candidate_id={candidate_id}, aweme_id={value.aweme_id}, "
+                f"allowed_candidate_ids={sorted(allowed_ids)}"
+            )
+        payload = value.model_dump(exclude_none=True)
+        payload["candidate_id"] = candidate_id
+        payload.pop("aweme_id", None)
+        normalized.append(payload)
+    requested = [int(item["candidate_id"]) for item in normalized]
+    if len(requested) != len(set(requested)):
+        raise ValueError("同一 candidate_id 不能重复评估")
+    missing = allowed_ids - set(requested)
+    if missing:
+        raise ValueError(f"必须一次评估完整分片，缺少 candidate_ids={sorted(missing)}")
+    return normalized
+
+
 def bound_candidate_tools(
     functions: tuple[ToolFn, ...], *, run_id: str, candidate_ids: list[int],
 ) -> tuple[ToolFn, ...]:
@@ -64,16 +139,20 @@ def bound_candidate_tools(
 
         def make_sync(fn: ToolFn, fn_name: str, worker_run_id: str, worker_allowed: set[int]):
             @tool(name=fn_name, description=getattr(fn, "_tool_description", ""))
-            def bound_sync(run_id: str, items: list[dict[str, Any]]) -> str:
+            def bound_sync(run_id: str, items: list[CandidateEvaluation]) -> str:
                 if run_id != worker_run_id:
-                    return json.dumps({"error": "run_id 不属于当前 Worker"}, ensure_ascii=False)
-                requested = {int(item.get("candidate_id") or 0) for item in items}
-                if not requested or not requested <= worker_allowed:
-                    return json.dumps({
-                        "error": "items 超出当前 Worker 分片",
-                        "allowed_candidate_ids": sorted(worker_allowed),
-                    }, ensure_ascii=False)
-                return fn(run_id=run_id, items=items)
+                    raise ValueError("run_id 不属于当前 Worker")
+                state = get_find_agent_v2_service().get_full_state(
+                    run_id, pending_only=True,
+                )
+                allowed_candidates = [
+                    item for item in state.get("candidates", [])
+                    if int(item["candidate_id"]) in worker_allowed
+                ]
+                normalized = normalize_evaluation_items(
+                    items, allowed_candidates=allowed_candidates,
+                )
+                return fn(run_id=run_id, items=normalized)
 
             return bound_sync
 
@@ -183,9 +262,10 @@ async def fetch_candidate_portraits_v2(run_id: str, candidate_ids: list[int]) ->
 
 
 @tool
-def evaluate_candidates_v2(run_id: str, items: list[dict[str, Any]]) -> str:
-    """写入 R/E/S/V 与 primary/rejected；程序会对 primary 强制执行 P0 门禁。"""
-    updated = get_find_agent_v2_service().evaluate(run_id, items)
+def evaluate_candidates_v2(run_id: str, items: list[CandidateEvaluation]) -> str:
+    """按 candidate_id 写入完整分片的 R/E/S/V 与分池；不要用 aweme_id 代替。"""
+    payload = [item.model_dump(exclude_none=True) if isinstance(item, CandidateEvaluation) else item for item in items]
+    updated = get_find_agent_v2_service().evaluate(run_id, payload)
     return json.dumps({"run_id": run_id, "updated": updated}, ensure_ascii=False)
 
 

tests/supply_agent/test_find_agent_v2.py

@@ -39,10 +39,12 @@ from find_agent_v2.prompts import COMMON_RULES
 from find_agent_v2.providers import normalize_age_pair
 from find_agent_v2.state import DiscoverySnapshot, FindAgentState, NodeRun
 from find_agent_v2.tools import (
+    CandidateEvaluation,
     EVALUATION_TOOLS,
     EVIDENCE_TOOLS,
     REPORT_TOOLS,
     SEARCH_TOOLS,
+    normalize_evaluation_items,
 )
 
 
@@ -50,6 +52,54 @@ def _names(functions) -> set[str]:
     return {getattr(fn, "_tool_name", fn.__name__) for fn in functions}
 
 
+def _evaluation(**overrides):
+    value = {
+        "candidate_id": 11,
+        "relevance_score": 0.9,
+        "elder_score": 0.8,
+        "share_score": 0.7,
+        "value_score": 0.85,
+        "decision_bucket": "primary",
+        "decision_reason": "证据充分且满足推荐要求",
+    }
+    value.update(overrides)
+    return value
+
+
+def test_evaluation_items_resolve_aweme_id_only_inside_worker_shard() -> None:
+    allowed = [
+        {"candidate_id": 11, "aweme_id": "video-11"},
+        {"candidate_id": 12, "aweme_id": "video-12"},
+    ]
+    normalized = normalize_evaluation_items([
+        _evaluation(candidate_id=None, aweme_id="video-11"),
+        _evaluation(candidate_id=12, aweme_id=None, decision_bucket="rejected"),
+    ], allowed_candidates=allowed)
+
+    assert [item["candidate_id"] for item in normalized] == [11, 12]
+    assert all("aweme_id" not in item for item in normalized)
+
+
+@pytest.mark.parametrize("items, message", [
+    ([_evaluation()], "缺少 candidate_ids=[12]"),
+    ([_evaluation(), _evaluation()], "不能重复评估"),
+    ([_evaluation(candidate_id=99), _evaluation(candidate_id=12)], "不属于当前 Worker 分片"),
+])
+def test_evaluation_items_fail_fast_on_incomplete_duplicate_or_foreign_items(items, message) -> None:
+    allowed = [
+        {"candidate_id": 11, "aweme_id": "video-11"},
+        {"candidate_id": 12, "aweme_id": "video-12"},
+    ]
+    with pytest.raises(ValueError, match=message.replace("[", r"\[").replace("]", r"\]")):
+        normalize_evaluation_items(items, allowed_candidates=allowed)
+
+
+def test_candidate_evaluation_schema_rejects_missing_identity() -> None:
+    payload = _evaluation(candidate_id=None)
+    with pytest.raises(ValueError, match="至少提供一个"):
+        CandidateEvaluation.model_validate(payload)
+
+
 def test_v2_orm_uses_only_new_table_namespace() -> None:
     assert {
         FindAgentV2Run.__tablename__,