用 Codex 构建能自我进化的税务 Agent

┌─────────────┐
│  税务规则库  │  ← IRS Publication、州税法、客户专项条款
└─────────────┘
       │
       ▼
┌─────────────┐
│  Codex Agent │  ← 接收任务、调用工具、生成填报代码
└─────────────┘
       │
       ▼
┌─────────────┐
│  执行 & 校验 │  ← 运行生成的代码，与规则库交叉校验
└─────────────┘
       │
       ▼
┌─────────────┐
│  错误回溯库  │  ← 记录每次偏差：哪条规则被误读、哪个字段被漏填
└─────────────┘
       │
       ▼
┌─────────────┐
│  Prompt 修订 │  ← 把回溯结果注入下次 Agent 的上下文
└─────────────┘
       │
       ▼
   回到 Codex Agent

# self_improving_tax_agent.py
# 依赖：pip install openai pyyaml

import json
import yaml
from datetime import datetime
from openai import OpenAI

client = OpenAI()  # 默认读取 OPENAI_API_KEY 环境变量

# ── 1. 规则库（简化示例，实际应从 IRS Publication 等结构化提取）──
RULES = yaml.safe_load("""
standard_deduction:
  single: 14600
  married_joint: 29200
  head_of_household: 21900
tax_brackets_2024:
  single:
    - {limit: 11600, rate: 0.10}
    - {limit: 47150, rate: 0.12}
    - {limit: 100525, rate: 0.22}
    - {limit: 191950, rate: 0.24}
    - {limit: 243725, rate: 0.32}
    - {limit: 609350, rate: 0.35}
    - {limit: null, rate: 0.37}
""")

# ── 2. 错题本（持久化到文件，跨任务累积）──
MISTAKES_FILE = "tax_mistakes.json"

def load_mistakes():
    try:
        with open(MISTAKES_FILE) as f:
            return json.load(f)
    except FileNotFoundError:
        return []

def save_mistake(mistake):
    mistakes = load_mistakes()
    mistakes.append({"timestamp": datetime.now().isoformat(), **mistake})
    with open(MISTAKES_FILE, "w") as f:
        json.dump(mistakes, f, indent=2)

# ── 3. 构建 Agent Prompt ──
def build_prompt(task: str, mistakes: list) -> str:
    mistake_block = ""
    if mistakes:
        # 只注入最近 5 条同类错误，避免上下文膨胀
        recent = mistakes[-5:]
        mistake_block = "\n## 已知错误（必须避免）\n"
        for m in recent:
            mistake_block += f"- {m['description']}\n"

    return f"""你是一名美国税务计算 Agent。根据以下规则和客户数据完成计算任务。

## 税务规则
{json.dumps(RULES, indent=2)}

{mistake_block}

## 任务
{task}

## 输出格式
返回 JSON，包含：
- "adjusted_gross_income": 数值
- "deduction_used": 数值及类型
- "tax_owed": 数值
- "calculation_steps": 计算过程数组
- "confidence_flags": 任何不确定的地方

只输出 JSON，不要附加解释。"""

# ── 4. 调用 Codex / GPT 模型 ──
def run_agent(task: str) -> dict:
    mistakes = load_mistakes()
    prompt = build_prompt(task, mistakes)

    response = client.chat.completions.create(
        model="gpt-4.1",  # 或替换为你的 Codex 端点
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
        temperature=0.1,  # 税务计算需要低随机性
    )

    result = json.loads(response.choices[0].message.content)
    return result

# ── 5. 校验：与规则库交叉比对 ──
def validate(result: dict, task_context: dict) -> list:
    errors = []
    filing_status = task_context.get("filing_status", "single")

    # 校验扣除额是否用了正确标准
    correct_deduction = RULES["standard_deduction"][filing_status]
    if result["deduction_used"] != correct_deduction:
        errors.append({
            "description": f"扣除额应为 {correct_deduction}（{filing_status}），"
                           f"但 Agent 返回了 {result['deduction_used']}",
            "rule": "standard_deduction",
            "expected": correct_deduction,
            "actual": result["deduction_used"],
        })

    # 校验税率档位是否正确应用（简化：只检查最高档）
    agi = result["adjusted_gross_income"]
    brackets = RULES["tax_brackets_2024"][filing_status]
    expected_top_rate = None
    for b in brackets:
        if b["limit"] is None or agi > b["limit"]:
            expected_top_rate = b["rate"]
    # 这里只做示意，完整校验应逐档累加

    return errors

# ── 6. 主流程 ──
def process_tax_filing(income: float, filing_status: str):
    task = f"客户年收入 {income}，报税身份 {filing_status}，使用标准扣除额计算应缴税额。"
    task_context = {"filing_status": filing_status, "income": income}

    # 第一次运行
    result = run_agent(task)
    print("Agent 输出：", json.dumps(result, indent=2))

    # 校验
    errors = validate(result, task_context)
    if errors:
        print(f"发现 {len(errors)} 个错误，写入错题本")
        for e in errors:
            save_mistake(e)
        # 用错题本重新运行——自我进化的核心
        result = run_agent(task)
        print("修正后输出：", json.dumps(result, indent=2))
    else:
        print("校验通过，无需修正")

    return result

# ── 运行示例 ──
if __name__ == "__main__":
    final = process_tax_filing(income=85000, filing_status="single")

export OPENAI_API_KEY="sk-..."
pip install openai pyyaml
python self_improving_tax_agent.py