DevNull 知识库

Appearance

Agent基本结构 ​

Agent 这个词通常被用来描述一整套系统，但放到 Python 脚本里看，核心就是一个会反复跑的循环：模型看当前问题和已有结果，判断要不要调用工具；Python 执行工具，把结果放回输入里；模型继续判断，直到给出回答。

函数调用那篇只跑了一次工具：模型要查联系人，Python 查完以后交回去。Agent 多出来的地方，是它可能不止查一次。比如一个问题里既要看待办清单，又要展开某一条详情，模型就会先调用 list_todos，拿到列表后再决定要不要调用 get_todo。

这个循环看着简单，但里面有几个容易糊在一起的东西：历史记录放在哪里，工具怎么路由，模型什么时候算结束，工具一直被调用怎么办。这些不处理清楚，脚本很容易变成“模型想调什么就调什么，调到哪算哪”。

一、循环 ​

普通模型调用是一次请求、一次回答。RAG 脚本也是固定流程：先检索，再生成。Agent 的味道不太一样，它不是在代码里写死每一步，而是让模型根据当前信息决定下一步动作。

比如问题是：

text
今天还有哪些没完成？把优先级最高的一项展开说明。

Python 手里有两个工具：

  • list_todos：返回所有待办的标题、状态和优先级。
  • get_todo：按 todo_id 返回某条待办的详细说明。

模型第一次看到问题时，不知道有哪些待办，于是会请求 list_todos。Python 返回列表后，模型看到 todo-002 优先级最高，可能继续请求 get_todo(todo_id="todo-002")。拿到详情以后，模型再组织最终回答。

我一开始以为 Agent 就是“多给模型几个工具”，后来才发现真正绕的地方在循环本身。工具只是入口，关键是每一轮模型输出什么、Python 执行什么、结果怎么放回去、什么时候停下来。

二、待办数据 ​

先准备一个很小的待办文件。这个例子不碰日志、不碰告警，只看 Agent 的基本形状。

新增文件：todos.json

json
[
  {
    "id": "todo-001",
    "title": "买菜",
    "status": "done",
    "priority": "low",
    "detail": "下班路上买鸡蛋、青菜和牛奶。"
  },
  {
    "id": "todo-002",
    "title": "写周报",
    "status": "todo",
    "priority": "high",
    "detail": "整理本周完成的接口联调、遗留问题和下周计划。"
  },
  {
    "id": "todo-003",
    "title": "回复邮件",
    "status": "todo",
    "priority": "medium",
    "detail": "回复项目会议纪要，确认接口字段和上线时间。"
  }
]

list_todos 不需要返回完整详情，只返回能判断下一步的信息。get_todo 再按 id 查详情。这样拆开以后，模型第一轮不会被一大堆细节淹住。

三、工具函数 ​

先写普通 Python 函数，暂时不接模型。

新增文件：scripts/todo_tools.py

python
#!/usr/bin/env python3
"""待办查询工具函数。"""

import json
from pathlib import Path


TODO_PATH = Path("todos.json")


def load_todos():
    """读取本地待办文件。"""
    return json.loads(TODO_PATH.read_text(encoding="utf-8"))


def list_todos():
    """返回待办摘要，先不返回 detail，避免列表结果太长。"""
    return {
        "ok": True,
        "items": [
            {
                "id": item["id"],
                "title": item["title"],
                "status": item["status"],
                "priority": item["priority"],
            }
            for item in load_todos()
        ],
    }


def get_todo(todo_id):
    """按 id 返回单条待办详情。"""
    for item in load_todos():
        if item["id"] == todo_id:
            return {"ok": True, "item": item}

    return {"ok": False, "error": f"待办不存在: {todo_id}"}


if __name__ == "__main__":
    print(list_todos())
    print(get_todo("todo-002"))

运行：

bash
uv run python scripts/todo_tools.py

输出会比较长，大概能看到两部分：

text
{'ok': True, 'items': [{'id': 'todo-001', 'title': '买菜', ...}]}
{'ok': True, 'item': {'id': 'todo-002', 'title': '写周报', ...}}

