从0到1设计一个简易版AgentLoop

⏱ 约 21 分钟

本文写于2026年09月06号上午十一点

零、需求

项目的 reports/ 目录里放着一份运营日报。要知道它统计了多少笔订单、对应多少营收，通常得先找到文件，打开以后再看里面的字段。

现在，希望把这个查询过程放进对话里，使用者只需要问一句：

帮我看看 reports 目录里的日报，告诉我这份日报的日期、订单数和营收。

用户关心的是结果，以及结果来自哪份数据。他不需要知道程序调用了什么函数，也不应该每次都自己打开文件、复制内容，再交给模型整理。

用户还可能基于此结果做进一步的数据分析和让AI做辅助决策，由此引入了一个新的需求：怎样把model和tool use结合，将如上交互封装为一个多轮对话（ReAct）？

随着Harness的爆火，越来越多的前后端开发工程师和算法工程师都去关注于这套交互原理。

本文将从从这个需求出发，用 Python 一步步实现一个简易的 Agent Loop。

提前说明，本文旨在由浅入深的给出一个清晰版本的Agent Loop简易架构设计，实际生成要比当前复杂得多。

我们调用一个兼容 OpenAI 接口的模型，本文使用本地搭建的 Qwen。这里只关心怎么调用它，不展开模型服务的内部实现。配套代码与启动说明见 github链接

一、项目背景与方案设计

用户想知道日期、订单数和营收。先看看这几项数据放在哪里，当前目录中有一份日报：

sample_project/
└── reports/
    └── daily-2026-09-06.json

文件内容如下：

{
  "date": "2026-09-06",
  "orders": 128,
  "revenue_yuan": 9600
}

从文件里看，答案应该包含日期 2026-09-06、订单数 128，以及以元为单位的营收 9600。程序的回答要能追溯到这些字段。

如果需求一直固定为读取这几个字段，直接写一个脚本会更简单。这里要做的是一个自然语言文件问答入口：用户描述想知道什么，程序去取得所需资料，再组织回答。日报查询是这个入口要完成的第一项需求。

但模型此时还没有看见文件内容。用户只给了存放目录，程序需要先知道目录下有哪些文件，才能读取相关内容。文件位置和文件数据，是当前缺少的两类信息。

于是，我们给模型提供两种能力：list_files 用来查目录，read_file 用来读文件。

模型可以根据已经拿到的信息请求下一步操作，Python 负责实际执行，再把结果交回去。把这个过程接续起来，直到能够回答问题，就是本文要实现的 Agent Loop。

flowchart LR
    U[用户问题] --> L[Agent Loop]
    L -->|上下文与工具声明| M[本地 Qwen]
    M -->|文本或工具调用| L
    L -->|执行| D[list_files]
    L -->|执行| R[read_file]
    D -->|目录中的文件路径| L
    R -->|文件内容| L
    L --> H[更新 history]

这张图里有四个值得分开的部分。

模型负责根据当前上下文给出下一步：可能是回答，也可能是工具调用。

循环接住这个输出，决定执行工具还是结束。

工具把结构化参数变成真正的操作。

历史保存已经发生的事，让下一次请求能接上前一次。

这里需要把调用请求和执行结果分清楚。模型说“我查看了日报”，只能算一段输出；只有程序真的读取了文件，我们才有可用于回答的数据。工具把这两件事连接起来将函数调用结果放到模型上下文中，也让整个交互过程有据可查。

还有一点也要提前说清楚：模型并不负责替我们维护这个 Python 程序里的会话。每次调用时，我们都要把当前需要的上下文交给它。这也就是后面为什么会有一个不断增长的 history。

二、小试牛刀：把这个问题直接交给模型

先从最直接的办法试起：把用户的这句提问发给本地模型，看看一次普通调用能走到哪里。

本文的示例使用 OpenAI Responses 风格接口，配套代码中的 ResponsesClient 负责 HTTP 请求。先复用这层连接，把注意力放在请求中提供了什么信息、模型又返回了什么：

