agent-loop

简单来说，agent-loop就是一个 while True 循环，循环直接检查响应里的内容块：包含tool_use则调用工具，不包含则退出循环。

flowchart LR
  accTitle: Agent Loop 消息闭环
  accDescr: 用户消息进入模型；模型若请求工具，Harness 执行工具并回传结果，然后再次调用模型；模型不再请求工具时输出最终回答。

  user["用户消息"] --> model["调用模型"]
  model --> decision{"是否包含 tool_calls"}
  decision -->|是| execute["执行工具"]
  execute --> result["追加 tool 消息"]
  result --> model
  decision -->|否| final(["输出最终回答"])

  class execute,result added
  class decision attention

工作原理

将这个过程翻译成代码。分步来看：

把用户的问题作为第一条消息。
将消息和工具定义一起发给 LLM。
追加模型回答，检查它是否调了工具。没调则结束。
执行模型要求的工具，收集结果。
把工具结果作为新消息追加，回到第 2 步。

把这个过程组装为一个完整函数： async def agent_loop(messages: list) -> None: while True: response = await client.chat.completions.create( model=MODEL, messages=messages, tools=TOOLS, tool_choice=“auto” ) message = response.choices[0].message messages.append(message.model_dump(exclude_none=True))


    tool_calls = message.tool_calls or []
    if not tool_calls:  # 模型没有调用工具，本轮结束
        return

    for call in tool_calls:
        command = json.loads(call.function.arguments)["command"]
        print(f"\033[33m$ {command}\033[0m")
        output = run_bash(command)
        print(output[:200])
        messages.append(
            {"role": "tool", "tool_call_id": call.id, "content": output}
        )

三十多行，这就是最小可运行的 agent harness 内核。它为模型提供持续行动的最小运行框架：模型负责决策（要不要调工具、调哪个），harness 负责执行（调用工具，把结果作为新消息追加）

最小实现

写一个OpenAI function 格式的工具，让ai能够执行bash命令，实现一个最小的agent-loop：

代码会执行模型生成的 shell 命令。建议在一个临时测试目录中运行，避免影响你的项目文件。


"""s01: 最小 Agent Loop。

一个 Agent 的本质就是一个 while 循环：把消息发给模型，模型要么给出最终答复，
要么请求调用工具；调用工具后把结果塞回消息，继续循环，直到模型不再调用工具。

运行：
    在 .env 配置 MODEL_API_KEY、BASE_URL，可选 MODEL
    uv run s01_agent_loop.py
"""

from __future__ import annotations

import asyncio
import json
import os
import subprocess

from dotenv import load_dotenv
from openai import AsyncOpenAI

load_dotenv()

client = AsyncOpenAI(api_key=os.getenv("MODEL_API_KEY"), base_url=os.getenv("BASE_URL"))
MODEL = os.getenv("MODEL", "doubao-seed-evolving")
SYSTEM = f"You are a coding agent at {os.getcwd()}. Use bash to solve tasks. Act, don't explain."

# 工具定义：bash（OpenAI function 格式）
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "bash",
            "description": "Run a shell command.",
            "parameters": {
                "type": "object",
                "properties": {"command": {"type": "string"}},
                "required": ["command"],
            },
        },
    }
]

# 工具执行：黑名单只拦最明显的破坏性命令
def run_bash(command: str) -> str:
    dangerous = ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"]
    if any(d in command for d in dangerous):
        return "Error: Dangerous command blocked"
    try:
        r = subprocess.run(
            command,
            shell=True,
            cwd=os.getcwd(),
            capture_output=True,
            text=True,
            errors="replace",
            timeout=120,
        )
        out = (r.stdout + r.stderr).strip()
        return out[:50000] if out else "(no output)"
    except subprocess.TimeoutExpired:
        return "Error: Timeout (120s)"
    except (FileNotFoundError, OSError) as e:
        return f"Error: {e}"

# 循环调用工具，直到模型不再请求工具
async def agent_loop(messages: list) -> None:
    while True:
        response = await client.chat.completions.create(
            model=MODEL, messages=messages, tools=TOOLS, tool_choice="auto"
        )
        message = response.choices[0].message
        messages.append(message.model_dump(exclude_none=True))

        tool_calls = message.tool_calls or []
        if not tool_calls:  # 模型没有调用工具，本轮结束
            return

        for call in tool_calls:
            command = json.loads(call.function.arguments)["command"]
            print(f"\033[33m$ {command}\033[0m")
            output = run_bash(command)
            print(output[:200])
            messages.append(
                {"role": "tool", "tool_call_id": call.id, "content": output}
            )

# 入口
async def main() -> None:
    print("s01: Agent Loop")
    print("输入问题回车发送，输入 q 退出。\n")

    messages = [{"role": "system", "content": SYSTEM}]
    while True:
        try:
            query = input("s01 >> ")
        except (EOFError, KeyboardInterrupt):
            break
        if query.strip().lower() in ("q", "exit", ""):
            break
        messages.append({"role": "user", "content": query})
        await agent_loop(messages)
        print(messages[-1].get("content") or "")
        print()

if __name__ == "__main__":
    asyncio.run(main())