用 Python 构建基于反射循环的 AI 智能体

import json
from typing import Any, Callable
import httpx

MAX_ITERATIONS = 4

def call_llm(prompt: str, system: str = "") -> str:
    # 在生产环境中，建议通过 n1n.ai 等统一网关路由调用
    # 这样只需一个 API 密钥即可访问 Claude、DeepSeek 或 OpenAI 模型
    response = httpx.post(
        "https://api.example-llm.com/v1/messages",
        headers={"x-api-key": "YOUR_KEY", "Content-Type": "application/json"},
        json={
            "model": "your-model",
            "max_tokens": 1024,
            "system": system,
            "messages": [{"role": "user", "content": prompt}],
        },
        timeout=30,
    )
    response.raise_for_status()
    return response.json()["content"][0]["text"]

class ReflectionAgent:
    def __init__(
        self,
        task_prompt: str,
        critique_prompt: str,
        validator: Callable[[str], tuple[bool, str]],
    ):
        self.task_prompt = task_prompt
        self.critique_prompt = critique_prompt
        self.validator = validator

    def run(self, user_input: str) -> dict[str, Any]:
        history: list[dict] = []
        attempt = 0

        while attempt < MAX_ITERATIONS:
            attempt += 1
            context = f"User input: {user_input}"
            if history:
                last = history[-1]
                context += (
                    f"\n\nPrevious attempt:\n{last['output']}"
                    f"\nCritique:\n{last['critique']}"
                )

            # 1. 生成步骤
            output = call_llm(f"{self.task_prompt}\n\n{context}")

            # 2. 确定性验证步骤
            ok, critique = self.validator(output)

            history.append(
                {"attempt": attempt, "output": output, "critique": critique, "passed": ok}
            )

            if ok:
                return {
                    "success": True,
                    "output": output,
                    "iterations": attempt,
                    "history": history
                }

            # 3. 语义批判步骤（仅在验证失败时运行）
            llm_critique = call_llm(
                f"{self.critique_prompt}\n\nOutput to critique:\n{output}",
                system="Be specific about what is wrong. Do not repeat the corrected version.",
            )
            history[-1]["critique"] = llm_critique

        return {
            "success": False,
            "output": history[-1]["output"],
            "iterations": attempt,
            "history": history
        }

import jsonschema

EXPECTED_SCHEMA = {
    "type": "object",
    "required": ["summary", "severity", "cve_ids"],
    "properties": {
        "summary": {"type": "string", "minLength": 10},
        "severity": {"type": "string", "enum": ["low", "medium", "high", "critical"]},
        "cve_ids": {
            "type": "array",
            "items": {"type": "string", "pattern": "^CVE-\\d{4}-\\d+$"},
        },
    },
}

def validate_security_report(text: str) -> tuple[bool, str]:
    clean = text.strip()
    # 如果 LLM 用 Markdown 代码块包裹了 JSON，进行剥离
    if clean.startswith("```json"):
        clean = "\n".join(clean.split("\n")[1:])
    if clean.startswith("```"):
        clean = "\n".join(clean.split("\n")[1:])
    if clean.endswith("```"):
        clean = "\n".join(clean.split("\n")[:-1])
    clean = clean.strip()

    try:
        data = json.loads(clean)
    except json.JSONDecodeError as exc:
        return False, f"Invalid JSON: \{exc\}"

    try:
        jsonschema.validate(data, EXPECTED_SCHEMA)
    except jsonschema.ValidationError as exc:
        return False, f"Schema violation: \{exc.message\}"

    return True, "OK"

agent = ReflectionAgent(
    task_prompt=(
        "Analyze the following security advisory and return a JSON object "
        "with keys 'summary' (string), 'severity' (low/medium/high/critical), "
        "and 'cve_ids' (array of CVE strings). Return only the JSON, no prose."
    ),
    critique_prompt=(
        "Review this JSON output for accuracy, completeness, and schema compliance. "
        "List each specific violation and explain why it fails."
    ),
    validator=validate_security_report,
)

result = agent.run(
    "CVE-2024-3094: backdoor found in XZ Utils 5.6.0 and 5.6.1 affecting liblzma, "
    "allowing remote code execution on affected Linux distributions."
)
print(json.dumps(result, indent=2))

COST_PER_1K_TOKENS = 0.003  # 根据具体模型定价调整

class BudgetedReflectionAgent(ReflectionAgent):
    def __init__(self, *args, max_cost_usd: float = 0.10, **kwargs):
        super().__init__(*args, **kwargs)
        self.max_cost_usd = max_cost_usd
        self._total_cost = 0.0

    def _check_budget(self, prompt: str) -> None:
        # 简单的 Token 估算启发式算法（1 个英文单词 ≈ 1.3 个 Token）
        # 在生产环境中，应从 API 响应体中读取实际消耗的 Token 数
        estimated_tokens = len(prompt.split()) * 1.3
        projected_cost = (estimated_tokens / 1000) * COST_PER_1K_TOKENS

        if self._total_cost + projected_cost > self.max_cost_usd:
            raise RuntimeError(
                f"Budget exceeded: $\{self._total_cost:.4f\} spent, "
                f"limit $\{self.max_cost_usd\}"
            )
        self._total_cost += projected_cost

CREATE TABLE reflection_logs (
    task_id TEXT,
    attempt INTEGER,
    model TEXT,
    input_tokens INTEGER,
    output_tokens INTEGER,
    passed BOOLEAN,
    critique_text TEXT,
    timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
);