工具函数能单独跑通，后面接模型时就少一层不确定。模型传错参数是一类问题，Python 函数自己读不到文件又是另一类问题，分开看会清楚很多。

四、工具定义 ​

工具定义是写给模型看的。Python 有 list_todos() 和 get_todo()，模型并不会自动知道这两个函数存在，需要把函数名、用途和参数结构告诉它。

python
TOOLS = [
    {
        "type": "function",
        "name": "list_todos",
        "description": "查询所有待办的标题、状态和优先级。",
        "parameters": {
            "type": "object",
            "properties": {},
            "required": [],
            "additionalProperties": False,
        },
        "strict": True,
    },
    {
        "type": "function",
        "name": "get_todo",
        "description": "按 todo_id 查询一条待办的完整详情。",
        "parameters": {
            "type": "object",
            "properties": {
                "todo_id": {
                    "type": "string",
                    "description": "待办 id，例如 todo-002",
                }
            },
            "required": ["todo_id"],
            "additionalProperties": False,
        },
        "strict": True,
    },
]

list_todos 没有参数，所以 properties 是空对象。get_todo 只有一个参数 todo_id。这里继续开 strict=True，模型返回的参数会更贴近 schema，Python 这边也更容易处理。

description 不用写得很花，关键是让模型知道什么时候该用。list_todos 是看全局列表，get_todo 是查单条详情，这两个用途分清楚，模型才不容易一上来就瞎填某个 id。

五、输入历史 ​

Agent 循环里最重要的对象是历史记录。这里用一个列表保存每一轮的输入、模型输出和工具结果：

python
input_items = [
    {
        "role": "user",
        "content": "今天还有哪些没完成？把优先级最高的一项展开说明。",
    }
]

第一次请求后，模型可能返回 function_call。这时要把模型的输出也放进 input_items，再追加工具结果。

python
input_items.extend(response.output)
input_items.append({
    "type": "function_call_output",
    "call_id": tool_call.call_id,
    "output": json.dumps(tool_result, ensure_ascii=False),
})

这块比较绕。function_call_output 不是随便一段文字，它要带上 call_id，这样模型才能知道这份结果对应刚才哪一次工具调用。少了这个关联，模型就像听到一句没头没尾的“查到了”，但不知道查的是谁。

05 里用过 previous_response_id，那种写法适合把一次工具回填快速跑通。这里为了把循环看得更清楚，手动保存 input_items：模型每一轮说了什么、Python 每次返回了什么，都在列表里。

六、工具路由 ​

模型返回工具调用以后，Python 不能直接按字符串执行函数。比如模型返回 name="delete_all"，本地根本不该有这种入口。工具路由要写成白名单。

python
def dispatch_tool(tool_call):
    """根据模型请求执行本地工具，所有工具都走白名单。"""
    try:
        arguments = json.loads(tool_call.arguments or "{}")
    except json.JSONDecodeError as exc:
        return {"ok": False, "error": f"工具参数不是合法 JSON: {exc}"}

    if tool_call.name == "list_todos":
        return list_todos()

    if tool_call.name == "get_todo":
        todo_id = arguments.get("todo_id")
        if not isinstance(todo_id, str) or not todo_id.strip():
            return {"ok": False, "error": f"todo_id 不合法: {todo_id!r}"}
        return get_todo(todo_id)

    return {"ok": False, "error": f"未知工具: {tool_call.name}"}

这里的错误也作为工具结果返回，而不是直接让脚本崩掉。比如模型把 todo_id 漏了，工具返回：

json
{"ok": false, "error": "todo_id 不合法: None"}

模型拿到这个结果后，可能会重新调用 list_todos 或直接说明资料不足。至少 Python 这边没有把一个坏参数继续传下去。

七、循环脚本 ​

完整脚本放在一个文件里，方便看清楚 Agent 循环长什么样。

新增文件：scripts/todo_agent_loop.py