from packages.pi_ai.responses import (
    ResponsesClient,
    output_text,
    response_output,
)

client = ResponsesClient(
    base_url="http://127.0.0.1:8021",
    api_key=None,
    timeout=120,
)

response = client.create({
    "model": "qwen3-0.6b",
    "input": [{
        "role": "user",
        "content": [{
            "type": "input_text",
            "text": "帮我看看 reports 目录里的日报，告诉我这份日报的日期、订单数和营收。",
        }],
    }],
    "tools": [],
    "tool_choice": "none",
    "store": False,
    "reasoning": {"effort": "none"},
    "max_output_tokens": 2048,
    "temperature": 0,
    "parallel_tool_calls": False,
    "instructions": "你还没有获得本地日报的内容。请说明回答问题还需要什么信息，不要猜测日报数据。",
})

print(output_text(response_output(response), response))

这段代码可以在配套代码目录中运行。完整的阶段入口是 step01_model_call.py。

运行第一步后，模型给出了这样的回复：

请提供报告文件的路径或具体文件名，以便我帮助您查看并提取日期、订单数和营收信息。

这是本次第一步的真实结果它还没有完成需求，而是把寻找文件这件事交回给了用户。即使用户补充了完整路径，路径也不等于文件内容，模型仍然不能自行打开电脑上的文件。

现在缺少的不是一句更长的提问，而是一条取得本地数据的通道。下一步要让模型能够请求程序查目录、读文件。

三、工具使用：让Agent能查目录、读文件

要取得日报中的数据，程序得先定位文件，再读取内容。我们把这两种操作分别提供给模型：

工具	用途	参数
list_files	列出指定目录的直接子项	相对于项目根目录的 path，根目录使用 .
read_file	读取项目中的 UTF-8 文本文件	相对于项目根目录的文件路径

一个工具可以拆成两半：一半是给模型看的说明，另一半是留在程序里的执行函数。

例如，read_file 发给模型的声明长这样。下面是 JSON 结构示意：

{
  "type": "function",
  "name": "read_file",
  "description": "读取项目中的 UTF-8 文本文件。",
  "parameters": {
    "type": "object",
    "properties": {
      "path": {
        "type": "string",
        "description": "相对于项目根目录的文件路径"
      }
    },
    "required": ["path"],
    "additionalProperties": false
  },
  "strict": true
}

name 用来标识工具，description 帮助模型判断什么时候使用它，parameters 描述参数格式。这里有意写清“相对于项目根目录”，而不是泛泛地说“一个文件路径”。工具描述不是装饰，它会直接影响模型怎么填参数。

但模型看见的只有这份说明，不会收到我们打开文件的 Python 函数。配套代码用 AgentTool 把两部分组合起来：

@dataclass(frozen=True)
class AgentTool:
    name: str
    description: str
    parameters: dict[str, Any]
    handler: ToolHandler

    def schema(self) -> dict[str, Any]:
        return {
            "type": "function",
            "name": self.name,
            "description": self.description,
            "parameters": self.parameters,
            "strict": True,
        }

    def execute(self, arguments: Mapping[str, Any]) -> ToolResult:
        try:
            return ToolResult(self.handler(arguments))
        except (MiniPiError, OSError, RuntimeError, ValueError) as error:
            return ToolResult(str(error), is_error=True)

这是tools.py中的完整类定义片段，类型和导入也在该文件中。schema() 返回能发给模型的声明；execute() 调用本地 handler，把预期的执行错误转成一个工具结果。

真正注册文件工具时，我们把同一个受限的项目目录对象绑定给它们。这样，read_file 收到的参数虽然来自模型，实际能访问的范围却由程序事先划定。

这里很容易踩一个坑：发了 JSON Schema，不等于本地可以省掉校验。 模型可能填错，接口的约束能力也可能与我们的预期不同。教学代码会在执行前再次检查：参数只能包含 path，并且它必须是非空字符串。随后，文件访问层还会检查路径是否越界、是否为符号链接、文件是否过大等。

