前段时间在梳理开源 Agent 项目时,看到 PrimeIntellect-ai / prime-agent 这个名字,第一反应是这类仓库到底在解决什么问题。很多人以为 Agent 就是“接个大模型 API,问一句答一句”,但实际动手做一个最小可运行的 Agent 项目就会发现,真正复杂的不是模型本身,而是“怎么让模型学会调用工具、怎么把外部结果回传给模型、怎么控制一轮又一轮的循环直到任务完成”。
这篇文章就以 PrimeIntellect-ai / prime-agent 这类开源智能体项目为灵感,带你从零搭建一个最小可运行的 AI Agent 实战 Demo。不需要大型分布式集群,也不需要复杂的 Agent 框架,只需要一个 Python 环境、一个 OpenAI 兼容的模型 API,就能把 Agent 的核心闭环跑起来。
我相信读完并照着敲完,你会比只看概念文章更能理解 Agent 的工作机制。
1. 背景与核心概念
1.1 什么是 AI Agent
AI Agent(智能体)可以简单理解成“一个会自主调用工具来完成任务的 AI 程序”。它和我们平时使用的 ChatBot 最大的区别在于:
- ChatBot 只负责生成文本,无法操作外部系统。
- Agent 则多了一个“感知环境 → 决策 → 调用工具 → 获取结果 → 继续决策”的循环能力。
比如用户问“帮我看看现在几点”,普通聊天模型只能基于训练数据回答,可能不准。而 Agent 可以驱动模型生成一次“工具调用指令”,由程序读取本地时间后再把结果交给模型,最终输出真实时间。
1.2 PrimeIntellect-ai / prime-agent 给开发者的启发
PrimeIntellect-ai / prime-agent 这类名字看起来很“硬核”,大概率是 GitHub 上某个和去中心化 AI、开源算力或智能体工程相关的仓库。它对我的启发是:
- 当前 AI 领域已经不满足于“单次问答”,而是追求“任务自动执行”。
- Agent 项目普遍采用“模型 + 工具 + 循环控制”三层结构。
- 开源项目之间互相借鉴很快,但只要吃透最小闭环,再复杂的新框架也无非是在这个闭环上做抽象和扩展。
换句话说,理解 Agent 的通用运行机制,比死记某个框架的 API 更重要。
1.3 Agent、RAG、工作流三者的区别
这几个概念经常混淆:
| 概念 | 核心能力 | 适用场景 |
|---|---|---|
| ChatBot | 对话生成 | 客服、闲聊 |
| RAG | 检索增强生成 | 知识库问答、文档问答 |
| Agent | 自主规划与工具调用 | 自动化任务、操作外部系统 |
| Workflow | 固定流程编排 | 流程明确、步骤固定的业务逻辑 |
Agent 和其他三者的最大差异在于,它让大模型参与了“决策”,而不是只做信息处理。
2. 环境准备与版本说明
2.1 运行环境
本文示例以以下环境为例:
- 操作系统:Windows 10 / macOS / Ubuntu 均可
- Python:3.10 或更高版本
- 包管理工具:pip
- 模型 API:任意支持 OpenAI 协议的大模型服务
- 开发 IDE:VS Code 或 PyCharm
请根据你自己的项目实际情况调整版本,这里重点演示完整落地思路。
2.2 创建虚拟环境
为了避免污染系统全局 Python 环境,建议先创建虚拟环境:
mkdir prime-agent-demo cd prime-agent-demo python -m venv venv # Windows venv\Scripts\activate # macOS / Linux source venv/bin/activate激活虚拟环境后,终端提示符前面会出现(venv)标记。
2.3 安装依赖
新建requirements.txt文件,写入以下内容:
openai>=1.0.0 python-dotenv>=1.0.0执行安装:
pip install -r requirements.txt其中openai官方 Python SDK 用于调用兼容 OpenAI 协议的接口,python-dotenv用于加载.env文件中的环境变量。
2.4 项目目录结构
设计一个尽量简化的项目结构:
prime-agent-demo/ ├── venv/ ├── .env.example ├── requirements.txt ├── tools.py ├── llm_client.py └── agent.py分工如下:
tools.py:定义 Agent 可以调用的工具。llm_client.py:封装大模型 API 调用。agent.py:Agent 主循环以及入口。.env.example:环境变量示例。
3. Agent 的核心机制设计
3.1 工具注册机制
Agent 要调用工具,首先得把工具“注册”到一个映射表里。这样程序才能根据模型返回的工具名称,找到对应的函数并执行。
注册机制有很多种写法,最简单的是用一个字典:
TOOL_REGISTRY = {} def register_tool(): def wrapper(func): TOOL_REGISTRY[func.__name__] = func return func return wrapper以后新增工具时,只需要在函数上加上@register_tool()装饰器即可。这种方式很容易扩展,也方便统一做参数校验。
3.2 LLM 对话协议
Agent 循环中,我们会在一条消息列表里持续追加消息:
system:系统指令,告诉模型它的身份和任务。user:用户输入。assistant:模型回答,可能包含tool_calls。tool:工具执行结果,必须和tool_call_id一一对应。
大模型在每一轮都会看到之前所有的消息,这样它才能知道自己刚才调用了什么工具、工具返回了什么结果。
3.3 终止条件
循环不能无限跑下去,否则会产生大量 API 调用费用,甚至可能陷入死循环。因此要设置两个终止条件:
- 模型返回了
content且没有工具调用,说明已经可以给出最终答案。 - 循环次数达到
max_steps上限,无论有没有结果都强制结束。
这种设计是 Agent 工程中最基础,也最重要的兜底策略。
4. 完整实战:实现一个最小 Agent
4.1 配置环境变量
创建.env.example文件:
OPENAI_API_KEY=sk-你的密钥 OPENAI_BASE_URL=https://api.openai.com/v1 OPENAI_MODEL=gpt-4o-mini如果你使用的是国内兼容 OpenAI 协议的模型服务,请改成对应的base_url和model。
实际使用时,把.env.example复制为.env:
cp .env.example .env4.2 编写工具层
文件路径:tools.py
import datetime import json import re TOOL_REGISTRY = {} def register_tool(): def wrapper(func): TOOL_REGISTRY[func.__name__] = func return func return wrapper @register_tool() def current_time(tz: str = "local"): """返回当前时间。示例参数:{"tz": "local"}""" if tz == "utc": return datetime.datetime.now(datetime.timezone.utc).isoformat() return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") @register_tool() def calculator(expr: str): """计算简单四则运算,例如 1+2 或 (3+4)*5。 注意:这是 Demo 级别的实现,生产环境请使用 AST 解析或更安全的表达式计算库。 """ expr = expr.replace(" ", "") if not re.fullmatch(r"[\d+\-*/().]+", expr): return "表达式包含非法字符" try: return str(eval(expr, {"__builtins__": {}}, {})) except Exception as e: return f"计算失败: {e}" def dispatch(tool_name: str, args: dict) -> str: if tool_name not in TOOL_REGISTRY: return f"未知工具: {tool_name}" try: return TOOL_REGISTRY[tool_name](**args) except TypeError as e: return f"工具参数不匹配: {e}"这里解释几个关键点:
register_tool装饰器把函数名作为 key 注册到全局字典。dispatch负责动态分发,参数用**args解包。eval在演示中很方便,但存在注入风险,生产环境一定要换成安全的表达式解析方案。
4.3 封装大模型客户端
文件路径:llm_client.py
import os from openai import OpenAI class LLMClient: def __init__(self): self.client = OpenAI( api_key=os.getenv("OPENAI_API_KEY", "EMPTY"), base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"), ) self.model = os.getenv("OPENAI_MODEL", "gpt-4o-mini") def chat(self, messages, tools): resp = self.client.chat.completions.create( model=self.model, messages=messages, tools=tools, tool_choice="auto", ) return resp.choices[0].messagetools参数需要传入工具描述列表,也就是 JSON Schema 格式。tool_choice="auto"表示让模型自己决定是否调用工具、调用哪个工具。
4.4 编写 Agent 主循环
文件路径:agent.py
import json import os from dotenv import load_dotenv from llm_client import LLMClient from tools import dispatch SYSTEM_PROMPT = """你是一个自动化 Agent。 你需要根据用户请求决定是否调用工具。 工具调用结果会以 observation 的形式返回。 当结果可以回答用户时,直接输出最终答案,不要再调用工具。""" TOOL_SCHEMAS = [ { "type": "function", "function": { "name": "current_time", "description": "获取当前本地时间", "parameters": { "type": "object", "properties": { "tz": { "type": "string", "enum": ["local", "utc"], "description": "时区" } }, "required": ["tz"] } } }, { "type": "function", "function": { "name": "calculator", "description": "计算简单四则运算,例如 1+2 或 (3+4)*5", "parameters": { "type": "object", "properties": { "expr": { "type": "string", "description": "数学表达式" } }, "required": ["expr"] } } } ] def run_agent(user_input: str, max_steps: int = 5): client = LLMClient() messages = [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_input} ] for step in range(max_steps): print(f"\n第 {step + 1} 轮模型请求...") message = client.chat(messages, TOOL_SCHEMAS) if message.tool_calls: # 将包含工具调用请求的 assistant 消息加入上下文 messages.append(message.model_dump()) for tc in message.tool_calls: tool_name = tc.function.name try: args = json.loads(tc.function.arguments or "{}") except json.JSONDecodeError: args = {} result = dispatch(tool_name, args) print(f"调用工具 {tool_name},参数 {args},结果 {result}") # 将工具执行结果加入上下文 messages.append({ "role": "tool", "tool_call_id": tc.id, "content": result }) continue if message.content: print("最终回答:", message.content) return message.content print("达到最大迭代次数,未生成最终回答。") return None if __name__ == "__main__": load_dotenv() user_input = input("请输入你的问题: ") run_agent(user_input)主循环可以做如下理解:
- 把
system和user消息发给模型。 - 模型返回的
message如果带tool_calls,说明模型想调用工具。 - 我们把这条 assistant 消息原样放入上下文。
- 逐个执行工具调用,并把
role=tool的结果消息放回上下文。 - 进入下一轮循环。
- 如果模型返回的内容里没有工具调用,直接作为最终结果输出。
4.5 为什么需要把 assistant 消息也追加回去
这个细节最容易踩坑。如果不把message.model_dump()对应的助理消息追加回messages,API 会报错。因为 OpenAI 协议要求tool消息必须紧跟在对应的assistant消息之后,并且tool_call_id必须真实存在。
把 assistant 消息加回去,模型才能知道“我刚才决定调用计算器”,结合工具返回的结果进行下一步判断。
5. 运行与验证
5.1 启动 Agent
确保.env文件配置正确后,执行:
python agent.py输入:
请输入你的问题: 计算 (3+4)*5预期输出类似:
第 1 轮模型请求... 调用工具 calculator,参数 {'expr': '(3+4)*5'},结果 35 第 2 轮模型请求... 最终回答: 计算结果是 35再测试时间查询:
请输入你的问题: 现在几点?预期输出类似:
第 1 轮模型请求... 调用工具 current_time,参数 {'tz': 'local'},结果 2024-06-01 15:30:22 第 2 轮模型请求... 最终回答: 当前时间是 2024-06-01 15:30:22。如果模型判断不需要调用工具的问题,比如“你好”,第一轮可能直接返回最终回答,不会进入工具调用分支。
5.2 验证循环终止
为了验证max_steps兜底是否有效,可以故意把max_steps改成1,然后问一个必须多次调用工具才能完成的问题。此时 Agent 会在第一轮循环结束后直接退出,不会无限请求。
这种边界测试在真实开发中非常重要,尤其是接入复杂工具链时。
6. 常见问题与排查思路
| 问题现象 | 常见原因 | 解决思路 |
|---|---|---|
| 调用 API 报 401 认证失败 | 密钥错误或没有配置.env | 检查OPENAI_API_KEY和base_url |
| 模型返回 content 为 None | 模型只生成了工具调用,未生成文本 | 不要急着判空,先检查tool_calls |
| 工具参数经常解析失败 | 模型生成的 JSON 参数格式异常 | 用json.loads捕获异常,返回给模型重试 |
| Agent 陷入无限循环 | 缺少终止条件 | 设置max_steps,建议 5 或 8 |
| 上下文越来越长,费用变高 | 每次循环都携带全部历史消息 | 对历史消息做裁剪或摘要 |
tool消息报错 | 缺少对应的 assistant 消息 | 确保tool_calls对应的 assistant 消息先追加 |
| 计算器工具执行危险代码 | 使用 eval 处理表达式 | 生产环境改用 AST 安全解析 |
如果遇到“工具明明存在,但模型一直不调用”的问题,优先检查工具描述是否写清楚。例如current_time的参数中required写了["tz"],模型就必须生成tz参数,否则工具调用会不完整。
7. 最佳实践与工程建议
7.1 工具设计原则
工具不宜太多,建议一次只给模型暴露 5~10 个。工具描述要明确:
- 这个工具是干什么的。
- 参数的类型和取值范围。
- 什么情况下应该调用这个工具。
- 什么情况下不应该调用。
如果模型经常选错工具,多数是描述写得不够清晰,而不是模型本身的问题。
7.2 参数校验与安全边界
Agent 的工具最终一定会执行真实操作,因此对参数必须做严格校验。尤其是涉及文件删除、数据库更新、网络请求的工具,必须遵循最小权限原则:
- 不允许传入任意 shell 命令。
- 数据库操作前必须备份。
- 生产系统变更前必须经过测试环境验证。
7.3 日志与可观测性
Agent 循环的每一步都要留下日志:
- 用户原始输入。
- 模型生成的工具调用。
- 工具返回结果。
- 当前消息列表长度。
- API 调用耗时。
这些信息能帮助你在线上快速定位问题。
7.4 成本控制
模型每轮循环都会调用一次 API,且上下文会不断膨胀。建议:
- 设置单次任务的最大轮数。
- 设置单次任务的 token 上限。
- 合并历史工具结果,只保留关键信息。
- 对简单任务使用更便宜、更小尺寸的模型。
7.5 从 Demo 到生产
Demo 里所有代码都在本地进程内同步执行,适合验证思路。生产环境通常还要考虑:
- 任务队列与异步执行。
- 可恢复的任务状态存储。
- 多用户并发隔离。
- 工具调用的超时限制。
- 模型输出的结构化验证。
建议先把工具层做成独立服务,再通过 HTTP 或消息队列和 Agent 主进程通信,这样后续扩展会容易很多。
8. 总结与下一步动手方向
这篇文章从一个开源 Agent 项目名字出发,拆解了 Agent 最核心的调用循环原理,并带你完成了一个可运行的最小 Demo。你至少已经掌握:
- 工具注册与动态分发。
- OpenAI 协议中
assistant与tool消息的搭配方式。 - Agent 主循环的终止条件。
- 常见报错的排查思路。
下一步可以做两个方向的尝试:一是给 Agent 增加更多真实工具,比如天气查询、数据库查询、文件读写;二是引入更强的结构化输出方式,要求模型严格按 JSON 返回决策结果,而不是依赖自然语言描述。
如果你只是想在项目里调研 Agent 是否可行,建议先用这个几十行的最小闭环跑通业务场景,再决定是否引入重框架。先把循环控制打扎实,后续无论换成什么 Agent 项目,都能很快上手。