python
#!/usr/bin/env python3
"""一个最小的待办 Agent 循环。"""

import argparse
import json
import os
import sys
from pathlib import Path

from openai import OpenAI


TODO_PATH = Path("todos.json")

TOOLS = [
    {
        "type": "function",
        "name": "list_todos",
        "description": "查询所有待办的标题、状态和优先级。",
        "parameters": {
            "type": "object",
            "properties": {},
            "required": [],
            "additionalProperties": False,
        },
        "strict": True,
    },
    {
        "type": "function",
        "name": "get_todo",
        "description": "按 todo_id 查询一条待办的完整详情。",
        "parameters": {
            "type": "object",
            "properties": {
                "todo_id": {
                    "type": "string",
                    "description": "待办 id，例如 todo-002",
                }
            },
            "required": ["todo_id"],
            "additionalProperties": False,
        },
        "strict": True,
    },
]


def parse_args():
    parser = argparse.ArgumentParser(description="run todo agent loop")
    parser.add_argument("question", help="要询问待办 Agent 的问题")
    parser.add_argument("--max-steps", type=int, default=5, help="最多允许几轮工具调用")
    return parser.parse_args()


def load_todos():
    if not TODO_PATH.exists():
        return []
    return json.loads(TODO_PATH.read_text(encoding="utf-8"))


def list_todos():
    return {
        "ok": True,
        "items": [
            {
                "id": item["id"],
                "title": item["title"],
                "status": item["status"],
                "priority": item["priority"],
            }
            for item in load_todos()
        ],
    }


def get_todo(todo_id):
    for item in load_todos():
        if item["id"] == todo_id:
            return {"ok": True, "item": item}
    return {"ok": False, "error": f"待办不存在: {todo_id}"}


def dispatch_tool(tool_call):
    """只执行白名单里的工具，参数异常也包装成工具结果。"""
    try:
        arguments = json.loads(tool_call.arguments or "{}")
    except json.JSONDecodeError as exc:
        return {"ok": False, "error": f"工具参数不是合法 JSON: {exc}"}

    if tool_call.name == "list_todos":
        return list_todos()

    if tool_call.name == "get_todo":
        todo_id = arguments.get("todo_id")
        if not isinstance(todo_id, str) or not todo_id.strip():
            return {"ok": False, "error": f"todo_id 不合法: {todo_id!r}"}
        return get_todo(todo_id)

    return {"ok": False, "error": f"未知工具: {tool_call.name}"}


def run_agent(client, model, question, max_steps):
    input_items = [
        {
            "role": "user",
            "content": question,
        }
    ]
    trace = []

    for step in range(1, max_steps + 1):
        response = client.responses.create(
            model=model,
            instructions=(
                "你是一个待办清单助手。"
                "需要查询待办时使用工具。"
                "回答只能基于工具返回的数据。"
            ),
            tools=TOOLS,
            input=input_items,
            max_output_tokens=500,
        )

        # 模型本轮输出也要保留下来，下一轮才能接着刚才的工具调用往下走。
        input_items.extend(response.output)

        function_calls = [
            item for item in response.output
            if item.type == "function_call"
        ]

        if not function_calls:
            return {
                "answer": response.output_text.strip(),
                "trace": trace,
                "stopped_by": "answer",
            }

        for tool_call in function_calls:
            tool_result = dispatch_tool(tool_call)
            trace.append({
                "step": step,
                "tool": tool_call.name,
                "arguments": tool_call.arguments,
                "result": tool_result,
            })

            print(
                f"step {step} call {tool_call.name} {tool_call.arguments}",
                file=sys.stderr,
            )

            input_items.append({
                "type": "function_call_output",
                "call_id": tool_call.call_id,
                "output": json.dumps(tool_result, ensure_ascii=False),
            })

    return {
        "answer": "工具调用次数超过上限，停止执行。",
        "trace": trace,
        "stopped_by": "max_steps",
    }


