"""One autonomous but policy-guarded business round compiled as LangGraph.""" from __future__ import annotations import asyncio import json import re from typing import Any, Protocol from langgraph.graph import END, START, StateGraph from find_agent_v2.context import assignment_slots, render_assignment from find_agent_v2.gates import evaluate_candidate_gate from find_agent_v2.observability import InputSlot, NullObserver, graph_spec_for from find_agent_v2.prompts import ( EVALUATOR_PROMPT, EVIDENCE_PROMPT, PLANNER_PROMPT, SEARCH_PROMPT, SUPERVISOR_PROMPT, ) from find_agent_v2.service import FindAgentV2Service from find_agent_v2.state import ( EvidenceAssignment, EvaluationAssignment, ExecutionPlan, FindAgentGraphState, FindAgentState, NodeRun, PlannerAssignment, PlanningDecision, SearchAssignment, SearchTask, SupervisorAssignment, SupervisorDecision, ) from find_agent_v2.tools import ( EVALUATION_TOOLS, EVIDENCE_TOOLS, SEARCH_TOOLS, ToolFn, bound_candidate_tools, normalize_evaluation_items, ) class NodeRunner(Protocol): async def run_node( self, *, node: str, round_index: int, system_prompt: str, user_content: str, tools: tuple[ToolFn, ...] = (), max_iterations: int = 12, slots: tuple[InputSlot, ...] = (), branch_key: str = "", allow_delegation: bool = True, ) -> NodeRun: ... class FindAgentRoundGraph: """Supervisor-directed graph with deterministic policy approval.""" def __init__( self, *, service: FindAgentV2Service, runner: NodeRunner, observer=None, max_actions: int = 16, max_search_actions: int = 3, ) -> None: self.service = service self.runner = runner self.observer = observer or NullObserver() self.max_actions = max(4, int(max_actions)) self.max_search_actions = max(1, int(max_search_actions)) self.app = self._build_graph() self.obagent_spec = graph_spec_for(self.app) @staticmethod def _assignment_input(assignment, *, source: str) -> tuple[str, tuple[InputSlot, ...]]: return render_assignment(assignment), assignment_slots(assignment, source=source) def _supervisor_state(self, state: FindAgentGraphState) -> dict[str, Any]: """Compact progress projection; routing never needs full candidate payloads.""" run = self.service.require_run(state["run_id"]) return { "run": { key: run.get(key) for key in ( "status", "outcome_status", "current_round", "search_count", "candidate_count", "valid_primary_count", ) }, "searches": [ { key: item.get(key) for key in ( "keyword", "provider", "status", "result_count", "has_more", ) } for item in self.service.get_search_summaries(state["run_id"]) ], "candidate_progress": self.service.get_candidate_progress(state["run_id"]), } def _deterministic_supervisor_decision( self, state: FindAgentGraphState, ) -> SupervisorDecision | None: """Route state-machine steps in code; reserve the LLM for exploration choices.""" progress = self.service.get_candidate_progress(state["run_id"]) pending = int(progress.get("pending_count") or 0) if not pending: return None detail_pending = int(progress.get("detail_pending_count") or 0) portrait_pending = int(progress.get("portrait_pending_count") or 0) worker_count = max(1, min(8, int(state.get("worker_count") or 4))) if detail_pending or portrait_pending: scope = ( "both" if detail_pending and portrait_pending else "detail" if detail_pending else "portrait" ) return SupervisorDecision( next_action="evidence", reason="宿主检测到待补证候选，执行确定性证据路由", worker_count=worker_count, evidence_scope=scope, ) return SupervisorDecision( next_action="evaluator", reason="宿主确认全部证据尝试已结束，执行确定性评估路由", worker_count=worker_count, evidence_scope="both", ) @staticmethod def _supervisor_plan_projection(execution_plan: ExecutionPlan) -> dict[str, Any]: """Exclude executed SearchTask details already represented by search summaries.""" return { "schema_version": execution_plan.schema_version, "demand_brief": execution_plan.demand_brief.model_dump(mode="json"), "evaluation_brief": execution_plan.evaluation_brief.model_dump(mode="json"), } @staticmethod def _evaluation_candidate_projection(item: dict[str, Any]) -> dict[str, Any]: keys = ( "candidate_id", "aweme_id", "title", "video_url", "source_keywords", "tags", "publish_at", "duration_seconds", "play_count", "like_count", "comment_count", "collect_count", "share_count", "content_50_plus_ratio", "account_50_plus_ratio", "detail_status", "portrait_status", ) return {key: item.get(key) for key in keys if item.get(key) is not None} def _video_understanding_ids( self, state: FindAgentGraphState, items: list[dict[str, Any]], ) -> list[int]: """Only candidates passing every deterministic hard gate may use video understanding.""" run = self.service.require_run(state["run_id"]) rules = run.get("rule_config") or {} selected: list[int] = [] for item in items: gate = evaluate_candidate_gate(item, rules) checks = list(gate.get("checks") or []) hard_gate_passed = gate.get("status") == "pass" and all( check.get("status") == "pass" and not check.get("compensated") for check in checks ) if str(item.get("video_url") or "").strip() and hard_gate_passed: selected.append(int(item["candidate_id"])) return selected @staticmethod def _parse_supervisor(content: str) -> dict[str, Any]: text = content.strip() fenced = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.S) if fenced: text = fenced.group(1) else: start, end = text.find("{"), text.rfind("}") if start >= 0 and end > start: text = text[start : end + 1] try: value = json.loads(text) return value if isinstance(value, dict) else {} except (TypeError, ValueError): return {} def _approve_action( self, state: FindAgentGraphState, proposal: dict[str, Any], ) -> tuple[str, str, int, str]: """Turn an LLM proposal into a safe, executable transition.""" progress = self.service.get_candidate_progress(state["run_id"]) pending_count = int(progress.get("pending_count") or 0) missing_detail = int(progress.get("detail_pending_count") or 0) > 0 missing_portrait = int(progress.get("portrait_pending_count") or 0) > 0 proposed = str(proposal.get("next_action") or "").lower() reason = str(proposal.get("reason") or "") searches = int(state.get("search_actions") or 0) actions = int(state.get("action_count") or 0) try: worker_count = max(1, min(8, int(proposal.get("worker_count") or 4))) except (TypeError, ValueError): worker_count = 4 scope = str(proposal.get("evidence_scope") or "both").lower() if scope not in {"detail", "portrait", "both"}: scope = "both" if actions > self.max_actions: if not pending_count: return "finish", "安全收敛动作已完成", worker_count, scope raise RuntimeError( f"Supervisor 安全收敛动作未产生进展：actions={actions}, pending={pending_count}" ) if actions == self.max_actions: if pending_count and not (missing_detail or missing_portrait): return "evaluator", "动作预算耗尽，强制消费待评估候选", worker_count, scope if pending_count: return "evidence", "动作预算耗尽，强制补齐缺失证据", worker_count, "both" return "finish", "达到单轮动作安全上限", worker_count, scope if pending_count: if missing_detail or missing_portrait: if scope == "detail" and not missing_detail: scope = "portrait" elif scope == "portrait" and not missing_portrait: scope = "detail" return "evidence", reason or "存在缺失证据，框架要求先补证", worker_count, scope return "evaluator", reason or "候选证据已处理，进入评估", worker_count, scope if searches == 0: return "search", reason or "本轮尚未搜索，框架要求先建立候选池", worker_count, scope if proposed == "search" and searches < self.max_search_actions: return "search", reason or "Supervisor 判断继续搜索仍有信息增益", worker_count, scope return "finish", reason or "没有待处理候选，结束本轮", worker_count, scope async def _supervisor(self, state: FindAgentGraphState) -> dict[str, Any]: execution_plan = state.get("execution_plan") observed_decision: dict[str, Any] = {} if execution_plan is None: assignment = PlannerAssignment( run_id=state["run_id"], round_index=state["round_index"], raw_demand=state["user_input"], ) user_content, slots = self._assignment_input( assignment, source="find_agent_v2_run.input_json", ) structured_runner = getattr(self.runner, "run_planning", None) if callable(structured_runner): run, planning = await structured_runner( round_index=state["round_index"], system_prompt=PLANNER_PROMPT, user_content=user_content, slots=slots, ) else: run = await self.runner.run_node( node="supervisor", round_index=state["round_index"], system_prompt=PLANNER_PROMPT, user_content=user_content, tools=(), max_iterations=2, slots=slots, allow_delegation=False, ) planning = PlanningDecision.model_validate(self._parse_supervisor(run.content)) execution_plan = planning.execution_plan proposal = planning.model_dump(mode="json", exclude={"execution_plan"}) else: assignment = SupervisorAssignment( run_id=state["run_id"], round_index=state["round_index"], execution_plan=self._supervisor_plan_projection(execution_plan), execution_state=self._supervisor_state(state), ) user_content, slots = self._assignment_input( assignment, source="ExecutionPlan compact projection + database aggregates", ) structured_runner = getattr(self.runner, "run_supervision", None) host_runner = getattr(self.runner, "run_host_supervision", None) def approve_for_observation(supervision: SupervisorDecision) -> dict[str, Any]: proposal_payload = supervision.model_dump(mode="json") approved, approved_reason, approved_workers, approved_scope = ( self._approve_action(state, proposal_payload) ) decision_payload = { "step": int(state.get("supervisor_step") or 0) + 1, "proposed_action": proposal_payload.get("next_action"), "approved_action": approved, "overridden": proposal_payload.get("next_action") != approved, "reason": approved_reason, "worker_count": approved_workers, "evidence_scope": approved_scope, } observed_decision.update(decision_payload) return {"Supervisor决策校验": decision_payload} deterministic = ( self._deterministic_supervisor_decision(state) if callable(host_runner) else None ) if deterministic is not None: run, supervision = await host_runner( round_index=state["round_index"], decision=deterministic, system_prompt=SUPERVISOR_PROMPT, user_content=user_content, slots=slots, output_enricher=approve_for_observation, ) elif callable(structured_runner): run, supervision = await structured_runner( round_index=state["round_index"], system_prompt=SUPERVISOR_PROMPT, user_content=user_content, slots=slots, output_enricher=approve_for_observation, ) else: run = await self.runner.run_node( node="supervisor", round_index=state["round_index"], system_prompt=SUPERVISOR_PROMPT, user_content=user_content, tools=(), max_iterations=2, slots=slots, allow_delegation=False, ) supervision = SupervisorDecision.model_validate( self._parse_supervisor(run.content), ) proposal = supervision.model_dump(mode="json") if supervision.additional_search_tasks: known = {item.task_id for item in execution_plan.search_tasks} merged = list(execution_plan.search_tasks) for item in supervision.additional_search_tasks: if item.task_id not in known and len(merged) < 24: merged.append(item) known.add(item.task_id) if len(merged) != len(execution_plan.search_tasks): execution_plan = type(execution_plan).model_validate( { **execution_plan.model_dump(), "search_tasks": merged, } ) if observed_decision: decision = observed_decision action = str(decision["approved_action"]) reason = str(decision["reason"]) workers = int(decision["worker_count"]) scope = str(decision["evidence_scope"]) else: action, reason, workers, scope = self._approve_action(state, proposal) decision = { "step": int(state.get("supervisor_step") or 0) + 1, "proposed_action": proposal.get("next_action"), "approved_action": action, "overridden": proposal.get("next_action") != action, "reason": reason, "worker_count": workers, "evidence_scope": scope, } self.service.update_round( state["run_id"], state["round_index"], phase="planning", plan=execution_plan.model_dump_json(exclude_none=True), ) return { "phase": "planning", "execution_plan": execution_plan, "approved_action": action, "supervisor_step": decision["step"], "worker_count": workers, "evidence_scope": scope, "decision_history": [*state.get("decision_history", []), decision], "node_runs": [*state.get("node_runs", []), run], } @staticmethod def _search_task_already_executed( task: SearchTask, existing: set[tuple[str, str, str]], ) -> bool: keyword = task.keyword reason = task.query_reason if not task.provider: return (keyword, "internal_keyword", reason) in existing return (keyword, task.provider, reason) in existing async def _search(self, state: FindAgentGraphState) -> dict[str, Any]: self.service.update_round(state["run_id"], state["round_index"], phase="searching") execution_plan = state.get("execution_plan") if execution_plan is None: raise RuntimeError("Search 缺少已校验的 ExecutionPlan") existing = { ( str(item.get("keyword") or ""), str(item.get("provider") or ""), str(item.get("query_reason") or ""), ) for item in self.service.get_search_summaries(state["run_id"]) } tasks = [ item for item in execution_plan.search_tasks if not self._search_task_already_executed(item, existing) ][:6] assignment = SearchAssignment( run_id=state["run_id"], round_index=state["round_index"], tasks=tasks, ) if not tasks: return { "phase": "searching", "search_actions": int(state.get("search_actions") or 0) + 1, "action_count": int(state.get("action_count") or 0) + 1, "node_runs": [ *state.get("node_runs", []), NodeRun("search", state["round_index"], "没有未执行的搜索任务", 0, 0), ], } user_content, slots = self._assignment_input( assignment, source="ExecutionPlan.search_tasks", ) host_runner = getattr(self.runner, "run_search_assignment", None) if callable(host_runner): run = await host_runner( round_index=state["round_index"], assignment=assignment, system_prompt=SEARCH_PROMPT, user_content=user_content, slots=slots, ) else: run = await self.runner.run_node( node="search", round_index=state["round_index"], system_prompt=SEARCH_PROMPT, user_content=user_content, tools=SEARCH_TOOLS, max_iterations=10, slots=slots, ) return { "phase": "searching", "search_actions": int(state.get("search_actions") or 0) + 1, "action_count": int(state.get("action_count") or 0) + 1, "node_runs": [*state.get("node_runs", []), run], } async def _evidence(self, state: FindAgentGraphState) -> dict[str, Any]: self.service.update_round(state["run_id"], state["round_index"], phase="evidence") scope = state.get("evidence_scope", "both") worker_count = max(1, min(8, int(state.get("worker_count") or 4))) batch_limit = worker_count * 8 evidence_types = ( ("detail",) if scope == "detail" else ("portrait",) if scope == "portrait" else ("detail", "portrait") ) semaphore = asyncio.Semaphore(worker_count) node_runs = list(state.get("node_runs", [])) batch_index = 0 async def run_shard( branch_key: str, evidence_type: str, items: list[dict[str, Any]], ) -> NodeRun: candidate_ids = [int(item["candidate_id"]) for item in items] assignment = EvidenceAssignment( run_id=state["run_id"], round_index=state["round_index"], candidate_ids=candidate_ids, evidence_type=evidence_type, candidates=items, ) user_content, slots = self._assignment_input( assignment, source="host evidence shard", ) selected = ( (EVIDENCE_TOOLS[0], EVIDENCE_TOOLS[2]) if evidence_type == "detail" else (EVIDENCE_TOOLS[1], EVIDENCE_TOOLS[2]) ) async with semaphore: host_runner = getattr(self.runner, "run_evidence_assignment", None) if callable(host_runner): return await host_runner( round_index=state["round_index"], assignment=assignment, system_prompt=EVIDENCE_PROMPT, user_content=user_content, slots=slots, branch_key=branch_key, ) return await self.runner.run_node( node="evidence", round_index=state["round_index"], system_prompt=EVIDENCE_PROMPT, user_content=user_content, tools=bound_candidate_tools( selected, run_id=state["run_id"], candidate_ids=candidate_ids, ), max_iterations=12, slots=slots, branch_key=branch_key, allow_delegation=False, ) while True: batch_index += 1 jobs: list[tuple[str, list[dict[str, Any]]]] = [] selected_ids: dict[str, list[int]] = {} for evidence_type in evidence_types: candidate_ids = self.service.list_pending_evidence_ids( state["run_id"], evidence_type, limit=batch_limit, ) selected_ids[evidence_type] = candidate_ids candidates = self.service.candidate_inputs(state["run_id"], candidate_ids) jobs.extend( (evidence_type, candidates[index : index + 8]) for index in range(0, len(candidates), 8) ) if not jobs: break runs = await asyncio.gather(*( run_shard( f"{evidence_type}-batch-{batch_index}-shard-{index}", evidence_type, items, ) for index, (evidence_type, items) in enumerate(jobs, start=1) )) node_runs.extend(runs) for evidence_type, candidate_ids in selected_ids.items(): status_key = f"{evidence_type}_status" remaining = [ int(item["candidate_id"]) for item in self.service.candidate_inputs(state["run_id"], candidate_ids) if item.get(status_key) == "pending" ] if remaining: raise RuntimeError( f"证据分批未完整消费：type={evidence_type}, pending={remaining}" ) return { "phase": "evidence", "action_count": int(state.get("action_count") or 0) + 1, "node_runs": node_runs, } async def _evaluator(self, state: FindAgentGraphState) -> dict[str, Any]: self.service.update_round(state["run_id"], state["round_index"], phase="evaluating") node_runs = list(state.get("node_runs", [])) run = self.service.require_run(state["run_id"]) rules = run.get("rule_config") or {} worker_count = max(1, min(8, int(state.get("worker_count") or 4))) batch_limit = worker_count * 8 semaphore = asyncio.Semaphore(worker_count) batch_index = 0 async def run_shard( batch_number: int, index: int, items: list[dict[str, Any]], ) -> NodeRun: candidate_ids = [int(item["candidate_id"]) for item in items] video_understanding_ids = self._video_understanding_ids(state, items) execution_plan = state.get("execution_plan") if execution_plan is None: raise RuntimeError("Evaluator 缺少已校验的 ExecutionPlan") assignment = EvaluationAssignment( run_id=state["run_id"], round_index=state["round_index"], candidate_ids=candidate_ids, evaluation_brief=execution_plan.evaluation_brief, quality_gate_rules=rules, current_datetime=str(rules.get("current_datetime") or ""), timezone=str(rules.get("timezone") or ""), video_understanding_candidate_ids=video_understanding_ids, candidates=[self._evaluation_candidate_projection(item) for item in items], ) user_content, slots = self._assignment_input( assignment, source="ExecutionPlan.evaluation_brief + host candidate shard", ) video_tools = ( bound_candidate_tools( (EVALUATION_TOOLS[0],), run_id=state["run_id"], candidate_ids=video_understanding_ids, ) if video_understanding_ids else () ) 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=user_content, slots=slots, branch_key=( f"step-{state.get('supervisor_step', 0)}-" f"batch-{batch_number}-shard-{index}" ), tools=video_tools, ) 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"], system_prompt=EVALUATOR_PROMPT, user_content=user_content, tools=( *video_tools, *bound_candidate_tools( EVALUATION_TOOLS[1:], run_id=state["run_id"], candidate_ids=candidate_ids, ), ), max_iterations=12, slots=slots, branch_key=( f"step-{state.get('supervisor_step', 0)}-" f"batch-{batch_number}-shard-{index}" ), allow_delegation=False, ) while True: batch_index += 1 candidate_ids = self.service.list_ready_evaluation_ids( state["run_id"], limit=batch_limit, ) if not candidate_ids: break items = self.service.candidate_inputs(state["run_id"], candidate_ids) eligible: list[dict[str, Any]] = [] gate_failures: list[tuple[int, dict[str, Any]]] = [] for item in items: gate = evaluate_candidate_gate(item, rules) if gate.get("status") == "pass": eligible.append(item) else: gate_failures.append((int(item["candidate_id"]), gate)) self.service.reject_failed_gates(state["run_id"], gate_failures) shards = [eligible[index : index + 8] for index in range(0, len(eligible), 8)] runs = await asyncio.gather(*( run_shard(batch_index, index, shard) for index, shard in enumerate(shards, start=1) )) if shards else [] node_runs.extend(runs) remaining = [ int(item["candidate_id"]) for item in self.service.candidate_inputs(state["run_id"], candidate_ids) if item.get("decision_bucket") == "pending_evaluation" ] if remaining: raise RuntimeError(f"评估分批未完整消费：pending={remaining}") self.service.recount_valid_primary(state["run_id"]) return { "phase": "evaluating", "node_runs": node_runs, "action_count": int(state.get("action_count") or 0) + 1, "evaluator_stagnation": 0, } @staticmethod def _route(state: FindAgentGraphState) -> str: return state.get("approved_action", "finish") def _build_graph(self): builder = StateGraph(FindAgentGraphState) builder.add_node("supervisor", self._supervisor) builder.add_node("search", self._search) builder.add_node("evidence", self._evidence) builder.add_node("evaluator", self._evaluator) builder.add_edge(START, "supervisor") builder.add_conditional_edges( "supervisor", self._route, { "search": "search", "evidence": "evidence", "evaluator": "evaluator", "finish": END, }, ) builder.add_edge("search", "supervisor") builder.add_edge("evidence", "supervisor") builder.add_edge("evaluator", "supervisor") return builder.compile() async def invoke(self, state: FindAgentState) -> FindAgentState: graph_state: FindAgentGraphState = { "run_id": state.run_id, "user_input": state.user_input, "round_index": state.round_index, "execution_plan": state.execution_plan, "phase": state.phase, "node_runs": [], "snapshot": state.snapshot, "supervisor_step": 0, "action_count": 0, "search_actions": 0, "worker_count": 4, "evidence_scope": "both", "decision_history": [], "evaluator_stagnation": 0, } with self.observer.round( round_index=state.round_index, spec=self.obagent_spec, ) as round_observation: output = await self.app.ainvoke( graph_state, config={"recursion_limit": max(64, self.max_actions * 4)}, ) state.execution_plan = output.get("execution_plan") state.node_runs.extend(output.get("node_runs") or []) state.snapshot = self.service.snapshot(state.run_id) state.phase = "done" self.service.update_round( state.run_id, state.round_index, phase="done", status="done", snapshot=state.snapshot, ) round_observation.set_output( { "状态快照": state.snapshot.__dict__, "本轮计划": ( state.execution_plan.model_dump(mode="json") if state.execution_plan is not None else {} ), "Supervisor决策轨迹": output.get("decision_history") or [], }, ok=True, ) return state