Codex智能体从零系统学习智能体：从代码生成到自主执行的完整路径

┌─────────────────────────────────┐
│         Agent 架构               │
│  ┌───────────┐  ┌───────────┐   │
│  │  感知      │  │  记忆      │   │
│  │ Perception│  │  Memory   │   │
│  └───────────┘  └───────────┘   │
│  ┌───────────┐  ┌───────────┐   │
│  │  规划      │  │  执行      │   │
│  │ Planning  │  │  Action   │   │
│  └───────────┘  └───────────┘   │
└─────────────────────────────────┘

from openai import OpenAI
client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4",
    messages=[
        {"role": "system", "content": "你是一个代码助手"},
        {"role": "user", "content": "写一个快速排序"}
    ]
)

用户需求 → 模型推理 → 代码生成 → 执行验证 → 反馈修正

tools = [
    {
        "type": "function",
        "function": {
            "name": "execute_code",
            "description": "执行Python代码并返回结果",
            "parameters": {
                "type": "object",
                "properties": {
                    "code": {"type": "string", "description": "要执行的代码"}
                },
                "required": ["code"]
            }
        }
    }
]

response = client.chat.completions.create(
    model="gpt-4",
    messages=messages,
    tools=tools,
    tool_choice="auto"
)

用户问题 → 向量化 → 检索相关文档 → 拼接上下文 → 模型生成

class CodexAgent:
    def __init__(self):
        self.planner = PlannerAgent()
        self.coder = CoderAgent()
        self.reviewer = ReviewerAgent()
        self.executor = ExecutorAgent()

    async def solve(self, task: str):
        plan = await self.planner.decompose(task)
        for step in plan.steps:
            code = await self.coder.generate(step)
            review = await self.reviewer.check(code)
            if review.approved:
                result = await self.executor.run(code)
                if not result.success:
                    code = await self.debugger.fix(code, result.error)
        return result

┌──────────────────────────────────────────────┐
│              Codex Agent System               │
│                                               │
│  ┌─────────┐   ┌─────────┐   ┌─────────┐    │
│  │ 用户接口 │──▶│ 任务规划 │──▶│ 代码生成 │    │
│  └─────────┘   └─────────┘   └─────────┘    │
│       ▲              │              │         │
│       │              ▼              ▼         │
│  ┌─────────┐   ┌─────────┐   ┌─────────┐    │
│  │ 结果反馈 │◀──│ 执行引擎 │◀──│ 代码审查 │    │
│  └─────────┘   └─────────┘   └─────────┘    │
│                      │                        │
│              ┌───────┴───────┐                │
│              ▼               ▼                │
│        ┌─────────┐    ┌─────────┐            │
│        │ 沙箱环境 │    │ 记忆库   │            │
│        └─────────┘    └─────────┘            │
└──────────────────────────────────────────────┘

import docker

def execute_in_sandbox(code: str, timeout: int = 30):
    client = docker.from_env()
    container = client.containers.run(
        "python:3.11-slim",
        command=f"python -c '{code}'",
        mem_limit="512m",
        cpu_period=100000,
        cpu_quota=50000,
        network_disabled=True,
        remove=True,
        timeout=timeout
    )
    return container

import openai
import subprocess
import json

class MiniCodex:
    def __init__(self):
        self.client = openai.OpenAI()
        self.memory = []

    def run(self, task: str, max_iterations: int = 5):
        messages = [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": task}
        ]

        for i in range(max_iterations):
            response = self.client.chat.completions.create(
                model="gpt-4",
                messages=messages,
                tools=TOOLS,
                tool_choice="auto"
            )

            msg = response.choices[0].message
            messages.append(msg)

            if msg.tool_calls:
                for tool_call in msg.tool_calls:
                    result = self.execute_tool(tool_call)
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": result
                    })
            else:
                return msg.content

        return "达到最大迭代次数"

    def execute_tool(self, tool_call):
        name = tool_call.function.name
        args = json.loads(tool_call.function.arguments)

        if name == "execute_python":
            return self.run_python(args["code"])
        elif name == "read_file":
            return self.read_file(args["path"])
        elif name == "write_file":
            return self.write_file(args["path"], args["content"])

    def run_python(self, code: str) -> str:
        result = subprocess.run(
            ["python", "-c", code],
            capture_output=True,
            text=True,
            timeout=30
        )
        return f"stdout: {result.stdout}\nstderr: {result.stderr}"

SYSTEM_PROMPT = """
你是一个专业的代码智能体。你的目标是帮助用户完成编程任务。

工作流程：
1. 理解用户需求，必要时提问澄清
2. 制定解决计划
3. 编写代码
4. 执行代码验证
5. 如果失败，分析错误并修复
6. 返回最终结果

原则：
- 代码必须可执行
- 优先使用标准库
- 添加必要注释
- 处理异常情况
- 不确定时明确说明

工具使用：
- execute_python: 执行Python代码
- read_file: 读取文件
- write_file: 写入文件
"""

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "execute_python",
            "description": "执行Python代码并返回输出",
            "parameters": {
                "type": "object",
                "properties": {
                    "code": {"type": "string", "description": "Python代码"}
                },
                "required": ["code"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "读取文件内容",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string"}
                },
                "required": ["path"]
            }
        }
    }
]