两层检查解决的是不同问题。参数检查决定“这是不是这个工具能接受的请求”；文件检查决定“即使参数格式正确，这个文件是不是允许读取”。不能只留其中一层。

如果运行第二个示例：

python3 -m examples.step02_tool_request

运行第二步后，模型返回了一个 list_files 请求，准备查询 reports 目录。程序此时只展示请求，还没有执行它。接下来要做的，是把这个请求变成一次真实的目录访问，并把结果交回模型。

四、先手动走通一轮，看看历史里究竟多了什么

假设我们将第二章的tool里头的字段填上之后，模型返回了下面这个调用（json格式返回 tool use 字段）：

{
  "type": "function_call",
  "call_id": "call_01",
  "name": "list_files",
  "arguments": "{\"path\":\"reports\"}"
}

注意 arguments：它在响应中可能是一段 JSON 字符串，而不是已经解析好的 Python 字典。因此，执行工具之前，还需要解析参数，并检查结果是不是对象。

配套代码的 parse_function_calls() 就做这件事。如果名字或调用标识缺失，或者参数不是合法的 JSON 对象，它会明确报错，而不是猜一个值继续跑。这个解析步骤也提醒我们：模型返回的内容属于程序的外部输入，要像处理其他外部输入一样认真。

找到目录工具并执行以后，我们得到日报的文件路径。接下来怎么交给模型？不是随便追加一条“文件在这里”的用户消息，而是生成一条与原调用对应的工具结果：

{
  "type": "function_call_output",
  "call_id": "call_01",
  "output": "reports/daily-2026-09-06.json"
}

注意，这条结果只有路径，没有订单数和营收。模型还必须请求 read_file，才能知道文件里写了什么。我们不会在目录工具里顺手读取内容，否则两个工具的职责就混在一起了。

call_id 用来说明“这个结果，对应刚才那一次请求”。本例中，目录查询和文件读取各有一个 ID，不能把日报内容填到目录查询的 ID 下。即使将来同一个工具被调用多次，也仍然按每次调用的 ID 配对，而不是只看工具名称。

模型拿到reports/daily-2026-09-06.json结果之后，在下一轮对话中会使用read_file工具，然后工具再返回结果，在之后模型根据工具返回的结果再调用工具，就这样一直循环，直到模型认为资料足够不再调用工具，而是输出一段话用于回答用户问题，这个就是最简单的AgentLoop。

下面是两轮工具调用的时序图

sequenceDiagram
    participant L as Agent Loop
    participant M as 本地 Qwen
    participant D as list_files
    participant R as read_file
    L->>M: 第1轮：用户问题
    M-->>L: 调用 A：列出 reports
    L->>D: path = reports
    D-->>L: reports/daily-2026-09-06.json
    Note over L: history 追加调用 A 与结果 A
    L->>M: 第2轮：带上目录查询结果
    M-->>L: 调用 B：读取发现的文件
    L->>R: path = 目录结果中的文件路径
    R-->>L: 日报内容：日期、订单数、营收
    Note over L: history 追加调用 B 与结果 B
    L->>M: 第3轮：带上两轮工具结果
    M-->>L: 2026-09-06，128笔，9600元

图 2：两轮工具调用分别使用 list_files 和 read_file，第三次模型请求才得到最终回答。A、B 是便于说明的调用标记。

历史的增长顺序也不能颠倒：

output = response_output(response)
calls = parse_function_calls(output)

# 先保留模型原本的输出，包括它发起的调用。
history.extend(output)

for call in calls:
    tool = tools_by_name.get(call.name)
    result = (
        tool.execute(call.arguments)
        if tool is not None
        else ToolResult(f"未知工具：{call.name}", is_error=True)
    )

    # 再把每次执行结果与原调用配对。
    history.append({
        "type": "function_call_output",
        "call_id": call.call_id,
        "output": result.content,
    })

