从代码补全到任务执行：Dropbox 如何重构 AI 时代的工程生产力

pip install openai subprocess-utils

import os
import subprocess
import json
from openai import OpenAI

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

# --- 工具定义：agent 可调用的操作 ---
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "读取指定路径的文件内容",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "文件路径"}
                },
                "required": ["path"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "write_file",
            "description": "将内容写入指定路径的文件",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "文件路径"},
                    "content": {"type": "string", "description": "写入内容"}
                },
                "required": ["path", "content"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "run_command",
            "description": "在 shell 中执行命令并返回输出",
            "parameters": {
                "type": "object",
                "properties": {
                    "command": {"type": "string", "description": "shell 命令"}
                },
                "required": ["command"]
            }
        }
    }
]

# --- 工具执行器 ---
def execute_tool(name: str, args: dict) -> str:
    """执行工具调用，返回结果字符串。"""
    if name == "read_file":
        with open(args["path"], "r") as f:
            return f.read()
    elif name == "write_file":
        with open(args["path"], "w") as f:
            f.write(args["content"])
        return f"已写入 {args['path']}"
    elif name == "run_command":
        result = subprocess.run(
            args["command"], shell=True, capture_output=True, text=True, timeout=60
        )
        return json.dumps({
            "stdout": result.stdout,
            "stderr": result.stderr,
            "exit_code": result.returncode
        })
    else:
        return f"未知工具: {name}"

# --- Agentic Loop：规划 → 执行 → 验证 ---
def run_agent(task: str, max_iterations: int = 10) -> str:
    """
    接收一个 scoped task，自主执行直到完成或达到迭代上限。
    """
    messages = [{"role": "system", "content": (
        "你是一个工程 agent。你通过调用工具来完成任务。"
        "每一步都要记录操作。任务完成后，用 run_command 运行 "
        "ruff check 和 pytest 来验证结果，然后汇报最终状态。"
    )}, {"role": "user", "content": task}]

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

        msg = response.choices[0].message

        # 如果模型不再调用工具，说明它认为任务已完成
        if not msg.tool_calls:
            return msg.content or "Agent 未返回最终结论"

        # 执行所有工具调用，把结果喂回模型
        messages.append(msg)
        for tool_call in msg.tool_calls:
            args = json.loads(tool_call.function.arguments)
            result = execute_tool(tool_call.function.name, args)
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })

    return "达到迭代上限，任务可能未完成"

# --- 运行示例 ---
if __name__ == "__main__":
    # scoped task：输入明确、范围可控、成功标准可验证
    task_description = (
        "修复 src/calculator.py 中的 lint 错误。"
        "先用 run_command 运行 'ruff check src/calculator.py' 查看问题，"
        "然后读取文件、修复、写回，最后再跑一次 ruff check 和 "
        "'pytest tests/test_calculator.py' 验证。"
    )
    result = run_agent(task_description)
    print("=== Agent 最终汇报 ===")
    print(result)