messages = [
    {"role": "system", "content": "你是一个编程助手。"},   # 一次性的行为设定
    {"role": "user",   "content": "帮我读一下 README.md"},  # 用户说的话
    {"role": "assistant", "content": "好的，我来读取文件。"}, # 模型的回复
]

# llm.py
import os
import litellm
from litellm import acompletion

MODEL = os.getenv("MODEL", "openai/gpt-4o-mini")


def check_config():
    if not os.getenv("OPENAI_API_KEY"):
        raise RuntimeError("请先设置 OPENAI_API_KEY 环境变量")
    if not litellm.supports_function_calling(model=MODEL):
        raise RuntimeError(f"{MODEL} 不支持 function calling，请换一个模型")


async def call_llm(messages, tools=None):
    response = await acompletion(
        model=MODEL,
        messages=messages,
        tools=tools,
        tool_choice="auto",
    )
    return response.choices[0].message

import asyncio
from llm import call_llm, check_config


async def main():
    check_config()
    messages = [{"role": "system", "content": "你是一个编程助手。"}]

    while True:
        user = input("你: ").strip()
        if user in {"quit", "exit", "q"}:
            break

        messages.append({"role": "user", "content": user})
        message = await call_llm(messages)
        messages.append(message.model_dump(exclude_none=True))
        print(f"助手: {message.content}\n")


asyncio.run(main())

# tools.py
import asyncio
import json
import os
import sys
from pathlib import Path

WORKSPACE = Path(os.getenv("AGENT_WORKSPACE", "workspace")).resolve()
MAX_FILE_CHARS = 20_000
MAX_FILE_BYTES = 256 * 1024
MAX_OUTPUT_CHARS = 12_000
SKIP_DIR_NAMES = {".git", ".venv", "venv", "node_modules", "__pycache__", ".pytest_cache"}
TOOLS = {}


def workspace_path(path: str) -> Path:
    requested = Path(path)
    if requested.is_absolute():
        raise ValueError("只允许 workspace 内的相对路径")

    target = (WORKSPACE / requested).resolve()
    if not target.is_relative_to(WORKSPACE):
        raise ValueError("路径不能离开 workspace")
    return target


def register(name, description, parameters):
    def decorator(func):
        TOOLS[name] = (
            {
                "type": "function",
                "function": {
                    "name": name,
                    "description": description,
                    "parameters": parameters,
                },
            },
            func,
        )
        return func
    return decorator


@register("list_files", "列出 workspace 中某个目录的文件", {
    "type": "object",
    "properties": {"path": {"type": "string", "description": "相对目录，默认 ."}},
    "required": [],
    "additionalProperties": False,
})
async def list_files(path="."):
    directory = workspace_path(path)
    if not directory.is_dir():
        return f"不是目录: {path}"
    items = sorted(
        p.relative_to(WORKSPACE).as_posix()
        for p in directory.iterdir()
        if p.name not in SKIP_DIR_NAMES
    )
    return "\n".join(items[:100]) or "目录为空"


@register("read_file", "读取 workspace 中的 UTF-8 文本文件", {
    "type": "object",
    "properties": {"path": {"type": "string", "description": "相对文件路径"}},
    "required": ["path"],
    "additionalProperties": False,
})
async def read_file(path):
    file_path = workspace_path(path)
    if not file_path.is_file():
        return f"文件不存在: {path}"
    if file_path.stat().st_size > MAX_FILE_BYTES:
        return f"文件过大，拒绝完整读取: {path}（上限 {MAX_FILE_BYTES} bytes）"
    with file_path.open("rb") as file:
        if b"\x00" in file.read(1024):
            return f"疑似二进制文件，拒绝读取: {path}"
    text = file_path.read_text(encoding="utf-8")
    if len(text) > MAX_FILE_CHARS:
        return text[:MAX_FILE_CHARS] + "\n[内容已截断]"
    return text


@register("write_file", "在 workspace 中创建或覆盖一个 UTF-8 文本文件", {
    "type": "object",
    "properties": {
        "path": {"type": "string", "description": "相对文件路径"},
        "content": {"type": "string", "description": "要写入的文本"},
    },
    "required": ["path", "content"],
    "additionalProperties": False,
})
async def write_file(path, content):
    file_path = workspace_path(path)
    file_path.parent.mkdir(parents=True, exist_ok=True)
    file_path.write_text(content, encoding="utf-8")
    return f"已写入: {file_path.relative_to(WORKSPACE)}"