这是完整流程中的一段，变量由前面的请求和工具注册步骤建立；可运行上下文见 code/examples/step03_one_round.py。这里最值得记住的是两句：history.extend(output) 和 history.append(...)。前一句留下模型说了什么、请求了什么，后一句记录程序做了什么、结果如何。

为什么保留原始 output，不只摘出其中的工具调用？因为一个响应可能不只有一个调用项。我们不应该在循环层随意丢掉模型已经返回、协议允许回放的内容，再期待下一次请求能完整接上。

这时再调用一次模型，输入变成了“问题 + 目录查询请求 + 发现的文件路径”。模型知道了下一步可以读取哪个文件，但还不能凭文件名回答订单数和营收。

第三个示例只演示这一批工具的执行和反馈。在本例中，它收到第二个 read_file 请求后就停下来，明确说明任务尚未完成。下一节要做的，就是把这个人工停下来的地方交给循环，让第二个工具真正得到执行。

五、把重复的部分收进 Agent Loop

走到这里，其实已经没有太多神秘的东西了。

每轮都做同样几件事：准备请求，调用模型，处理输出；如果有工具调用，就执行并回填，然后开始下一轮。

flowchart TD
    A[初始化用户消息和工具表] --> B{还有可用轮次?}
    B -->|否| X[轮数耗尽：报错]
    B -->|是| C[请求模型]
    C --> D[解析并追加模型原始输出]
    D --> E{有工具调用?}
    E -->|是| F[查找并执行工具]
    F --> G[按 call_id 追加工具结果]
    G --> B
    E -->|否| H{有有效文本?}
    H -->|是| I[返回最终回答]
    H -->|否| J[空响应：报错]

图 3：第一次回到循环时，模型拿到文件路径；第二次回到循环时，它才拿到日报内容。请求或协议出错时也会明确中止。

先准备三样东西。

第一样是工具注册表。把工具名映射到工具对象，执行时就能直接查找。注册阶段还应该拒绝重名，否则后来的工具可能悄悄覆盖前一个，模型看到的说明与程序真正执行的函数也可能不一致。

第二样是历史。它一开始只有用户的问题，之后每轮增加模型输出和工具结果。

第三样是运行配置，例如模型名、系统提示词和最大轮数。这里的一轮按一次模型请求来算，不是按工具调用的数量来算。

下面是配套代码实际使用的 run_agent() 完整函数，依赖类型与两个小辅助函数都在同一个code/packages/pi_agent/loop.py 中。它比最短的伪代码多了一些配置和事件处理。

