ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

从零搭建Agent系统:最小可运行闭环与工具调用实战

从零搭建Agent系统:最小可运行闭环与工具调用实战 简介这份资源是面向AI应用开发者与软件工程师的Agent系统搭建指南配套源码包帮助读者从零理解Agent的基本概念、角色分工与工程实现路径。包内共9个文件以4个Python脚本为核心分别对应研究者、编辑者与笔记记录者三类角色的功能模块另含依赖清单、环境变量示例、项目说明文档及配置文件压缩包约15KB结构精简、便于直接运行与二次修改。资源围绕将笔记系统从离线版升级为联机版这一实践场景新增AI搜索与报告生成能力并自动归档笔记同时涉及RAG与Agent在AI应用中的衔接思路。目前已有150人学习适合具备一定Python基础、希望快速跑通多角色协作Agent流程并理解检索增强生成架构的开发者参考。1. 从零搭一套 Agent 系统为什么“能跑起来”比“架构漂亮”重要十倍很多人第一次搭 Agent 系统卡住的地方不是不会写代码而是被各种概念绕晕规划器、执行器、记忆、工具调用、反思循环……看了一堆架构图回头发现连一个能跑的最小闭环都没有。我见过太多团队花两周画了一张漂亮的 Agent 架构图结果连“让模型查一次天气并返回结果”都跑不通。Agent 系统的本质不是架构是一个能自主决策、调用工具、根据结果调整下一步动作的循环。你需要的不是完美设计而是一个今天就能跑起来、明天能加工具、后天能换模型的最小可运行源码。这篇文章面向两类人一是想动手搭 Agent 但不知道从哪下手的工程师二是已经用过大模型 API 但没做过工具调用编排的开发者。我会从最小可运行闭环讲起逐步加上工具注册、记忆管理、多步规划每一步都给可运行的代码和参数说明。热搜词里提到的“系统提示词工程和 skill agent 有什么区别”我也会在工具注册那一章用实际代码说清楚——不是概念辨析是你在写代码时到底该把逻辑放在 system prompt 里还是放在工具函数里。读完你应该能拿到一套自己能改、能扩展、能调试的 Agent 骨架而不是一个只能看不能动的 demo。2. 最小可运行 Agent 闭环从一次工具调用开始2.1 为什么先写循环而不是先写架构Agent 和普通 LLM 调用的核心区别只有一个模型输出不再直接返回给用户而是先经过一层“决策解析”判断是否需要调用工具如果需要就执行工具把结果塞回上下文再让模型继续决策。这个循环就是 Agent 的心脏。你先把这颗心脏搭出来哪怕只有一个工具、没有记忆、没有规划它也是一个真正的 Agent。我一般建议新手先写一个“单工具 Agent”只注册一个计算器工具让模型自己决定什么时候调用。跑通之后加工具、加记忆、加多步规划都是在这个循环上做加法。反过来先设计一堆抽象层再写循环大概率会翻车——因为你对模型实际怎么输出工具调用格式还没有体感。下面是最小闭环的 Python 实现依赖只有openai包你也可以换成任何兼容 OpenAI 接口的模型服务import json from openai import OpenAI client OpenAI(api_keyyour-key, base_urlyour-base-url) # 1. 定义工具计算器 tools [ { type: function, function: { name: calculator, description: 执行数学计算支持加减乘除和幂运算, parameters: { type: object, properties: { expression: { type: string, description: 数学表达式如 2 3 * 4 } }, required: [expression] } } } ] # 2. 工具的实际执行函数 def run_calculator(expression: str) - str: try: # 注意生产环境不要直接用 eval这里仅演示 result eval(expression, {__builtins__: {}}, {}) return str(result) except Exception as e: return f计算错误: {e} # 3. Agent 主循环 def agent_loop(user_input: str, max_turns: int 5): messages [ {role: system, content: 你是一个助手需要计算时调用 calculator 工具。}, {role: user, content: user_input} ] for turn in range(max_turns): response client.chat.completions.create( modelgpt-4o-mini, # 换成你实际用的模型 messagesmessages, toolstools, tool_choiceauto ) msg response.choices[0].message messages.append(msg) # 如果没有工具调用直接返回文本 if not msg.tool_calls: return msg.content # 执行每个工具调用 for tool_call in msg.tool_calls: fn_name tool_call.function.name fn_args json.loads(tool_call.function.arguments) if fn_name calculator: result run_calculator(fn_args[expression]) else: result f未知工具: {fn_name} messages.append({ role: tool, tool_call_id: tool_call.id, content: result }) return 达到最大轮次限制未能完成。 # 4. 跑起来 if __name__ __main__: print(agent_loop(帮我算一下 (15 27) * 3 等于多少))这段代码的关键在agent_loop里的for turn in range(max_turns)循环。每一轮做三件事调模型、检查是否有tool_calls、有就执行并把结果以role: tool追加回messages。max_turns是安全阀防止模型陷入无限调用循环——这个参数我建议设 5 到 10太小会导致复杂任务做不完太大会浪费 token。tool_choiceauto让模型自己决定是否调用工具。如果你明确知道某轮必须调用工具可以设成{type: function, function: {name: calculator}}强制指定。tools数组里的description和parameters描述非常关键——模型完全靠这些文字判断什么时候该调用、参数怎么填。描述写得含糊模型就会该调不调、该传参不传参。2.2 工具注册的三种写法与选型上面是最直白的“手写 JSON Schema”方式。实际项目里工具多了之后手写 schema 会变得难以维护。常见做法有三种第一种装饰器自动生成 schema。用 Python 的inspect模块从函数签名和 docstring 自动提取参数定义。优点是写工具函数时不用管 schema缺点是复杂嵌套参数不好表达。import inspect def tool(name: str, description: str): def decorator(fn): sig inspect.signature(fn) properties {} required [] for param_name, param in sig.parameters.items(): properties[param_name] { type: string, # 简化处理实际需根据注解映射 description: param.annotation.__doc__ if param.annotation ! inspect.Parameter.empty else } if param.default inspect.Parameter.empty: required.append(param_name) fn._tool_schema { type: function, function: { name: name, description: description, parameters: { type: object, properties: properties, required: required } } } return fn return decorator tool(nameget_weather, description查询指定城市的天气) def get_weather(city: str) - str: city: 城市名称如 北京 return f{city}今天晴25度第二种用 Pydantic 模型定义参数。适合参数结构复杂、需要校验的场景。OpenAI 官方 SDK 就支持直接传 Pydantic 模型。第三种配置文件驱动。把工具定义写在 YAML 或 JSON 里运行时加载。适合工具数量多、需要动态增减的场景但调试时多一层间接。我一般会这样选工具少于 10 个用装饰器10 到 30 个用 Pydantic超过 30 个或者需要非开发者也能加工具时上配置文件。不要一上来就搞最复杂的方案。2.3 系统提示词工程和 skill agent 的区别在代码里怎么看热搜里那个问题——“系统提示词工程和 skill agent 有什么区别”——放到代码里其实很清楚。系统提示词工程是把行为约束写在systemmessage 里比如“你是一个数学助手遇到计算必须调用 calculator”。skill agent 是把能力封装成独立的工具函数或子 Agent模型通过工具调用来使用。区别在于提示词是“告诉模型怎么做”工具是“给模型一个能做这件事的手柄”。提示词能约束风格、语气、输出格式但没法让模型真的去查数据库、发请求、读文件。工具能扩展模型的能力边界但工具本身不知道什么时候该被调用——那部分还是靠提示词和工具描述来引导。实际项目里两者是配合的system prompt 里写“你有以下工具可用遇到 X 场景调用 Y 工具”工具函数里写具体执行逻辑。如果你把本该做成工具的逻辑硬塞进 system prompt比如让模型自己“模拟”查天气结果就是模型编造数据。反过来把本该用提示词约束的格式要求做成工具会浪费调用轮次。提示判断标准很简单——需要访问外部系统或执行副作用的做成工具只影响模型输出内容和风格的写在 system prompt 里。3. 给 Agent 加上记忆和多步规划从玩具到能干活的系统3.1 短期记忆与长期记忆的分层设计最小闭环里的messages数组就是短期记忆——它保存了当前对话的所有轮次。但有两个问题一是 token 会随着轮次增长线性膨胀二是关掉进程就没了。所以实际系统里需要分层短期记忆当前会话的messages保留最近 N 轮或最近 M 个 token。超出后做摘要压缩。常见做法是保留 system message 最近 10 轮对话 一个“之前对话摘要”的 system message。长期记忆跨会话持久化的信息比如用户偏好、历史任务结果。通常用向量数据库存 embedding需要时检索相关片段塞回上下文。下面是一个带滑动窗口和摘要的短期记忆实现class ConversationMemory: def __init__(self, max_tokens: int 3000, keep_recent: int 6): self.system_prompt self.messages [] # 完整历史 self.summary # 旧对话摘要 self.max_tokens max_tokens self.keep_recent keep_recent def add(self, role: str, content: str): self.messages.append({role: role, content: content}) self._compress_if_needed() def _compress_if_needed(self): # 粗略估算1 token ≈ 4 字符中文约 1.5 字符 total_chars sum(len(m[content]) for m in self.messages) if total_chars self.max_tokens * 3: return # 保留最近 keep_recent 条其余压缩成摘要 old self.messages[:-self.keep_recent] recent self.messages[-self.keep_recent:] # 实际项目里这里调一次模型做摘要 old_text \n.join(f{m[role]}: {m[content]} for m in old) self.summary f之前对话摘要{old_text[:500]}... # 简化演示 self.messages recent def get_context(self) - list: ctx [{role: system, content: self.system_prompt}] if self.summary: ctx.append({role: system, content: self.summary}) ctx.extend(self.messages) return ctxmax_tokens和keep_recent这两个参数需要根据你的模型上下文窗口来调。如果模型支持 128K 上下文max_tokens可以设大一些减少摘要频率如果只有 8K就得压缩得激进一点。keep_recent我一般设 6 到 10保证最近几轮的工具调用结果还在上下文里模型不会“忘记”自己刚查了什么。长期记忆的接入方式是在get_context里加一步检索用当前用户输入去向量库查 top-k 相关片段拼到 system message 里。注意检索结果要标注来源和时间否则模型会把旧信息当当前事实用。3.2 多步规划让 Agent 自己拆任务单轮工具调用只能解决“查一次天气”这种一步任务。真实场景往往是“帮我分析这份销售数据找出下降原因并生成一份报告”——这需要多步读文件、计算、可能再查外部数据、最后生成文本。有两种做法做法一ReAct 循环。就是第 2 章那个循环的扩展版模型每轮输出“思考 行动”执行后观察结果再决定下一步。优点是灵活缺点是轮次多、token 消耗大、容易跑偏。做法二先规划再执行。让模型先输出一个任务列表plan然后逐步执行每个子任务每步可以调用工具。优点是结构清晰、可控缺点是规划错了后面全错。我一般会混合用先让模型出一个粗粒度 plan然后每个 plan 步骤内部用 ReAct 循环执行。下面是规划器的核心代码PLANNER_PROMPT 你是一个任务规划器。根据用户需求输出一个 JSON 格式的任务列表。 每个任务包含: {step: 序号, action: 动作描述, tool: 需要的工具名或null} 只输出 JSON不要其他内容。 用户需求: {user_input} 可用工具: {tool_names} def plan_task(user_input: str, tool_names: list) - list: prompt PLANNER_PROMPT.format( user_inputuser_input, tool_names, .join(tool_names) ) response client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], response_format{type: json_object} # 强制 JSON 输出 ) plan json.loads(response.choices[0].message.content) return plan.get(tasks, [])response_format{type: json_object}这个参数很关键——不加的话模型可能输出带 markdown 代码块的 JSON解析会失败。即使加了也建议在json.loads外面包一层 try-except并且对解析失败做重试。规划器的 prompt 里我特意加了“可用工具”列表这样模型规划时会考虑哪些步骤有工具支撑不会规划出“用浏览器打开网页”这种没有对应工具的动作。如果你的工具很多这里可以只传工具名和一句话描述不用传完整 schema。3.3 工具执行的安全边界与超时控制Agent 自己决定调用什么工具、传什么参数这意味着你必须假设它会传错参数、调用不该调的工具、甚至陷入死循环。三个必须做的防护参数校验工具函数入口做类型和范围检查。比如文件路径工具要限制在允许的目录内SQL 工具要拒绝 DROP 和 DELETE 语句。超时控制每个工具调用设独立超时不能让一个卡住的 HTTP 请求拖死整个 Agent。Python 里可以用concurrent.futures或asyncio.wait_for。import concurrent.futures def safe_tool_call(fn, args: dict, timeout: int 10): with concurrent.futures.ThreadPoolExecutor(max_workers1) as executor: future executor.submit(fn, **args) try: return future.result(timeouttimeout) except concurrent.futures.TimeoutError: return f工具执行超时{timeout}秒 except Exception as e: return f工具执行异常: {e}轮次上限前面提过的max_turns在规划执行模式里还要加一个“单步最大重试次数”防止某个子任务反复失败反复重试。注意不要给 Agent 直接执行 shell 命令或写文件的权限除非你做了严格的沙箱和白名单。我见过 Agent 把测试环境的配置文件覆盖掉的案例血泪经验。4. 避坑与排查Agent 跑不起来时先查这五个地方4.1 模型不调用工具只输出文本现象明明注册了工具模型却直接回答“我无法查询天气”或者自己编一个天气结果。原因三种可能——工具描述太模糊模型没理解什么时候该用system prompt 里没有引导模型本身对 function calling 支持不好。解决先检查tools数组里的description是否写清楚了“什么场景下用这个工具”。然后检查 system prompt 里有没有类似“遇到 X 必须调用 Y 工具”的指令。如果都写了还不调换一个 function calling 支持更好的模型试试。有些小模型虽然接口兼容但工具调用能力很弱。4.2 工具调用参数解析失败现象json.loads(tool_call.function.arguments)报 JSONDecodeError。原因模型输出的 arguments 不是合法 JSON常见于参数值里包含引号、换行符或者模型输出了多余的文字。解决在解析外面包 try-except解析失败时把原始字符串作为参数传给模型让它重新生成。更稳妥的做法是在工具 schema 里把参数类型都设成 string让模型不用处理复杂嵌套。如果用的是支持 structured output 的模型优先用那个。4.3 多轮对话后模型“忘记”了之前的工具结果现象Agent 查了天气之后下一轮又问“你查到了吗”或者基于错误的信息继续推理。原因记忆压缩时把工具调用结果截断或摘要掉了或者messages里 tool 消息的tool_call_id对不上。解决检查压缩逻辑是否保留了最近的 tool 消息。tool_call_id必须和模型输出的tool_calls[].id完全一致不能自己生成。如果用了摘要确保摘要里包含了关键的工具返回数据。4.4 Agent 陷入无限循环现象Agent 反复调用同一个工具或者两个工具来回调用直到达到 max_turns。原因工具返回的结果没有让模型得到新信息模型认为任务没完成继续调或者规划器的 plan 里有循环依赖。解决在工具返回结果里加上明确的状态标识比如“查询成功结果如下”或“查询失败原因如下”。模型看到“成功”就知道不用再查了。另外可以在循环里加一个检测如果连续两轮调用了同一个工具且参数相同强制中断并返回当前结果。4.5 换了模型之后整个 Agent 行为大变现象开发时用 A 模型跑得好好的换成 B 模型后工具不调了、格式乱了、规划不出来了。原因不同模型的 function calling 格式支持程度不同有的对 system prompt 的遵循程度不同有的对 JSON 输出更严格。解决把模型名做成配置项不要硬编码。换模型时重点测三个地方工具调用是否正常触发、JSON 解析是否通过、多步规划是否合理。如果 B 模型不支持 function calling就得退回到“让模型输出特定格式文本自己解析”的方案。5. 进阶技巧用“工具描述 A/B 测试”把 Agent 成功率从 60% 拉到 90%Agent 系统搭起来之后最影响成功率的往往不是模型能力而是工具描述的质量。同一个工具描述写得好和写得差模型调用准确率能差 30 个百分点。我一般会做一轮“工具描述 A/B 测试”准备 20 条典型用户输入每条标注“应该调用哪个工具、传什么参数”然后跑两版描述对比准确率。具体做法是写一个评测脚本test_cases [ {input: 北京今天多少度, expect_tool: get_weather, expect_args: {city: 北京}}, {input: 帮我算 100 除以 7, expect_tool: calculator, expect_args: {expression: 100 / 7}}, # ... 更多用例 ] def evaluate(tools_config): correct 0 for case in test_cases: # 用当前 tools_config 跑一次 agent_loop只取第一轮的工具调用 result agent_loop(case[input], toolstools_config, max_turns1) if result.tool_name case[expect_tool]: correct 1 return correct / len(test_cases)跑完对比两版描述把准确率高的那版留下来。描述优化的方向通常是把工具名写得更具体get_weather比weather好、在 description 里写清楚“什么时候用”和“什么时候不用”、参数描述里给一个具体例子。另一个技巧是给工具加“前置条件”描述。比如一个“发送邮件”工具description 里写“仅在用户明确要求发送邮件时调用不要用于查询邮件”。这样能减少误调用。最后说一个我自己的习惯每次改完工具描述或 system prompt都跑一遍那 20 条评测用例记录准确率变化。不跑评测就改 prompt等于闭着眼睛调参。这套评测集不用很大20 到 50 条就够关键是覆盖你实际场景里的高频输入和边界情况。希望帮到你。本文还有配套的精品资源点击获取
返回列表