从零构建 CLI Coding Agent：ReAct 驱动的实战实现

# agent/tools.py
import subprocess
from pathlib import Path
from typing import Any

WORKSPACE = Path(".").resolve()


def _safe_path(path: str) -> Path:
    """
    将相对路径解析为绝对路径，并验证路径在工作区范围内。
    path: 用户或 LLM 提供的文件路径字符串
    """
    target = (WORKSPACE / path).resolve()
    if not str(target).startswith(str(WORKSPACE)):
        raise ValueError(f"路径越界：{path}")
    return target


def read_file(path: str) -> dict[str, Any]:
    """
    读取文件内容，同时返回行数信息。
    path: 相对于工作区根目录的文件路径
    """
    try:
        p = _safe_path(path)
        content = p.read_text(encoding="utf-8")
        lines = content.splitlines()
        return {
            "success": True,
            "content": content,
            "lines": len(lines),
            "path": str(p.relative_to(WORKSPACE)),
        }
    except FileNotFoundError:
        return {"success": False, "error": f"文件不存在：{path}"}
    except Exception as e:
        return {"success": False, "error": str(e)}


def write_file(path: str, content: str) -> dict[str, Any]:
    """
    将内容写入文件，自动创建不存在的父目录。
    path: 目标文件路径（相对路径）
    content: 要写入的完整文件内容
    """
    try:
        p = _safe_path(path)
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(content, encoding="utf-8")
        return {
            "success": True,
            "path": str(p.relative_to(WORKSPACE)),
            "bytes_written": len(content.encode()),
        }
    except Exception as e:
        return {"success": False, "error": str(e)}


def run_command(command: str, timeout: int = 30) -> dict[str, Any]:
    """
    在工作区目录执行 Shell 命令，返回完整的执行结果。
    command: 要执行的 Shell 命令字符串
    timeout: 命令超时秒数，超时后强制终止
    """
    if is_dangerous(command):
        return {"success": False, "error": f"拒绝执行危险命令：{command}"}
    try:
        result = subprocess.run(
            command,
            shell=True,
            capture_output=True,
            text=True,
            timeout=timeout,
            cwd=str(WORKSPACE),
        )
        return {
            "success": result.returncode == 0,
            "stdout": result.stdout,
            "stderr": result.stderr,
            "returncode": result.returncode,
        }
    except subprocess.TimeoutExpired:
        return {"success": False, "error": f"命令超时（{timeout}s）：{command}"}
    except Exception as e:
        return {"success": False, "error": str(e)}


def list_dir(path: str = ".") -> dict[str, Any]:
    """
    列出目录下的文件和子目录（跳过隐藏文件）。
    path: 目录路径，默认为工作区根目录
    """
    try:
        p = _safe_path(path)
        items = []
        for item in sorted(p.iterdir()):
            if item.name.startswith("."):
                continue
            entry = {
                "name": item.name,
                "type": "dir" if item.is_dir() else "file",
                "path": str(item.relative_to(WORKSPACE)),
            }
            if item.is_file():
                entry["size"] = item.stat().st_size
            items.append(entry)
        return {"success": True, "path": str(p.relative_to(WORKSPACE)), "items": items}
    except Exception as e:
        return {"success": False, "error": str(e)}


DANGEROUS_PATTERNS = ["rm -rf", "sudo", "curl | bash", "wget | sh", "> /dev/"]


def is_dangerous(command: str) -> bool:
    """
    检测命令是否包含危险操作模式。
    command: 待检测的命令字符串
    """
    return any(p in command for p in DANGEROUS_PATTERNS)


# 工具注册表：name → callable
TOOL_REGISTRY = {
    "read_file": read_file,
    "write_file": write_file,
    "run_command": run_command,
    "list_dir": list_dir,
}

# OpenAI function calling 格式：每个工具包裹在 {"type": "function", "function": {...}} 中
TOOL_SCHEMAS = [
    {
        "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 命令。返回 stdout、stderr 和退出码。",
            "parameters": {
                "type": "object",
                "properties": {
                    "command": {"type": "string", "description": "要执行的 Shell 命令"},
                    "timeout": {"type": "integer", "description": "超时秒数，默认 30"},
                },
                "required": ["command"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "list_dir",
            "description": "列出目录下的文件和子目录。",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "目录路径，默认为工作区根目录"}
                },
            },
        },
    },
]

# agent/prompts.py

SYSTEM_PROMPT = """你是一个运行在命令行的 Coding Agent，帮助用户完成编程任务。

## 能力范围
你可以读取、创建、修改文件，执行 Shell 命令，并基于执行结果继续推理。

## 工作原则

**先探索，再行动。** 在修改任何文件之前，先用 list_dir 和 read_file 了解项目结构。
不要假设文件的内容或结构。

**小步执行，验证推进。** 每次只做一件事。写完代码就运行测试。执行命令后检查输出。
不要一次写完所有代码再测试。

**错误是信息。** 命令返回非零退出码、文件读取失败，都是你需要处理的信息，而不是需要隐藏的问题。
把错误信息传递给用户。

**保守修改。** 只修改任务要求修改的文件。不要"顺便"优化无关代码。

## 工具使用规范
- 每次只调用一个工具，等待结果后再决定下一步
- run_command 执行的命令必须是幂等的，或者你已经理解它的副作用
- write_file 会完全覆盖目标文件，写入前确认内容正确
- 路径使用相对路径，相对于工作区根目录

## 完成标准
当任务完成时，向用户简洁地说明：做了什么、验证结果是什么、有什么需要用户注意的地方。
"""

# agent/core.py
import json
import openai
from typing import Iterator
from .tools import TOOL_REGISTRY, TOOL_SCHEMAS
from .prompts import SYSTEM_PROMPT