def run_agent(
    client: ModelClient,
    tools: Sequence[AgentTool],
    config: AgentLoopConfig,
    question: str,
    emit: EventSink | None = None,
) -> str:
    """运行一次无状态 Agent 循环并返回最终文本。"""
    tools_by_name = _tool_map(tools)
    unknown_required = set(config.required_tool_names) - set(tools_by_name)
    if unknown_required:
        names = "、".join(sorted(unknown_required))
        raise MiniPiError(f"必选工具未注册：{names}")
    tool_schemas = [tool.schema() for tool in tools]
    history: list[JsonObject] = [
        {"role": "user", "content": [{"type": "input_text", "text": question}]}
    ]
    required_index = 0

    for turn in range(1, config.max_turns + 1):
        tool_choice: str | JsonObject = "auto"
        if required_index < len(config.required_tool_names):
            tool_choice = {
                "type": "function",
                "name": config.required_tool_names[required_index],
            }
        request_payload: JsonObject = {
            "model": config.model,
            "instructions": config.instructions,
            "input": history,
            "tools": tool_schemas,
            "tool_choice": tool_choice,
            "parallel_tool_calls": False,
            "store": False,
        }
        reserved = set(request_payload) & set(config.request_options)
        if reserved:
            names = "、".join(sorted(reserved))
            raise MiniPiError(f"模型请求参数不能覆盖 Agent 核心字段：{names}")
        request_payload.update(config.request_options)
        response = client.create(request_payload)
        output = response_output(response)
        text = output_text(output, response)
        if text:
            _emit(emit, AgentEvent(type="assistant_text", turn=turn, text=text))

        calls = parse_function_calls(output)
        # 必须先回放模型的原始 output，再追加 call_id 对应的工具结果。
        history.extend(output)
        if not calls:
            if not text:
                raise MiniPiError("模型既没有返回文本，也没有调用工具")
            return text

        for call in calls:
            _emit(emit, AgentEvent(type="tool_call", turn=turn, call=call))
            tool = tools_by_name.get(call.name)
            result = (
                tool.execute(call.arguments)
                if tool is not None
                else ToolResult(f"未知工具：{call.name}", is_error=True)
            )
            _emit(emit, AgentEvent(type="tool_result", turn=turn, call=call, result=result))
            history.append(
                {
                    "type": "function_call_output",
                    "call_id": call.call_id,
                    "output": result.content,
                }
            )
            if (
                required_index < len(config.required_tool_names)
                and call.name == config.required_tool_names[required_index]
                and not result.is_error
            ):
                required_index += 1

    raise MiniPiError(f"达到最大轮数 {config.max_turns}，模型仍未给出最终回答")

第一遍读这个函数，可以先跳过 required_index。它是后面用来对照强制工具策略的部分；主线配置没有必选工具，tool_choice 因此保持 auto。

5.1 请求里带的是同一份不断增长的历史

history 在进入循环前创建，后面每轮都把它放进请求的 input。这意味着每次请求都能看见前面发生的事，而不是从零开始。

不过，它只属于当前这次 run_agent() 调用。函数返回以后，并没有把会话写进数据库，下一次运行也不会自动继承它。所以，说这个循环“不持久化会话”是准确的，说它“没有状态”却容易让人误解：它运行时明明有历史、轮次和策略位置。

tools_by_name 和工具声明也在循环外准备。工具集合没有变化时，不必每轮重新构建。这里无需追求复杂的缓存，放对位置就够了。

5.2 模型已经说了一句话，为什么还不能返回？

关键是先看有没有工具调用，而不是先看有没有文本。

模型可能一边说“我先看看日报”，一边请求 read_file。如果程序看见文本就 return，用户得到的只会是一句动作说明，日报还没被读取。

因此，代码允许通过事件显示中间文本，但只有 not calls 时才考虑正常结束。如果这时连文本也没有，就明确报错。它不把一个空响应解释成“模型已经完成了工作”。

也要注意，文本可以返回只是这段循环的停止规则，不是答案正确性的证明。拿到目录清单就猜订单数，显然不算完成了任务。

5.3 工具执行完，为什么不直接把结果交给用户？

第一轮的结果只是文件路径，显然还不是答案。第二轮读到的则是一段 JSON，模型还需要从中找出 orders 和 revenue_yuan，用自然语言回答用户。

列出目录以后，日报数据仍然缺失，需要接着读取文件；读到文件以后，还需要把字段整理成用户关心的答案。循环每次都根据最新的响应继续处理，而不是事先为任务规定一个固定的执行次数。

当然，不是所有产品都必须让模型重新整理工具结果。某些固定查询直接把结果返回用户就很好。但那是另一种明确的产品选择，不是这里这个通用循环默认应该做的事。

5.4 为什么要限制轮数？

一个会持续调用工具的模型，不一定在持续取得进展。它可能反复读同一个文件，也可能不停请求一个不存在的路径。

max_turns 让我们能够在有限次请求后停下来。注意这里是抛出错误，而不是把最后看到的文本拿来当最终回答。失败可以告诉用户，假装完成却会让排查变得更难。

