从零开始的 aget 开发：核心循环

agent 可能是今年最热门的领域之一了，在各类媒体的炒作下，这个词可以说几乎要变得家喻户晓了。然而，处于漩涡中心的 agent ，其实际上可能并没有多么复杂。

这是我在初步学习了一些 agent 相关知识得出的结论，至于这个结论的正确与否，我想通过自己动手来确定。这篇文章（以及相关的几篇），就是用来记录我的学习过程的。既然是自己动手，那代码当然应该出自自己的键盘，而非 coding agent 。所以这个系列的代码是、也将是出自“古法编程”的，我希望这样能对我的学习有所帮助。

本篇核心探讨：agent loop 及相关概念

一、相关概念

agent loop & ReAct

agent loop 就是智能体不断重复“观察环境 → 推理决策 → 执行动作 → 获取反馈”的循环，直到任务完成或触发停止条件。

以上是询问 LLM 获得的答案，这里描述的其实就是一个典型的 ReAct 架构。ReAct 是 agent loop 最流行的实现架构方案。简单来说，所谓 ReAct agent loop 就是一个 while 循环，循环内不断执行获取上下文、思考（Re）、调用工具（Act）、获得反馈，只要还没出结果就一直循环。

举例来说，让 agent 下载某篇论文，可能的流程是这样的：

• 第一次循环
  • 观察环境：获得用户提示词
  • 推理决策：思考用户的需求，考虑该怎么做
  • 执行动作：用搜索工具查询这篇论文
  • 获取反馈：工具返回查询结果，其中可能有某个权威网站
• 第二次循环
  • 观察环境：了解到上一轮的结果
  • 推理决策：思考搜索到的哪个结果是需要的
  • 执行动作：调用下载工具下载对应网站的论文
  • 获取反馈：获知下载成功与否
• 第三次循环
  • 观察环境：了解到下载成功
  • 推理决策：成功了，可以结束
  • 执行动作：返回最终结果

大概示意图就是这样：

三、接入 LLM

agent的核心归根结底还是大模型，所以就先从能和大模型一问一答开始吧。这个项目将使用 uv 进行管理，相关操作也将用 uv 的方式来呈现。

# 初始化项目
uv init your-project-name

LLM 需要配置 api key 来连接，这里就以 deepseek 为例，密钥写在项目的 .env 文件中，这样就不用配置环境变量污染全局环境了（.env需要被.gitignore，以免上传到仓库）：

LLM_MODEL_NAME=deepseek-v4-flash
LLM_BASE_URL=https://api.deepseek.com
LLM_API_KEY=sk-...

OpenAI 为 python 提供了便捷的库，市面上的大多数提供商也兼容 OpenAI 格式，直接使用这个库来调用 LLM：

# 添加依赖
uv add openai
uv add dotenv # 读取 .env 文件

之后，即可利用 openai 库获取大模型的回复，基于此可以直接写一个简单的对话函数：

# llm.py
import os
from dotenv import load_dotenv
from openai import AsyncOpenAI

# 读取配置文件
load_dotenv()

MODEL_NAME = os.environ["LLM_MODEL_NAME"]
asynv_client = None
# 加载client
def _get_async_client():
    """懒加载 OpenAI 客户端"""
    golbal async_cliet
    if async_client is None:
        async_client = AsyncOpenAI(
            api_key=os.getenv("LLM_API_KEY"), base_url=os.getenv("LLM_BASE_URL")
            )
    return async_client
# 流式输出
async def generate_stream(
    prompt: list[ChatCompletionMessageParam],
    timeout: float = 120,
    max_tokens: int = 3000,
) -> AsyncIterable[ChatCompletionChunk]:
    """
    异步流式输出
    """
    response = await _get_async_client().chat.completions.create(
        model=MODEL_NAME,
        messages=prompt,
        stream=True,
        reasoning_effort="high",
        extra_body={"thinking": {"type": "enabled"}},
        max_tokens=max_tokens,
        timeout=timeout,
    )
    async for chunk in response:
        yield chunk

这里使用懒加载的模式加载 OpenAI 客户端，也就是按需加载，只有在必要的时候才加载客户端，这样写有推迟获取环境变量的时机以免有时序问题、import 时启动更快、测试更友好等等优点。

另外，这里直接启用了流式输出，流式下，其中的某个chunk大概会长这样：

data: {"choices": [{"delta": {"content": " How", "role": "assistant"}, "finish_reason": null, "index": 0, "logprobs": null}], "created": 1718345013, "id": "1f633d8bfc032625086f14113c411638", "model": "deepseek-flash", "object": "chat.completion.chunk", "system_fingerprint": "fp_a49d71b8a1"}

