基于 OpenAI Codex 构建具有自我改进能力的税务智能体

[输入文档与规则]
        │
        ▼
[Codex 生成解析代码]
        │
        ▼
[在沙箱环境中执行]
        │
   ┌────┴────┐
   ▼         ▼
[执行成功]  [执行失败 / 异常]
   │         │
   │         ▼
   │       [提取错误堆栈与上下文]
   │         │
   │         ▼
   │       [Codex 自我修复代码] ──┐
   │         ▲                    │
   │         └────────────────────┘
   ▼
[输出最终结构化数据与经验证的代码]

import sys
import traceback
import openai

# 配置 API 访问（在生产环境中推荐使用 n1n.ai 统一网关以确保高可用性）
openai.api_base = "https://api.n1n.ai/v1"
openai.api_key = "your-n1n-api-key"

def generate_parser_code(prompt: str, error_context: str = None) -> str:
    system_message = (
        "你是一位精通金融软件工程的专家。请根据要求编写干净、健壮的 Python 代码来解析税务文档。"
        "请仅返回包裹在代码块中的可执行 Python 代码，不要在代码块之外包含任何解释文字。"
    )

    user_prompt = prompt
    if error_context:
        user_prompt += f"\n\n代码在执行时发生了错误：\n{error_context}\n\n请修正代码以完美兼容此情况。"

    response = openai.ChatCompletion.create(
        model="gpt-4-turbo",
        messages=[
            {"role": "system", "content": system_message},
            {"role": "user", "content": user_prompt}
        ],
        temperature=0.1
    )

    # 从 Markdown 中提取代码块
    raw_content = response.choices[0].message.content
    code = raw_content.replace("```python", "").replace("```", "").strip()
    return code

def execute_sandbox(code_str: str, data_input: dict) -> dict:
    # 定义受限的执行上下文
    local_vars = {"data": data_input, "result": None}
    global_vars = {}

    # 在受控环境中执行生成的代码
    exec(code_str, global_vars, local_vars)
    return local_vars.get("result")

# 定义税务 Schema 和解析指令
tax_data = {
    "raw_text": "W-2 Form: Box 1 (Wages): 85000.00 | Box 2 (Federal Tax): 12000.00 | Box 3 (Social Security): 5270.00"
}

parsing_instructions = """
编写一段 Python 脚本，解析包含 'raw_text' 的 `data` 字典。
提取 'wages'、'federal_tax' 和 'social_security' 并将其转换为浮点数。
验证 'federal_tax' 是否小于 'wages' 的 30%。如果不是，抛出 ValueError('Tax rate exceeds maximum threshold')。
将最终解析字典存入变量 `result`，格式为 `{'wages': float, 'fed_tax': float, 'ss_tax': float}`。
"""

# 自我改进循环执行
max_attempts = 3
current_attempt = 0
error_log = None
success = False
compiled_code = ""

print("开始运行自我改进税务智能体闭环...")

while current_attempt < max_attempts and not success:
    current_attempt += 1
    print(f"\n--- 尝试次数 {current_attempt} ---")

    try:
        # 从 Codex/GPT-4 生成代码
        compiled_code = generate_parser_code(parsing_instructions, error_log)
        print("生成的代码：\n", compiled_code)

        # 在沙箱中执行代码
        parsed_result = execute_sandbox(compiled_code, tax_data)
        print("执行成功！输出结果：", parsed_result)
        success = True
    except Exception as e:
        exc_type, exc_value, exc_tb = sys.exc_info()
        tb_str = "".join(traceback.format_exception(exc_type, exc_value, exc_tb))
        print(f"执行失败：{e}")
        # 记录错误上下文以备下一次迭代使用
        error_log = f"代码：\n{compiled_code}\n\n错误堆栈：\n{tb_str}"

if success:
    print("\n智能体成功生成了具有自我纠错能力的解析器！")
else:
    print("\n在达到最大尝试次数后，智能体未能自动解决该解析问题。")