200行代码实现Claude Code青春版

- 你提出一个目标
- 系统读取一些代码或环境信息
- 模型输出一个"下一步行动"
- 如果这个行动需要访问外部世界，就调用工具
- 工具执行完后，把结果再交还给模型
- 重复以上过程，中途也可以人工干预

while not done:
    observation = collect_context()
    action = llm(observation)
    if action is tool_call:
        result = execute(action)
        append_to_context(result)
    else:
        done = True

def execute_bash(command: str) -> str:
def read_file(path: str) -> str:
def write_file(path: str, content: str) -> str:
def list_files(path: str = ".") -> str:

TOOLS_SCHEMA = [
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "...",
            "parameters": {...}
        }
    }
]

system_prompt = """
你是一个智能编程助手。
工作原则：
1. 修改前先读文件
2. 修改后尽量运行测试
"""

while True:  # ← 内层循环：Agent 自主执行
    # Think: LLM 推理
    response = client.chat.completions.create(
        messages=messages,  # ← 完整的上下文历史
        tools=TOOLS_SCHEMA,  # ← 可用工具列表
    )

    # Act: 执行工具
    if message.tool_calls:
        for tool_call in message.tool_calls:
            result = AVAILABLE_FUNCTIONS[func_name](**func_args)

            # Observe: 将结果反馈给 Agent
            messages.append({
                "role": "tool",
                "content": result,  # ← 执行结果注入 context
            })
        continue  # ← 继续循环

    else:
        break  # ← 没有工具调用，任务完成

import os
import json
import subprocess
from openai import OpenAI

# =============================================================================
# 1. 定义工具 (Define Tools)
# =============================================================================

def execute_bash(command: str) -> str:
    """执行 Bash 命令"""
    print(f"  > [执行命令] \033[93m{command}\033[0m")
    print(f"  > [执行中] {command}")
    result = subprocess.run(
        command, shell=True, capture_output=True, text=True, timeout=30
    )
    output = result.stdout + result.stderr
    if output:
        print(f"  > [输出]\n{output.rstrip()}")
    # 截断过长的输出
    if len(output) > 2000:
        output = output[:2000] + "\n... (输出过长已截断)"
    return output.strip() or "(无输出)"
def read_file(path: str) -> str:
    """读取文件内容"""
    print(f"  > [读取文件] {path}")
    if not os.path.exists(path):
        return f"文件不存在: {path}"
    with open(path, "r", encoding="utf-8") as f:
        content = f.read()
        # 给代码加上行号，方便 Agent 指定修改位置
        lines = content.splitlines()
        numbered_lines = [f"{i+1:4d} | {line}" for i, line in enumerate(lines)]
        return "\n".join(numbered_lines)
def write_file(path: str, content: str) -> str:
    """写入文件内容"""
    print(f"  > [写入文件] {path}")
    os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        f.write(content)
    return f"成功写入 {len(content)} 字节到 {path}"
def list_files(path: str = ".") -> str:
    """列出当前目录下的文件结构"""
    print(f"  > [列出文件] {path}")
    # 简单的递归列出文件
    file_list = []
    for root, dirs, files in os.walk(path):
        for file in files:
            if not file.startswith('.'):
                rel_path = os.path.relpath(os.path.join(root, file), path)
                file_list.append(rel_path)

    result = "\n".join(file_list[:50]) # 限制返回数量
    print(f"  > 包含文件:\n{result}")
    return result
# =============================================================================
# 2. 定义工具描述 (Define Tool Schemas)
# =============================================================================
TOOLS_SCHEMA = [
    {
        "type": "function",
        "function": {
            "name": "execute_bash",
            "description": "执行 shell 命令。用于运行测试、安装依赖或文件操作。",
            "parameters": {
                "type": "object",
                "properties": {
                    "command": {"type": "string", "description": "要执行的 bash 命令"}
                },
                "required": ["command"],
            },
        },
    },
    {
        "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": "list_files",
            "description": "列出当前项目的文件结构。",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "目录路径，默认为 '.'"}
                },
                "required": [],
            },
        },
    },
]
AVAILABLE_FUNCTIONS = {
    "execute_bash": execute_bash,
    "read_file": read_file,
    "write_file": write_file,
    "list_files": list_files,
}
# =============================================================================
# 3. Agent 主循环 (The Agent Loop)
# =============================================================================
def run_toy_claude():
    client = OpenAI()

    # 系统提示词：设定人设
    system_prompt = """
    你是一个智能编程助手。
    你的目标是帮助用户编写、调试和理解代码。

    工作原则：
    1. 在修改代码前，先使用 list_files 查看项目结构，或使用 read_file 读取相关文件。
    2. 每次修改文件后，尽量运行代码或测试来验证修改（使用 execute_bash）。
    3. 保持回答简洁，专注于解决问题。
    """
    messages = [{"role": "system", "content": system_prompt}]
    print("🤖 Toy Claude Code 已启动 (输入 'exit' 退出)")
    print("--------------------------------------------")
    # 外层循环：处理用户输入 (REPL)
    while True:
        try:
            user_input = input("\n👤 你: ")
            if user_input.lower() in ['exit', 'quit']:
                print("再见！")
                break

            messages.append({"role": "user", "content": user_input})
            # 内层循环：Agent 自主执行 (Think -> Act -> Observe)
            # 只要 Agent 还在调用工具，就一直在这个循环里
            while True:
                response = client.chat.completions.create(
                    model="gpt-5-codex",
                    messages=messages,
                    tools=TOOLS_SCHEMA,
                )

                message = response.choices[0].message
                messages.append(message)
                if message.tool_calls:
                    print(f"\n🤖 思考: {message.content or '(准备调用工具...)'}")

                    for tool_call in message.tool_calls:
                        func_name = tool_call.function.name
                        func_args = json.loads(tool_call.function.arguments)

                        if func_name in AVAILABLE_FUNCTIONS:
                            # 执行工具
                            result = AVAILABLE_FUNCTIONS[func_name](**func_args)
                        else:
                            result = f"错误: 未知工具 {func_name}"
                        # 将结果反馈给 Agent
                        messages.append({
                            "role": "tool",
                            "tool_call_id": tool_call.id,
                            "name": func_name,
                            "content": result,
                        })
                else:
                    # 没有工具调用，说明是最终回复
                    print(f"\n🤖 回复:\n{message.content}")
                    break

        except KeyboardInterrupt:
            print("\n操作已取消。")
            break
        except Exception as e:
            print(f"\n发生错误: {e}")
            break

if __name__ == "__main__":
    run_toy_claude()

def add(a, b):
    return a - b  # 错误：应该是加法

print(f"1 + 1 = {add(1, 1)}")