Agentic RAG 工程指南：控制循环、证据与评测

{
  "name": "search_support_runbooks",
  "purpose": "检索当前租户可访问的已审核运维手册",
  "input": {
    "query": "string",
    "tenant_id": "服务端注入",
    "product": "白名单枚举",
    "top_k": "1 到 8 的整数"
  },
  "output": {
    "evidence": [
      {
        "source_id": "不可变文档 ID",
        "revision": "不可变版本",
        "title": "展示标题",
        "excerpt": "长度受限的文本",
        "retrieved_at": "时间戳",
        "acl_scope": "授权范围"
      }
    ]
  },
  "limits": {
    "timeout_ms": 1500,
    "max_result_bytes": 40000
  }
}

answered
clarification_required
insufficient_evidence
conflicting_evidence
authorization_denied
budget_exhausted
dependency_failed
policy_blocked

from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import Any


TERMINAL_REASONS = {
    "answered",
    "clarification_required",
    "insufficient_evidence",
    "conflicting_evidence",
    "authorization_denied",
    "budget_exhausted",
    "dependency_failed",
    "policy_blocked",
}


def is_number(value: Any) -> bool:
    return isinstance(value, (int, float)) and not isinstance(value, bool)


def validate(trace: dict[str, Any]) -> list[str]:
    errors: list[str] = []
    policy = trace.get("policy")
    steps = trace.get("steps")
    result = trace.get("result")
    if not isinstance(policy, dict):
        return ["policy must be an object"]
    if not isinstance(steps, list):
        return ["steps must be a list"]
    if not isinstance(result, dict):
        return ["result must be an object"]

    max_steps = policy.get("max_steps")
    max_tokens = policy.get("max_tokens")
    allowed_tools = policy.get("allowed_tools")
    allowed_sources = policy.get("allowed_source_scopes")
    if not isinstance(max_steps, int) or isinstance(max_steps, bool) or max_steps < 1:
        errors.append("policy.max_steps must be a positive integer")
    if not isinstance(max_tokens, int) or isinstance(max_tokens, bool) or max_tokens < 1:
        errors.append("policy.max_tokens must be a positive integer")
    if not isinstance(allowed_tools, list) or not all(
        isinstance(item, str) and item for item in allowed_tools
    ):
        errors.append("policy.allowed_tools must contain strings")
        allowed_tools = []
    if not isinstance(allowed_sources, list) or not all(
        isinstance(item, str) and item for item in allowed_sources
    ):
        errors.append("policy.allowed_source_scopes must contain strings")
        allowed_sources = []

    if isinstance(max_steps, int) and len(steps) > max_steps:
        errors.append("step budget exceeded")

    total_tokens = 0
    evidence_ids: set[str] = set()
    for index, step in enumerate(steps):
        if not isinstance(step, dict):
            errors.append(f"steps[{index}] must be an object")
            continue
        tool = step.get("tool")
        if tool not in allowed_tools:
            errors.append(f"steps[{index}] uses unauthorized tool {tool!r}")
        tokens = step.get("tokens", 0)
        if not is_number(tokens) or tokens < 0:
            errors.append(f"steps[{index}].tokens must be non-negative")
        else:
            total_tokens += tokens
        for item in step.get("evidence", []):
            if not isinstance(item, dict):
                errors.append(f"steps[{index}] has malformed evidence")
                continue
            evidence_id = item.get("id")
            scope = item.get("scope")
            if not isinstance(evidence_id, str) or not evidence_id:
                errors.append(f"steps[{index}] evidence needs an id")
            else:
                evidence_ids.add(evidence_id)
            if scope not in allowed_sources:
                errors.append(
                    f"steps[{index}] evidence uses unauthorized scope {scope!r}"
                )

    if isinstance(max_tokens, int) and total_tokens > max_tokens:
        errors.append("token budget exceeded")

    reason = result.get("stop_reason")
    if reason not in TERMINAL_REASONS:
        errors.append("result.stop_reason is invalid")
    citations = result.get("citation_ids", [])
    if not isinstance(citations, list) or not all(
        isinstance(item, str) and item for item in citations
    ):
        errors.append("result.citation_ids must contain strings")
        citations = []
    missing = set(citations) - evidence_ids
    if missing:
        errors.append(f"result cites unknown evidence: {sorted(missing)}")
    if reason == "answered" and not citations:
        errors.append("answered result must cite evidence")
    return errors


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("trace", type=Path)
    args = parser.parse_args()
    try:
        value = json.loads(args.trace.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        parser.error(str(error))
    if not isinstance(value, dict):
        parser.error("trace root must be an object")
    errors = validate(value)
    if errors:
        for error in errors:
            print(f"ERROR: {error}")
        raise SystemExit(1)
    print(f"trace passed: {args.trace}")


if __name__ == "__main__":
    main()

python agentic_rag_trace_gate.py trace.json