来源：deepseek API 文档

chunk 处理起来确实比较费劲，更不要提各大模型提供商各有各的规矩，不过流式输出基本是必要的需求，也就只好多花写时间来研究了。

已经能够让 llm 开口说话了，但是它没有手脚，无法对世界产生什么影响，那么接下来就是编写一些给 agent 提供的 tools。

四、工具

相关理论

让大模型使用工具这件事其实很简单：给它一个清单，它如果想用哪个，按照指定格式输出，大模型外面的程序认识这个格式，帮它执行了，再把结果给它就行。

之前大家的方案是在提示词里面写：

所有步骤请严格使用以下 XML 标签格式输出：
- <question> 用户问题
- <thought> 思考
- <action> 采取的工具操作
- <observation> 工具或环境返回的结果
- <final_answer> 最终答案

然后期待大模型能给一个符合格式的输出。

现在的模型厂商基本都已经支持工具调用了，再也不需要自己来写了，根据 OpenAI 的格式，大概的规则如下：

• 在client.chat.completions.create时，就传入工具列表，参数 tools是由ChatCompletionFunctionToolParam组成的列表，这里的类型是openai 库中的类型。
• 大模型如果调用了工具，会在chunk.choices[0].delta.tool_calls这里输出，当这个对象不为空时，可以判断大模型调用了工具。

输入

为了获得一份工具列表，我们需要定下每个工具的名称、调用参数类型和相关信息，利用定义若干继承自BaseModel的类（pydantic库）来表示，并编写注册器来实现便捷管理：

# 注册器
CLASS_REGISTER: dict[str, type[BaseModel]] = {}

def tool(cl: type[BaseModel]):  # 表示类本身而非实例
    CLASS_REGISTER[cl.__name__] = cl
    return cl

@tool
class Ask_Question(BaseModel):
    """
    向用户问问题
    """
    question: str
    
@tool
class Web_Search(BaseModel):
    """
    通过搜索引擎搜索
    """
    question: str

这样就能获得一个key为类名称，value为类的注册器。这样做的好处是，以后你新增工具直接使用 @tool 装饰器即可，注册器会自动注册你的工具，不需要手写。

那么 LLM 所需的工具列表，即可通过：

[pydantic_function_tool(m) for m in CLASS_REGISTER.values()]

来获取。此处的 pydantic_function_tool 是 pydantic 库中，将BaseModel转换为 OpenAI 工具格式的函数。这个列表可以直接作为 tools 参数。另外，它会自动将class的__doc__转为工具描述，所以为每个class写__doc__也是很有必要的。

当然，这只能算定义了一个接口，具体的实现也需要做，可以考虑定义一些和工具名有共同点的函数（方便快捷找到对应函数，我的做法就是变更大小写），用这些函数来做实际的工作。

输出

输出方面，首先是要能听懂大模型在请求工具调用：

# 工具调用
tool_calls = chunk.choices[0].delta.tool_calls
if tool_calls:
	for tc in tool_calls or []:
		acc = tool_calls_acc.setdefault(
			tc.index, {"id": "", "name": "", "arguments": ""}
		)  # acc 作为草稿
		if tc.id:
			acc["id"] = tc.id  # 只在第一个分片出现
		if tc.function and tc.function.name:
			acc["name"] = tc.function.name  # 只在第一个分片出现
		if tc.function and tc.function.arguments:
			acc["arguments"] += tc.function.arguments  # 增量拼接

需要注意的是，大模型一次可能输出多个工具调用，所以解析时也应当将工具名和参数放到一个列表中。

之后，我的做法是将这个列表中的调用转换为真实的 python 代码来执行：

for slot in tool_calls_acc.values():  # 取出所有工具调用
	name = slot["name"].lower() # 工具对应函数就是全小写的工具名
	args = json.loads(slot["arguments"]) if slot["arguments"] else {}
	tool_res = tools.REGISTER[name](**args) # 函数注册器，和工具注册器类似，但value为函数

这样就实际执行了一次工具。

如此一来，这个 agent 已经获得了改变世界的力量，限制只在于你给的工具是什么。然而，执行完一次工具之后，agent 如果获知结果是什么也是白搭，那么接下来需要做的就是把它放在一个循环里。

五、核心循环