def main():
    args = parse_args()
    api_key = os.getenv("OPENAI_API_KEY")
    model = os.getenv("OPENAI_MODEL")

    if not api_key:
        print("缺少环境变量: OPENAI_API_KEY", file=sys.stderr)
        return 2
    if not model:
        print("缺少环境变量: OPENAI_MODEL", file=sys.stderr)
        return 2

    client = OpenAI(api_key=api_key, timeout=60.0)
    result = run_agent(
        client=client,
        model=model,
        question=args.question,
        max_steps=args.max_steps,
    )

    output_path = Path("outputs/todo-agent-trace.json")
    output_path.parent.mkdir(exist_ok=True)
    output_path.write_text(
        json.dumps(result, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    print(result["answer"])
    print(f"trace 已保存: {output_path}", file=sys.stderr)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

运行：

bash
uv run python scripts/todo_agent_loop.py "今天还有哪些没完成？把优先级最高的一项展开说明。"

stderr 里会看到工具调用过程：

text
step 1 call list_todos {}
step 2 call get_todo {"todo_id":"todo-002"}
trace 已保存: outputs/todo-agent-trace.json

标准输出可能类似：

text
今天还有两项没完成：写周报和回复邮件。优先级最高的是“写周报”，需要整理本周完成的接口联调、遗留问题和下周计划。

这个回答经历了两轮工具调用。第一轮拿列表，第二轮拿详情。模型不是一开始就知道 todo-002 的内容，它是在看到 list_todos 的结果以后，才继续请求详情。

八、停止条件 ​

Agent 循环不能无限跑。这个脚本有两个停止条件。

第一种是模型不再返回 function_call，而是返回普通回答。代码里看到 function_calls 为空，就把 response.output_text 当成最终结果。

第二种是超过 --max-steps。默认最多 5 轮工具调用：

bash
uv run python scripts/todo_agent_loop.py "今天还有哪些没完成？" --max-steps 1

如果模型第一轮只调用了 list_todos，还没来得及整理回答，脚本就会停在：

text
工具调用次数超过上限，停止执行。

这个上限不是为了刁难模型，是为了防止它一直查同一个工具。比如工具描述写得太含糊，模型可能反复调用 list_todos，每次都拿到同一份列表。没有上限的话，脚本会一直烧请求。

还有一种细节：有些模型可能在同一轮里既返回一段说明文字，又返回 function_call。这个脚本只要看到 function_call，就继续执行工具，暂时不把那段说明当最终回答。因为那段文字经常只是“我需要先查询待办列表”这种中间话，不是最后结果。

九、中间结果 ​

Agent 出问题时，最怕只看到最后一句回答，不知道中间查了什么。脚本里用 trace 保存了每一步工具调用：

json
{
  "step": 1,
  "tool": "list_todos",
  "arguments": "{}",
  "result": {
    "ok": true,
    "items": [
      {
        "id": "todo-001",
        "title": "买菜",
        "status": "done",
        "priority": "low"
      }
    ]
  }
}

完整结果会写到：

text
outputs/todo-agent-trace.json

如果回答说“优先级最高的是回复邮件”，但 trace 里明明 todo-002 的 priority 是 high，那问题多半出在模型理解工具结果这一段。如果 trace 里压根没有 todo-002，那就回到 list_todos 或 todos.json 看数据有没有读出来。

这比单纯打印最终回答好查很多。Agent 不是只看结果，过程也要留痕；工具调用、参数、返回值都保存下来，后面才知道它是怎么走到这个答案的。

十、当前结构 ​

这篇里的文件关系很少：

text
todos.json
scripts/todo_tools.py
scripts/todo_agent_loop.py
outputs/todo-agent-trace.json

todos.json 是资料，todo_tools.py 是普通 Python 工具，todo_agent_loop.py 把模型判断和工具执行串起来。outputs/todo-agent-trace.json 保存每次运行的中间过程。

这就是一个很小的 Agent 骨架：模型负责判断下一步，Python 负责执行工具，历史记录把每一步接起来，停止条件防止循环失控。工具换成 FAQ、文档检索、接口查询、日志读取时，骨架还是这一套。