轮数限制也不是总耗时限制。每个模型请求还有超时，工具本身也有自己的成本。真要做面向多人使用的服务，还要考虑更完整的时间预算和取消机制；本文先把这个最小边界说明白，不把它包装成生产级调度器。

六、把错误送回去之前，先分清是哪一层出了问题

读到工具的 execute() 时，你可能会想：是不是所有异常都应该变成工具结果，这样模型下一轮就能自己修正？

并不是。

如果模型请求了一个没有注册的工具，或者读取一个不存在的文件，通常还有继续反馈的价值。我们能明确知道哪次调用失败了，也能给出原因。模型下一轮可以改参数、换工具，或者告诉用户当前无法完成。

但如果整个模型接口连不上，连合法的响应都没有拿到，我们就不应该把这件事伪装成一次普通文件读取失败。调用结构已经损坏、参数不是合法 JSON 时，也不适合靠猜测补齐后继续执行，这里才是Harness框架需要做的事情之一，也就是与模型能力无关的容错机制。

遇到的问题	当前实现怎么处理
未知工具	生成带原调用 ID 的失败结果，交给下一轮
路径参数不符合工具要求	在执行前拒绝，回填受控错误
文件不存在、越界、不是文本或太大	拒绝读取，把原因作为工具结果
调用结构缺失、参数 JSON 无法解析	明确报协议错误，中止运行
模型请求超时、连接失败、响应非 JSON	明确报请求错误
空响应、轮数耗尽	明确失败，不返回半成品冒充完成

工具失败可以让模型获得调整机会，但“有机会”不等于“肯定会修复”。没有这个区别，错误回填就很容易被讲成一种自动成功的魔法。实际运行往往没有那么理想。

文件安全也是同样的道理。我们在系统提示词里要求模型只读项目文件，但真正拦住越界路径的是代码，不是那句话。配套的 ProjectFiles 从限定目录出发检查路径，拒绝绝对路径、.. 和符号链接，只读取大小受限的 UTF-8 普通文件。

这里不逐行展开底层文件 API，但也不会把保护换成一句没有边界的 Path(path).read_text()。尤其是“先打开再判断”的顺序还可能带来阻塞问题：教学副本用非阻塞方式打开文件，避免在检查并拒绝命名管道之前卡住。

这些检查并不意味着它已经是一个完整的多租户沙箱，实际的生产环境Harness的搭建要比这个复杂得多。

七、接上本地 Qwen，看看它到底有没有读文件

现在我们可以把模型、循环和文件工具组合起来了。

from packages.pi_agent import run_agent
from packages.pi_coding_agent.tools import (
    ProjectFiles,
    create_project_tools,
)
from examples.common import connect, options, report

args = options()
client, config = connect(args)

with ProjectFiles(args.project, 32 * 1024) as files:
    answer = run_agent(
        client,
        create_project_tools(files),
        config,
        args.question,
        report,
    )

print(answer)

这一层负责把项目目录、模型连接和工具放到一起。事件回调 report 则把每轮的模型文本、调用参数和工具结果打印出来。打印日志是观察过程的手段，不是另外一份会话历史；下一轮真正发送的仍然是循环里的 history。

现在回到开头的提问。在配套代码目录运行默认任务：

python3 -m examples.step04_agent_loop

用户只描述了要查询的内容，没有指定工具或执行顺序。下面是本地 Qwen3-0.6B-Q8_0 的真实运行节选：tool_choice=auto，没有必选工具；reasoning=none，temperature=0，最大轮数为 8。省去了开头的问题提示与末尾重复打印的最终回答，调用 ID 和工具结果保持原样。

[第 1 轮调用] list_files {"path": "reports"}
  call_id=call_3539f9029c4841edadd60c45fa82f292
[工具成功]
reports/daily-2026-09-06.json
[第 2 轮调用] read_file {"path": "reports/daily-2026-09-06.json"}
  call_id=call_b2da63943ab8499b8730b75a5f94c05e