正如一开始说的，ReAct架构其实就是把大模型放在了一个循环里，每次循环开始时它都会获得之前的所有信息，并基于这些信息思考、动作。因此，这个循环中最重要的就是把信息拼接好，从用户问题、到大模型的输出、再到工具的结果，将一切信息组合好，再喂给下一轮的大模型。

在 OpenAI 的库中，每个消息的类型是ChatCompletionMessageParam，将这些消息组合在一起，即可作为大模型的输入。

这个类型类似于字典，区别在于字段被限制在了”role”、“content”等，而”role”又只能有”assistent” “user” “tool”。通过这些字段大模型很容易就能判断出消息的来源。

还有需要注意的一点：每次 tool 调用的结果必须被填入到列表中，否则必然报错。填入的格式是：{"role":"tool","content": tool_res, "tool_call_id": tool_call_id,}。所以，每次工具调用必须保留其 tool_call_id 并填入到工具的结果中，这个字段如果不填，也会报错。

async def core_loop(user_message: str, max_loop: int = 10) 
-> AsyncGenerator[dict[str, Any], None]:
	"""
	agent 核心循环
	"""
	# 用户的输入
	messages = [{"role": "user", "content": user_message}]

	loop = 0
	flag = True
	while (loop < max_loop) and flag: # 限制最大循环次数以免意外
		loop += 1
		tool_calls_acc: dict[Any, Any] = {} #key: index value:{"id","name","arguments"}
		content_parts = []
		# 调用 LLM 获取 chunk
		async for chunk in llm.generate_stream(messages):
			if not chunk.choices:
				continue
			choice = chunk.choices[0]
			delta = choice.delta

			# 思考内容
			reasoning_content = getattr(delta, "reasoning_content", None)
			if reasoning_content:
				yield {"type": "reasoning", "content": reasoning_content}

			# 正文
			content = getattr(delta, "content", None)
			if content:
				content_parts.append(content)
				yield {"type": "assistant", "content": content}

			# 工具调用
			tool_calls = delta.tool_calls
			if tool_calls:
				yield {"type": "tool_calls", "content": tool_calls}
				for tc in tool_calls or []:
					acc = tool_calls_acc.setdefault(
						tc.index, {"id": "", "name": "", "arguments": ""}
					)  # acc 作为草稿
					if tc.id:
						acc["id"] = tc.id  # 只在第一个分片出现
					if tc.function and tc.function.name:
						acc["name"] = tc.function.name  # 只在第一个分片出现
					if tc.function and tc.function.arguments:
						acc["arguments"] += tc.function.arguments  # 增量拼接

		new_message: ChatCompletionAssistantMessageParam = {
			"role": "assistant",
			"content": "".join(content_parts) or None,
		}
		logger.info(f"Assistant message: \n------{new_message}\n------")
		# 是否有工具调用
		if tool_calls_acc:
			# 先构建完整消息
			new_message["tool_calls"] = [
				{
					"id": slot["id"],
					"type": "function",
					"function": {
						"name": slot["name"],
						"arguments": slot["arguments"],
					},
				}
				for _, slot in sorted(tool_calls_acc.items())
			]
			messages.append(new_message)
			for slot in tool_calls_acc.values():  # 取出所有工具调用
				name = slot["name"].lower()
				args = json.loads(slot["arguments"]) if slot["arguments"] else {}
				logger.info(f"tool_call: {name}, {args}")
				try:
					tool_res = tools.REGISTER[name](**args)
					logger.info(f"tool_res:{tool_res}")
					messages.append(
						{
							"role": "tool",
							"content": str(tool_res),
							"tool_call_id": slot["id"],
						}
					)				
				# 如果出错，同样将错误传给 llm 让它自行判断
				except Exception as e:  # noqa: BLE001
					messages.append(
						{
							"role": "tool",
							"content": f"Error: {e!s}",
							"tool_call_id": slot["id"],
						}
					)
					logger.info(f"TOOL_CALL_ERROR: {e!s}")
		# 如果没有工具调用，那么可以认为输出已经结束了
		else:
			messages.append(new_message)
			flag = False

上面这个函数只要稍加改造，即可支持多轮对话（messages持久化存储，可以是内存，也可以直接入库）。

这样，一个极简的 agent 就完成了，接下来，给它 bash 工具，它就可以做到几乎任何计算机能够做到的事了（当然这样很危险）。

然而，这只是一个内核，完全称不上完整。接下来，我会在这个内核上，继续搭建一个完整的 agent 应用。

本系列代码开源于 https://github.com/l-library/maieutic ，如果有任何问题欢迎 Issues / Pr