# 兼容任意 OpenAI 格式的接口：修改 base_url 即可切换到 DeepSeek、Qwen 等模型
client = openai.OpenAI(
    base_url="https://api.openai.com/v1",  # 替换为你的 API 地址
    api_key="your-api-key",                # 或从环境变量读取
)
MODEL = "gpt-4o"
MAX_ITERATIONS = 30


def trim_history(history: list[dict], max_turns: int = 20) -> list[dict]:
    """
    截断过长的对话历史，只保留最近 max_turns 轮。
    history: 当前完整对话历史
    max_turns: 保留的最大轮次数
    """
    if len(history) <= max_turns * 2:
        return history
    trimmed = history[-(max_turns * 2):]
    while trimmed and trimmed[0]["role"] != "user":
        trimmed = trimmed[1:]
    return trimmed


def run_agent(
    user_message: str,
    conversation_history: list[dict],
) -> Iterator[str]:
    """
    运行一轮 Agent 交互，以 generator 形式流式 yield 状态和输出。
    user_message: 用户本轮输入的任务描述
    conversation_history: 跨轮次共享的对话历史列表（会被原地修改）
    """
    conversation_history.append({"role": "user", "content": user_message})

    for iteration in range(MAX_ITERATIONS):
        yield f"\n[思考中... 第 {iteration + 1} 轮]\n"

        # system prompt 以第一条消息的形式传入，符合 OpenAI 格式
        messages = [{"role": "system", "content": SYSTEM_PROMPT}] + trim_history(conversation_history)

        response = client.chat.completions.create(
            model=MODEL,
            max_tokens=8096,
            tools=TOOL_SCHEMAS,
            messages=messages,
        )

        choice = response.choices[0]
        finish_reason = choice.finish_reason
        message = choice.message

        # 将 assistant 消息（含可能的 tool_calls）追加进历史
        assistant_msg: dict = {"role": "assistant", "content": message.content}
        if message.tool_calls:
            assistant_msg["tool_calls"] = [
                {
                    "id": tc.id,
                    "type": "function",
                    "function": {"name": tc.function.name, "arguments": tc.function.arguments},
                }
                for tc in message.tool_calls
            ]
        conversation_history.append(assistant_msg)

        if message.content:
            yield message.content

        if finish_reason == "stop":
            yield "\n[✓ 任务完成]\n"
            break

        if finish_reason == "tool_calls":
            for tc in message.tool_calls:
                tool_name = tc.function.name
                tool_input = json.loads(tc.function.arguments)
                tool_call_id = tc.id

                yield f"\n[调用工具] {tool_name}({json.dumps(tool_input, ensure_ascii=False)})\n"

                if tool_name in TOOL_REGISTRY:
                    try:
                        result = TOOL_REGISTRY[tool_name](**tool_input)
                    except Exception as e:
                        result = {"success": False, "error": f"工具执行异常：{str(e)}"}
                else:
                    result = {"success": False, "error": f"未知工具：{tool_name}"}

                result_str = json.dumps(result, ensure_ascii=False, indent=2)
                yield f"[工具结果] {result_str[:200]}{'...' if len(result_str) > 200 else ''}\n"

                # OpenAI 格式：每个工具结果是一条独立的 tool 消息
                conversation_history.append({
                    "role": "tool",
                    "tool_call_id": tool_call_id,
                    "content": result_str,
                })

        else:
            yield f"\n[停止原因：{finish_reason}]\n"
            break

    else:
        yield f"\n[达到最大迭代次数 {MAX_ITERATIONS}，任务中止]\n"

# cli/main.py
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent))

from agent.core import run_agent
from agent.tools import WORKSPACE


def print_banner():
    """打印启动横幅，显示工作区路径。"""
    print("\n╔════════════════════════════════╗")
    print("║     Coding Agent (CLI)         ║")
    print(f"║  工作区：{str(WORKSPACE)[:20]:<20} ║")
    print("╚════════════════════════════════╝\n")
    print("输入你的编程任务，按 Enter 执行。")
    print("输入 /exit 退出，/clear 清空对话历史。\n")


def main():
    """CLI 主入口，负责读取用户输入并调用 Agent。"""
    print_banner()
    conversation_history = []

    while True:
        try:
            user_input = input("\n> ").strip()
        except (KeyboardInterrupt, EOFError):
            print("\n再见！")
            break

        if not user_input:
            continue

        if user_input == "/exit":
            print("再见！")
            break

        if user_input == "/clear":
            conversation_history.clear()
            print("[对话历史已清空]")
            continue

        if user_input == "/history":
            print(f"[当前对话轮次：{len(conversation_history)}]")
            continue

        print("\n" + "─" * 50)

        try:
            for chunk in run_agent(
                user_message=user_input,
                conversation_history=conversation_history,
            ):
                print(chunk, end="", flush=True)
        except KeyboardInterrupt:
            print("\n\n[用户中断]")
        except Exception as e:
            print(f"\n[错误] {e}")

        print("\n" + "─" * 50)


if __name__ == "__main__":
    main()

stream = client.chat.completions.create(
    model=MODEL,
    max_tokens=8096,
    tools=TOOL_SCHEMAS,
    messages=messages,
    stream=True,
)
for chunk in stream:
    delta = chunk.choices[0].delta
    if delta.content:
        print(delta.content, end="", flush=True)

pip install openai

export OPENAI_API_KEY=sk-...

python cli/main.py

# DeepSeek
client = openai.OpenAI(base_url="https://api.deepseek.com/v1", api_key="sk-...")

# 本地 Ollama
client = openai.OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")