[工具成功]
{
  "date": "2026-09-06",
  "orders": 128,
  "revenue_yuan": 9600
}

[第 3 轮模型文本]
这份日报的日期是2026年9月6日，订单数为128，营收为9600元。

完整记录见 evidence/step04_agent_loop-daily.json。沿着日志核对，read_file 收到的路径来自目录结果；最终回答中的日期、订单数和营收，则来自那份文件。用户提出的需求到这里才算有了数据支撑。

模型请求	模型此时知道什么	这一轮发生了什么
第1轮	用户要查日报，目录是 reports	调用 list_files，发现文件
第2轮	已知道日报的真实路径，还不知道内容	调用 read_file，取得数据
第3轮	已有路径和日报内容	不再调用工具，回答日期、订单数和营收

7.1 停在第二轮，会发生什么？

前面只写到“执行一次工具并反馈”的程序，还不能完成这个查询。可以重新运行第三个阶段，看看它停在哪里：

python3 -m examples.step03_one_round

它会执行 list_files，把目录结果交给模型，然后看见第二个 read_file 请求。到这里，程序按这个阶段的约定停下，并明确输出“模型还需要工具，任务尚未完成”。evidence/step03_one_round-daily.json 里可以看到：第二个请求已经出现，但文件还没有被读取。

第四阶段会继续处理这个读取请求，追加日报内容，再让模型给出答案。接上循环以后，程序才真正把查询做完，而不只是告诉用户下一步该做什么。

还可以把完整循环的上限调成 2：

python3 -m examples.step04_agent_loop --max-turns 2

这次两个工具都会执行，但程序仍会明确报错。因为轮数按模型请求计数：两轮都用来发起工具调用了，还缺第三轮组织最终回答。本次实测的evidence/max-turns-2.json正是这个结果。把“两个工具调用”误认为“最多只需要两次模型请求”，很容易在这里把程序提前截断。

7.2 工具之间，也需要说同一种“路径语言”

为了让第二个工具能直接使用第一个工具的结果，list_files 统一返回相对于项目根目录的路径。列出 reports 时，返回的是 reports/daily-2026-09-06.json，不是省掉目录前缀的 daily-2026-09-06.json。

read_file 也按项目根目录解释路径，因此可以直接使用目录工具的结果。路径约定统一以后，调用方不需要猜文件名是相对于当前目录，还是相对于项目根目录，后续定位问题也更容易。

提示词还要求模型完成任务后再回答，不要把“请继续打开某个文件”当成答案。用户交代的是查询需求，查找和读取应该由程序接着处理，而不是反复要求用户补做这些操作。

一次成功并不意味着模型永远不会选错工具或提前回答。实际使用时，仍然要检查数据有没有取得、答案能否对应到文件内容。日志的价值就在这里：当查询没有完成，或者数字看起来不对时，可以找到问题出在哪一步。

八、自动选择和必选工具，区别不只是一个参数

前面的查询中，用户只提出了想知道什么。程序把两个工具的声明一起交给模型，tool_choice 保持 auto，模型根据问题和已有结果选择下一步。

如果产品确实要求先列一次目录，也可以把这个流程要求明确写进配置：

required_tool_names=("list_files",)

在这个策略下，循环先把 tool_choice 设为指定的 list_files。只要列目录还没有成功，策略位置就不前进；成功之后，后面的请求恢复自动选择。第二轮的读取路径仍然来自目录结果，而不是程序预先知道的文件名。

第五个示例就是这个对照：

python3 -m examples.step05_required_tool

第五步在本次环境中也完成了日报查询，记录见 evidence/step05_required_tool-daily.json。虽然最终结果相同，两种模式的区别仍然存在：前者根据当前信息选择工具，后者由程序明确要求先完成目录查询。

自动选择	必选工具策略
模型在给定工具和提示下选择下一步	程序规定某个工具必须先成功执行
主线示例不设必选工具	对照示例只要求先成功列一次目录
适合观察完整的工具选择与反馈过程	适合表达明确的产品流程要求
仍可能填错参数、提前结束	也不能保证参数和最终答案都正确