@register("run_tests", "在 workspace 中运行固定的 pytest 测试命令", {
    "type": "object",
    "properties": {},
    "required": [],
    "additionalProperties": False,
})
async def run_tests():
    process = await asyncio.create_subprocess_exec(
        sys.executable, "-m", "pytest", "-q",
        cwd=WORKSPACE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.STDOUT,
    )
    try:
        output, _ = await asyncio.wait_for(process.communicate(), timeout=30)
    except asyncio.TimeoutError:
        process.kill()
        await process.communicate()
        return "测试超过 30 秒，已终止"

    text = output.decode("utf-8", errors="replace")
    return f"退出码: {process.returncode}\n{text[:MAX_OUTPUT_CHARS]}"


def tool_schemas():
    return [schema for schema, _ in TOOLS.values()]


async def execute_tool(name, raw_arguments):
    if name not in TOOLS:
        return f"未知工具: {name}"

    try:
        arguments = json.loads(raw_arguments or "{}")
        if not isinstance(arguments, dict):
            return "工具参数必须是 JSON 对象"
    except json.JSONDecodeError as error:
        return f"工具参数不是合法 JSON: {error}"

    _, function = TOOLS[name]
    try:
        return str(await function(**arguments))
    except Exception as error:
        return f"工具执行失败: {type(error).__name__}: {error}"

# agent.py
from llm import call_llm
from tools import execute_tool, tool_schemas

MAX_TURNS = 10


async def run_agent(messages, user_input):
    messages.append({"role": "user", "content": user_input})

    for _ in range(MAX_TURNS):
        message = await call_llm(messages, tools=tool_schemas())
        messages.append(message.model_dump(exclude_none=True))

        tool_calls = message.tool_calls or []
        if not tool_calls:
            return message.content or "模型没有返回文本"

        for tool_call in tool_calls:
            name = tool_call.function.name
            print(f"  调用工具: {name}")
            result = await execute_tool(name, tool_call.function.arguments)
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "name": name,
                "content": result,
            })

    return "本轮工具调用已达到 10 次上限，请缩小任务后再试。"

# main.py
import asyncio
from agent import run_agent
from llm import check_config
from tools import WORKSPACE


async def main():
    check_config()
    WORKSPACE.mkdir(parents=True, exist_ok=True)
    messages = [{
        "role": "system",
        "content": (
            "你是编程助手。只能使用提供的工具处理 workspace 内的文件。"
            "修改前先阅读相关文件；测试失败时如实说明结果。"
        ),
    }]
    print(f"Agent 已启动，工作目录: {WORKSPACE}，输入 quit 退出。\n")

    while True:
        user = input("你: ").strip()
        if user in {"quit", "exit", "q"}:
            break
        reply = await run_agent(messages, user)
        print(f"助手: {reply}\n")


asyncio.run(main())

[
  # 0. 系统提示，一开始就放好，之后每次请求都带着
  {"role": "system", "content": "你是编程助手。只能使用提供的工具……"},

  # 1. 用户这轮的输入
  {"role": "user", "content": "列出当前工作目录的文件，然后读取 README.md，用一句话说明项目用途。"},

  # 2. 模型第一次回复：它没说话（content 为空），而是要求调用两个工具
  {"role": "assistant", "content": None, "tool_calls": [
    {"id": "call_a1", "type": "function",
     "function": {"name": "list_files", "arguments": "{\"path\": \".\"}"}},
    {"id": "call_b2", "type": "function",
     "function": {"name": "read_file", "arguments": "{\"path\": \"README.md\"}"}}
  ]},

  # 3 & 4. 你的程序执行工具后，为每个 tool_call 各回一条结果，
  #        靠 tool_call_id 和上面的 id 对上号
  {"role": "tool", "tool_call_id": "call_a1", "name": "list_files", "content": "README.md"},
  {"role": "tool", "tool_call_id": "call_b2", "name": "read_file", "content": "# Demo project\n"},

  # 5. 带着工具结果再问一次模型，这次它不再要工具，直接给出最终文本
  {"role": "assistant", "content": "这个项目是一个演示用的 Demo project，目前只有一个 README。"}
]