是否设置必选工具，要看业务有没有明确的流程要求。对于这里的日报查询，保留自动选择就能让程序按已有信息继续工作；如果某个产品要求每次都先确认目录状态，再把这个要求写进配置。工具策略应该服务于任务，而不是反过来决定用户必须怎样提问。

九、配套代码怎么读，怎么一步步运行

完整代码按职责放在几个模块里。下面这份目录可以对应回前面的实现过程：

code/
├── examples/
│   ├── step01_model_call.py
│   ├── step02_tool_request.py
│   ├── step03_one_round.py
│   ├── step04_agent_loop.py
│   ├── step05_required_tool.py
│   └── common.py
├── packages/
│   ├── pi_ai/                 # 模型请求与响应解析
│   ├── pi_agent/              # 工具定义和核心循环
│   ├── pi_coding_agent/       # 项目文件工具与原组合代码
│   └── qianchat/              # 必要运行配套，正文不展开
└── sample_project/

可以顺着第一、二、三、四步运行同一个查询：先观察只有模型时还缺什么，再给程序加上工具，执行目录查询，最后把读取与回答接起来。每个阶段都对应需求中尚未完成的一部分。

第二至第五步使用同一句自然语言提问。随着实现逐渐补齐，用户不必把问题改写成工具调用指令；改变的是程序能够替他完成多少事情。

连接配置通过环境变量传入：

export QWEN_BASE_URL=http://127.0.0.1:8021
export QWEN_MODEL=qwen3-0.6b
export QWEN_REASONING=none
export QWEN_MAX_OUTPUT_TOKENS=2048

本次实测连接的是本机 8021 端口。端口不是循环逻辑的一部分，你可以通过 QWEN_BASE_URL 按本地实际情况配置，但接口协议必须匹配。

完整环境与模型文件校验值见 evidence/runtime.json，五个阶段的结果以 -daily.json 命名保存在 evidence/。其中第三步的未完成状态、第四步的三轮成功轨迹，以及轮数限制实验，是三种不同的结果，不能混成一段日志。

代码还有一组不调用真实模型的边界检查，用来稳定地制造一些模型不一定每次都会触发的情况。除了文本与调用共存、错误反馈和轮数耗尽，这次还专门检查：第一轮的目录结果能否进入第二次请求，第二轮是否按这个路径读取文件，以及两轮工具结果是否都能进入第三次请求。模型能力要用真实模型看，控制流程是否写对则不必每次都碰运气。

在配套代码目录运行 python3 -m checks.loop_boundaries，可以复现这些检查。

后续如果增加一个新的只读工具，先给它明确的输入输出，再注册进工具列表即可，不需要在循环里新增一条按工具名判断的业务分支。如果不得不让循环认识每个工具的内部细节，就该回头看看职责是不是又混在了一起。

配套源码、运行说明和真实日志已随文章整理在本地交付目录。正式开源发布前，还需要确认原代码与运行配套的授权、补齐许可证，并固定文章对应的代码版本。

十、回到最初的需求

用户想知道日报的日期、订单数和营收。现在，程序可以从这句提问出发，找到日报、读取内容，再给出有文件依据的回答。整个过程中，用户不用自己打开文件，也不用把操作拆成一条条指令。

实现里多出的历史、工具和循环，分别解决了很具体的问题：历史让后续请求知道前面查到了什么，工具负责取得模型看不到的本地数据，循环则让还没做完的查询继续往下走。

这就是我们从这个需求里得到的简易 Agent Loop。

它还没有流式输出、持久化会话或并行调度。如果后续的使用场景确实需要这些能力，可以再围绕新的问题增加。当前更重要的是，让这条最基本的查询链路清楚、可追踪，并且真正把用户的问题回答完。

评论

加载评论 滚动到评论区附近时自动加载