很多开发者第一次接触 Agent 时,都会有一个共同困惑:我已经学会调用大模型的 API 了,每次都能拿到不错的回答,为什么还要学 Agent?这其实是把“聊天”和“干活”混为一谈了。大模型 API 本质上是一次性的问答引擎,给它一句输入,它返回一句输出,说完就结束了。AI Agent 则是围绕大模型搭起来的一套“感知-决策-行动-记忆”循环系统,它能把模型能力接入真实的业务流程里,自动调用工具、查询数据、修正错误,最后交付一个实际结果。
这篇文章不是简单堆概念,我会从零基础假设开始,带你从环境准备、最小 Agent 循环、Function Calling、记忆管理,一直写到企业级工单处理 Agent 的完整工程实现。如果你正在做 RAG 应用、大模型应用落地,或者准备转型 AI 应用开发工程师,这篇文章应该能帮你省掉不少弯路。读完你可以获得几条直接可用的代码框架、一套企业级 Agent 设计的核心思路,以及大量真实项目中才会遇到的坑和排查方法。
1. 这篇教程真正要解决的问题
过去几年,大模型应用开发经历了一个明显变化:第一波热潮是“Prompt 工程”,大家研究提示词怎么写;第二波热潮是“RAG”,把知识库检索和生成结合起来;到了现在,“Agent”成了新的关键词。但很多团队的困惑在于,看了大量 Agent 概念文章之后,动手写第一个项目时仍然不知道从哪开始。
我们真正要解决的是四个问题。
第一个问题:聊天不等于干活。大模型 API 只能基于训练数据和上下文回答问题,它不能替你创建工单、不能查库存、不能发告警。Agent 通过工具调用机制让模型具备“行动能力”。
第二个问题:无记忆等于失忆。企业级对话场景里,用户前一句说“帮我建一个工单”,后一句说“状态改成都处理中”,Agent 必须能记住上下文,否则业务根本无法闭环。
第三个问题:工程化不等于写脚本。一个能跑的 Agent Demo 和生产级 Agent 之间有巨大鸿沟,涉及存储、鉴权、限流、可观测性、数据隔离、模型切换、成本控制等一堆问题。
第四个问题:评测难。普通接口返回的是字符串,对比容易;Agent 会调用工具、产生多步推理,到底算不算成功,需要一套评测思路。
所以这篇文章的定位很明确:不讨论过于玄幻的“通用人工智能”,只讲 2026 年当下可落地、可运行、可上生产的 AI Agent 开发方法。适合三种读者:刚入门想系统学习 Agent 智能体开发教程的开发者;已经在做 RAG 但想更进一步接入工具的工程师;需要设计企业级 Agent 方案的架构师。
2. AI Agent 的核心概念与运行逻辑
理解 Agent,最忌讳一上来就看各种复杂的 Agent 框架。更靠谱的方式是先搞清楚 Agent 的本质是一个循环控制结构。
2.1 Agent 与普通 API 调用的区别
普通大模型 API 调用是“请求-响应”模式:
- 模型收到用户消息
- 返回一个回复
- 结束
Agent 则是“循环执行”模式:
- 模型收到用户消息后,根据任务判断是否需要调用工具
- 如果需要,生成一个结构化的工具调用请求
- 程序执行工具,将结果返回给模型
- 模型基于工具结果继续推理,再次决定下一步动作
- 直到模型认为任务完成,输出最终答案
这个模式在学术上通常被称为ReAct(Reasoning + Acting),即“推理 + 行动”交替进行。模型先思考“我需要什么信息”,再行动“调用某个工具”,最后观察“工具返回的结果”,如此反复。
2.2 Agent 四大核心要素
一个完整的 Agent 系统,通常由四个部分组成。
大脑:大模型本身,负责理解任务、生成决策、判断何时停止。它决定 Agent 的“聪明程度”。
工具:Agent 能调用的外部能力,例如查询接口、写入接口、搜索引擎、计算器、数据库操作。它决定 Agent 的“行动边界”。
记忆:分为短期记忆和长期记忆。短期记忆就是当前对话上下文,长期记忆则依赖外部存储,让 Agent 跨会话记住用户偏好和历史事实。
编排循环:把大脑、工具、记忆连接在一起的执行引擎。它决定 Agent 是严格按照固定流程执行,还是可以自由决策。
很多初学者误以为 Agent = “给模型写了很长的系统提示词”。这是认知上的偏差。系统提示词只能改变模型的说话风格和思维倾向,但无法让模型真正调用一个业务系统。Agent 和普通聊天机器人的本质区别,不在于“怎么说”,而在于“能不能做事”。
2.3 Agent 到底适合解决什么问题
适合 Agent 的任务通常有三个特征:多步骤、需要外部信息、存在动态决策。
比如“根据用户描述,自动创建工单并通知相关责任人”,这是多步骤任务;“查一下最近三个月的销售数据并生成分析报告”,这是需要外部信息;“根据用户意图决定调用哪个业务接口”,这是动态决策。
反过来说,如果你的业务是标准的“输入-处理-输出”固定流程,比如每天定时跑一个数据同步任务,那完全不需要引入 Agent,传统程序更快、更稳定、更好排查。Agent 的价值在于处理那些无法用规则穷尽、需要模型逐步判断的场景。
这一章的小结论是:Agent 不是万能银弹,它是把大模型的“语言理解”转化为“业务行动”的编排范式。学习 Agent 的关键不是背框架,而是先弄懂 ReAct 循环和数据流。
3. 2026 年的 Agent 技术栈全景与选型判断
学习 Agent 的时候,最容易被信息轰炸。今天有人推 LangGraph,明天有人推 Dify,后天又有人讲自己手写框架。我的建议是:先看清技术栈分成几层,再决定从哪一层入手。
3.1 五层技术栈
第一层是模型层。包括 OpenAI、DeepSeek、Qwen 等云上模型,也包括通过 Ollama 部署的本地开源模型。大多数模型服务商都提供了 OpenAI 兼容的/v1接口,这对开发非常友好,可以让代码和具体供应商解耦。
第二层是Agent 编排层。目前主流方案分为三类:通用框架如 LangChain、LangGraph;可视化平台如 Dify、Coze;自研编排代码。编排层负责实现 ReAct 循环、状态流转、多 Agent 通信。
第三层是工具层。工具可以是简单的函数、REST API、数据库操作,也可以是基于 MCP(Model Context Protocol)的标准化工具接入。企业落地时,把内部系统包装成工具供 Agent 调用是核心工程。
第四层是记忆层。常见实现包括 Redis 保存短期会话、向量数据库保存长期语义记忆、关系型数据库保存结构化业务记忆。
第五层是可观测与评测层。包括 Langfuse、LangSmith 等链路追踪工具,以及自建的评价脚本和数据集。
3.2 选型判断
对于零基础学习者,强烈建议先手写一个最小 Agent 循环,不要一上来就套 LangGraph。原因很简单:手写一遍才能真正理解工具调用的消息格式、循环终止条件、上下文如何累积这些核心细节。等你理解了底层原理,再用框架,会发现框架文档里的每个概念都能和你的代码对应上。
对于企业项目,选型则要看团队情况。如果团队缺少算法背景,想快速上线内部工具,Dify、Coze 这类平台上手最快;如果团队工程能力强,业务链路复杂,需要精细控制状态流转,LangGraph 或自研编排更可控;如果公司已经有很多内部系统,想把它们统一开放给 Agent,工具层设计远比框架选择重要。
从技术趋势看,2026 年的一个明确方向是 Agent 从“单 Agent 演示”走向“多 Agent 协同和工程化落地”。但不管上层平台怎么变,工具调用和记忆管理这些地基能力是通用的。这篇文章后面采用“手写核心循环 + FastAPI 封装”的方式,就是为了让你真正掌握不依赖特定框架的底层能力。
4. 环境准备与最小可运行示例
磨刀不误砍柴工。环境准备环节很简单,但却是很多人放弃的开端。我们统一按下面的方案配置。
4.1 运行环境
- Python 3.10 或更高版本,3.11、3.12 均可;
- pip 包管理工具;
- 一个 OpenAI 兼容的模型 API。如果你使用的是 DeepSeek、Qwen、Ollama 本地模型,只要确认它们提供了 OpenAI 兼容接口即可;
- API Key,通过环境变量注入,不要写死在代码里。
安装依赖:
pip install openai python-dotenv fastapi uvicornopenai是官方 SDK,用于调用模型;python-dotenv用于从.env文件读取环境变量;fastapi和uvicorn用于后续的企业级部署。
4.2 验证模型连通性
在项目目录下创建.env文件:
# 请替换为你自己的 API Key OPENAI_API_KEY=sk-xxxx OPENAI_BASE_URL=https://api.openai.com/v1 AGENT_MODEL=gpt-4o-mini注意,OPENAI_BASE_URL是兼容层设计的关键。如果使用其他提供 OpenAI 兼容接口的服务商,替换成对方提供的 base_url 即可。比如本地 Ollama 的默认地址可以是http://localhost:11434/v1。
接下来写一个最小调用脚本check_api.py:
import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() client = OpenAI( api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"), ) def chat_once(user_content: str) -> str: response = client.chat.completions.create( model=os.getenv("AGENT_MODEL", "gpt-4o-mini"), messages=[{"role": "user", "content": user_content}], ) return response.choices[0].message.content if __name__ == "__main__": print(chat_once("你好,请用一句话说明什么是 AI Agent。"))这段代码如果正常运行,说明你的 API Key、网络环境和 SDK 都是通的。这就是 Agent 开发的地基,后续所有复杂逻辑都建立在chat.completions.create这个调用之上。
回到前面的判断:这段代码只能“聊天”,不能“干活”。要让程序具备 Agent 能力,必须进入下一步:让模型返回结构化的工具调用指令,然后由你编写的代码真正执行工具。这是 Agent 开发教程里最核心的分水岭。
5. Function Calling 与工具调用全流程
Function Calling,中文常称为“函数调用”或“工具调用”。它是 Agent 能够“行动”的关键机制。大模型在训练阶段学习了大量工具使用的知识,当你在请求中声明可用工具后,模型会根据用户意图,返回一个结构化的调用意图,而不是直接执行工具。
5.1 工具声明的 JSON Schema
在 OpenAI 兼容接口中,工具声明遵循 JSON Schema 规范。一个工具通常包含名称、描述、参数等字段。描述非常重要,因为模型依靠描述来判断“什么时候该用这个工具”,描述越清晰,调度准确率越高。
示例:
{ "type": "function", "function": { "name": "get_weather", "description": "查询指定城市的当前天气", "parameters": { "type": "object", "properties": { "city": { "type": "string", "description": "城市名,例如北京、上海" } }, "required": ["city"] } } }工具调用的完整流程如下:
- 客户端构造消息列表,并携带
tools参数发送给模型; - 模型返回一个
tool_calls字段,内部包含函数名和参数 JSON 字符串; - 客户端根据函数名找到本地实现,真正执行函数;
- 客户端把执行结果以
role=tool的消息追加到对话中,再次发送给模型; - 模型看到工具结果后,继续推理,直到不再要求调用工具。
5.2 一个通用 Agent 循环模板
下面的agent_loop函数是手写 Agent 循环的通用骨架。它不绑定任何特定业务,只负责完成“模型决策 -> 工具执行 -> 结果回填 -> 再决策”的闭环。
import json import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() client = OpenAI( api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"), ) def get_weather(city: str) -> str: # 真实项目中请替换为天气服务 API table = {"北京": "晴,25度", "上海": "小雨,22度"} return table.get(city, f"暂未收录 {city} 的天气数据") WEATHER_TOOLS = [ { "type": "function", "function": { "name": "get_weather", "description": "查询指定城市的当前天气", "parameters": { "type": "object", "properties": { "city": {"type": "string", "description": "城市名,例如北京"} }, "required": ["city"], }, }, } ] TOOL_MAP = { "get_weather": get_weather, } def execute_tool(name: str, arguments: dict) -> str: func = TOOL_MAP.get(name) if not func: return json.dumps({"error": f"未知工具: {name}"}, ensure_ascii=False) try: result = func(**arguments) return result if isinstance(result, str) else json.dumps(result, ensure_ascii=False) except Exception as e: return json.dumps({"error": str(e)}, ensure_ascii=False) def agent_loop(user_message: str, tools: list, max_steps: int = 5) -> str: messages = [{"role": "user", "content": user_message}] for _ in range(max_steps): response = client.chat.completions.create( model=os.getenv("AGENT_MODEL", "gpt-4o-mini"), messages=messages, tools=tools, ) assistant_msg = response.choices[0].message messages.append(assistant_msg) # 模型没有要求调用工具,说明已经得到最终答案 if not assistant_msg.tool_calls: return assistant_msg.content for tool_call in assistant_msg.tool_calls: name = tool_call.function.name args = json.loads(tool_call.function.arguments or "{}") result = execute_tool(name, args) messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": result, }) return "已达到最大执行步数,任务未完成" if __name__ == "__main__": print(agent_loop("北京天气怎么样?", WEATHER_TOOLS))这段代码想说明三件事。第一,工具执行必须放在客户端,模型只负责生成调用意图,不负责真正执行,这是安全边界。第二,每一次工具返回结果后,都要重新把完整消息列表发给模型,模型才能基于新信息继续推理。第三,必须设置max_steps上限,否则某些复杂场景下模型可能陷入无限调用循环。
这里真正容易踩坑的地方是:有人把工具执行结果直接打印在控制台,却没有以role=tool的消息回传给模型,结果模型完全不知道工具执行了什么,只能瞎猜。Agent 循环中,工具结果回填是消息协议的一部分,不是可选项。
5.3 工具设计的三个原则
工具设计直接决定 Agent 上线的成功率,我总结出三个原则。
原则一:原子化。工具函数尽量只做一件事。比如“获取天气”和“获取城市列表”应该拆成两个工具,而不是做成一个带多个分支开关的大函数。模型调度更准确,代码也更好维护。
原则二:描述即文档。工具的 description 要写清楚“什么时候调用、参数含义、返回什么”。很多团队忽略描述,导致模型在七八个工具中选错。
原则三:容错返回。工具内部异常不要直接抛到上层,而是把错误信息格式化成 JSON 字符串返回给模型。模型看到错误结果后,可能自动调整参数重试,或者向用户解释失败原因。这在企业场景中非常实用。
6. 记忆管理:让 Agent 从无状态到有状态
前面第 5 章的agent_loop有一个明显缺陷:每次调用都是全新的消息列表,Agent 不记得上一次用户说过什么。在单个任务里这没问题,但真实业务场景是持续的对话交互。
6.1 短期记忆与上下文窗口
短期记忆就是当前会话中的消息数组。随着对话变长,会出现一个绕不开的问题:上下文窗口有限。消息越来越多,最终会超出模型最大 token 限制,同时也会增加延迟和成本。
解决上下文膨胀,常见策略有三种:
- 裁剪:只保留最近 N 条消息,最古老的消息直接丢弃;
- 摘要:用模型把较早的对话压缩成摘要,然后替换原来的长消息;
- 关键信息抽取:从旧对话中抽取出用户偏好、任务状态等关键字段,以结构化形式保留。
6.2 手写一个简单记忆管理类
下面这个SimpleMemory类实现了最基础的“裁剪型短期记忆”。它保存用户和助手的消息,超过阈值后自动丢弃最早的对话,你也可以在此基础上扩展摘要逻辑。
class SimpleMemory: def __init__(self, max_history: int = 10): self.history = [] self.max_history = max_history def add_user(self, content: str): self.history.append({"role": "user", "content": content}) self._trim() def add_assistant(self, content: str): self.history.append({"role": "assistant", "content": content}) self._trim() def _trim(self): if len(self.history) > self.max_history: self.history = self.history[-self.max_history:] def get_messages(self, system_prompt: str = None) -> list: messages = [] if system_prompt: messages.append({"role": "system", "content": system_prompt}) messages.extend(self.history) return messages使用方式:
memory = SimpleMemory(max_history=5) memory.add_user("帮我创建一个网络故障工单") memory.add_assistant("好的,工单已创建,编号 TKT1001。") print(memory.get_messages("你是企业工单助手。"))这个类的思想是:把消息历史放在 Agent 对象之外,由上层负责持久化。做企业级项目时,history不能用内存里的 Python 列表,而要放到 Redis 或数据库中,并且按session_id隔离,这样多个用户并发访问时不会互相串数据。
6.3 长期记忆与业务记忆
短期记忆解决“本轮对话别忘”,长期记忆解决“跨会话也能记住”。
长期记忆的实现没有统一标准,常见做法是把用户和 Agent 交互中产生的关键信息写入数据库。比如在电商场景中,用户偏好“只看 500 元以下的商品”,这个信息可以用结构化的user_preference表存储;在客服场景中,用户正在处理的工单编号、当前状态也可以存成业务记忆字段。
向量数据库则更适合存储非结构化语义记忆。比如用户过去问过的问题、知识库文档的切片向量。检索时先做语义相似度匹配,把最相关的内容作为上下文注入 Agent。严格来说,这就是 RAG,但它也可以被视为 Agent 长期记忆的一种实现方式。
这里的工程判断是:能结构化存储的信息,优先结构化存储;无法结构化、需要语义检索的,才上向量库。不要为了技术而技术,一上来就搭一堆向量库,最后维护成本远大于收益。
7. 企业级项目实战:工单处理 Agent 完整实现
前面几章的知识点,现在综合起来做一个真实感很强的项目:一个企业内部的工单处理 Agent。它接收用户自然语言描述,能创建工单、查询工单状态、修改工单状态,同时具备多轮对话记忆,最后通过 FastAPI 暴露成 HTTP 服务。
7.1 场景与架构设计
业务需求如下:
- 用户说“网络断了,帮我提交工单”,Agent 自动创建工单;
- 用户说“查一下 TKT1001 什么状态”,Agent 查询工单状态;
- 用户说“把 TKT1001 改成处理中”,Agent 更新工单状态;
- 用户多次询问时,Agent 能记住之前创建过哪个工单。
架构分四层:
- HTTP 层:FastAPI 接收请求,解析
session_id和message; - 会话层:维护每个会话的
SimpleMemory实例; - Agent 核心:ReAct 循环,调度模型和工具;
- 工具层:工单相关的三个工具函数,业务数据用内存字典模拟。
这个设计在企业环境中的对应关系是:内存字典换成真实数据库,SimpleMemory 换成 Redis 会话存储,FastAPI 服务后面再挂鉴权、限流、网关。
7.2 工具定义与实现
创建tools.py:
import json TICKET_DB = {} _ticket_seq = 1000 def create_ticket(topic: str, description: str) -> str: global _ticket_seq _ticket_seq += 1 ticket_id = f"TKT{_ticket_seq}" TICKET_DB[ticket_id] = { "ticket_id": ticket_id, "topic": topic, "description": description, "status": "open", } return json.dumps(TICKET_DB[ticket_id], ensure_ascii=False) def get_ticket_status(ticket_id: str) -> str: ticket = TICKET_DB.get(ticket_id) if not ticket: return json.dumps({"error": f"工单 {ticket_id} 不存在"}, ensure_ascii=False) return json.dumps(ticket, ensure_ascii=False) def update_ticket_status(ticket_id: str, status: str) -> str: if ticket_id not in TICKET_DB: return json.dumps({"error": f"工单 {ticket_id} 不存在"}, ensure_ascii=False) if status not in ("open", "processing", "resolved", "closed"): return json.dumps({"error": "非法状态"}, ensure_ascii=False) TICKET_DB[ticket_id]["status"] = status return json.dumps(TICKET_DB[ticket_id], ensure_ascii=False) TOOLS = [ { "type": "function", "function": { "name": "create_ticket", "description": "创建一条新的工单", "parameters": { "type": "object", "properties": { "topic": {"type": "string", "description": "工单主题,例如网络故障"}, "description": {"type": "string", "description": "问题详细描述"}, }, "required": ["topic", "description"], }, }, }, { "type": "function", "function": { "name": "get_ticket_status", "description": "根据工单 ID 查询工单当前状态", "parameters": { "type": "object", "properties": { "ticket_id": {"type": "string", "description": "工单 ID,例如 TKT1001"}, }, "required": ["ticket_id"], }, }, }, { "type": "function", "function": { "name": "update_ticket_status", "description": "更新工单状态,可选值 open、processing、resolved、closed", "parameters": { "type": "object", "properties": { "ticket_id": {"type": "string", "description": "工单 ID"}, "status": {"type": "string", "enum": ["open", "processing", "resolved", "closed"]}, }, "required": ["ticket_id", "status"], }, }, }, ] TOOL_MAP = { "create_ticket": create_ticket, "get_ticket_status": get_ticket_status, "update_ticket_status": update_ticket_status, }这里有几个企业级细节值得注意。第一,每个工具返回值都是 JSON 字符串,这样模型解析结果时格式一致,准确率更高。第二,工具函数内部有参数校验,比如更新状态时检查状态是否合法,工单不存在时返回明确的错误信息。第三,真实团队会把TICKET_DB换成数据库访问层,而不是把 SQL 写在工具函数里。
创建agent.py,完整实现 Agent 核心逻辑:
import json import os from openai import OpenAI from dotenv import load_dotenv from tools import TOOLS, TOOL_MAP from memory import SimpleMemory load_dotenv() client = OpenAI( api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"), ) SYSTEM_PROMPT = ( "你是一个企业工单助手。用户会提交工单、查询进度或修改状态。" "需要调用工具时就调用工具,工具返回后基于结果继续回答。" "回复要简洁、准确,不要编造工单信息。" ) MAX_STEPS = 5 def execute_tool(name: str, arguments: dict) -> str: func = TOOL_MAP.get(name) if not func: return json.dumps({"error": f"未知工具: {name}"}, ensure_ascii=False) try: result = func(**arguments) return result if isinstance(result, str) else json.dumps(result, ensure_ascii=False) except Exception as e: return json.dumps({"error": str(e)}, ensure_ascii=False) def run_agent(user_input: str, memory: SimpleMemory) -> str: memory.add_user(user_input) messages = memory.get_messages(SYSTEM_PROMPT) for _ in range(MAX_STEPS): response = client.chat.completions.create( model=os.getenv("AGENT_MODEL", "gpt-4o-mini"), messages=messages, tools=TOOLS, ) assistant_msg = response.choices[0].message messages.append(assistant_msg) if not assistant_msg.tool_calls: reply = assistant_msg.content or "处理完成。" memory.add_assistant(reply) return reply for tool_call in assistant_msg.tool_calls: name = tool_call.function.name args = json.loads(tool_call.function.arguments or "{}") tool_result = execute_tool(name, args) messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": tool_result, }) fallback = "处理步骤较多,暂时无法完成,请尝试重新描述或联系人工客服。" memory.add_assistant(fallback) return fallbackrun_agent和前面的通用循环结构完全一致,区别在于:第一,它使用了SimpleMemory持久化用户和助手的最终消息;第二,系统提示词针对工单业务做了定制;第三,中间的工具调用过程保持在局部变量messages中,不污染用户可见记忆,这是企业级体验设计里很值得注意的一点。用户不应该看到“模型思考过程”,只需要看到最终结果。
创建memory.py,直接复用第 6 章的SimpleMemory类。创建main.py,提供 FastAPI 接口:
from fastapi import FastAPI, HTTPException from pydantic import BaseModel from agent import run_agent from memory import SimpleMemory app = FastAPI(title="工单处理 Agent API") # 演示用内存存储,生产环境请替换为 Redis 等外部存储 _session_memory = {} class ChatBody(BaseModel): session_id: str message: str @app.post("/chat") def chat(body: ChatBody): if body.session_id not in _session_memory: _session_memory[body.session_id] = SimpleMemory() memory = _session_memory[body.session_id] try: reply = run_agent(body.message, memory) return {"session_id": body.session_id, "reply": reply} except Exception as e: raise HTTPException(status_code=500, detail=str(e))启动服务:
uvicorn main:app --host 0.0.0.0 --port 80007.3 接口验证方法
服务启动后,可以用curl验证:
curl -X POST http://127.0.0.1:8000/chat \ -H "Content-Type: application/json" \ -d '{"session_id": "user-001", "message": "我的网络连接不上,帮我创建一个网络故障工单"}'预期输出中应包含类似TKT1001的工单编号。再发第二条消息:
curl -X POST http://127.0.0.1:8000/chat \ -H "Content-Type: application/json" \ -d '{"session_id": "user-001", "message": "刚刚创建的工单是什么状态?"}'这一步验证了记忆能力:Agent 必须记得当前会话用户刚刚创建的工单编号,而不是让用户重新提供。如果 Agent 回答“您还没有创建工单”,说明记忆或工具调度有问题,需要检查消息列表是否正确累积。
这个企业级实例虽然业务简单,但已经覆盖了 Agent 工程的完整链路:对话接入、工具调度、记忆管理、HTTP 服务、会话隔离。后续把_session_memory换成 Redis,把TICKET_DB换成真实数据库,就是一个可以直接内测的 MVP。
8. 评测、可观测性与生产部署
企业级 Agent 和 Demo 的最大区别,在于是否具备可评测、可观测、可部署的工程体系。很多人模型调用写得顺手,一谈上线就头疼,因为 Agent 的输出不再是一个稳定的函数返回值。
8.1 Agent 评测的三种方法
第一种是 Golden Set 评测。准备一组输入-期望输出的数据集,跑完后规则比对。对工单场景,可以检查输出是否包含工单号、状态字段是否正确。这种方法成本低、适合回归测试。
下面是一个简单的评测脚本思路:
# eval_agent.py import re from agent import run_agent from memory import SimpleMemory def main(): memory = SimpleMemory() first = run_agent("帮我创建一个邮件系统无法登录的工单", memory) print("第一步输出:", first) ticket_ids = re.findall(r"TKT\d+", first) if not ticket_ids: print("FAIL: 未生成工单编号") return ticket_id = ticket_ids[0] second = run_agent(f"请查询 {ticket_id} 的最新状态", memory) print("第二步输出:", second) if "open" in second.lower(): print("PASS: 工单状态正确") else: print("WARN: 状态可能不是 open,需要人工确认") if __name__ == "__main__": main()第二种是 LLM-as-Judge 评测。让另一个更强的模型当评委,输入用户问题、Agent 完整轨迹和参考答案,让评委模型打分。它适合评估“回答是否自然”“工具选得是否合理”这类开放性问题。缺点是会引入额外的模型成本和不确定性,需要设计明确的评分标准。
第三种是人工评测。上线前抽取 100 条典型用户问题,让业务人员和研发一起打分。成本最高,但最能反映真实体验。实际项目通常是三种方法结合:自动化回归跑规则、大模型打分跑批量、人工抽评测保底。
8.2 链路可观测性
Agent 出问题时的排查难度,远高于普通接口。普通接口无非是入参、返回值、异常栈;Agent 则有“模型思考、工具调用、结果回填、二次推理”等多个环节,任何一个环节出错都会导致最终结果异常。
生产环境推荐接入 Langfuse 或 LangSmith 这类 Agent 可观测平台,或者至少自建日志表,记录以下关键信息:
- 每次请求的
session_id - 模型参数和提示词版本
- 模型返回的
tool_calls完整内容 - 工具执行结果
- 每一步的 token 消耗和耗时
- 最终回答文本
没有这些数据,Agent 上线后一旦效果变差,你连问题出在模型还是工具还是提示词都无法判断。
8.3 Docker 部署示例
FastAPI 服务化后,Docker 部署是主流选择。在项目根目录创建Dockerfile:
FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8000 CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]requirements.txt内容:
openai python-dotenv fastapi uvicorn构建并启动:
docker build -t ticket-agent . docker run -p 8000:8000 --env-file .env ticket-agent生产部署时,API Key 不要打包进镜像,要通过环境变量注入。另外,建议在 FastAPI 外层加一层网关,统一处理鉴权、限流、审计日志。Agent 接口涉及调用外部模型,单用户高频调用会带来显著成本,必须做按用户限流。
9. 常见问题与排查思路
Agent 开发过程中,我见过最多的问题集中在下面几类。这里列成表格,方便收藏备用。
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 模型从不调用工具,只返回一段文字建议 | 工具描述不清晰,或模型不具备工具调用能力 | 在 API 请求中打印返回的tool_calls字段,确认模型是否返回空值 | 优化工具名称和 description;换用支持 Function Calling 的模型;确认tools参数已正确传入 |
| 模型调用工具但参数错误 | 工具参数描述不明确,多工具之间参数歧义 | 查看tool_call.function.arguments的 JSON 内容,确认模型传了哪些参数 | 参数名使用业务通用说法;枚举值用enum约束;在工具函数内部增加参数校验 |
| 工具执行成功,但模型回答与工具结果不一致 | 工具结果没有正确回填,或回填的消息格式不对 | 检查消息列表中是否存在role=tool消息,且tool_call_id和 assistant 返回的tool_call.id一致 | 严格按 OpenAI 兼容协议回填工具消息;打印完整 messages 对照检查 |
| 对话轮次稍多就报上下文超限 | 历史消息无裁剪 | 查看请求的 messages 总 token 数 | 使用 SimpleMemory 裁剪;对旧消息做摘要;只保留最近 N 轮 |
| Agent 无限循环停不下来 | 缺少最大步数限制 | 检查循环是否设置了max_steps | 为所有 Agent 循环增加步数上限,达到上限后转人工或返回明确错误 |
| 多用户同时使用,会话串线 | 记忆使用了全局变量,没有按 session_id 隔离 | 检查_session_memory的键设计 | 用 Redis Hash 或数据库按 session_id 存储会话;生产禁止用进程内全局字典 |
| 上线后效果变差 | 模型版本变动、提示词被修改、工具接口变更 | 对比 Langfuse 中前后链路 trace 差异 | 固定模型版本;提示词纳入版本管理;工具接口变更前后都要跑回归评测 |
| 成本飙升 | 每轮请求历史消息过长,或模型频繁调用不必要工具 | 查看可观测平台中每次请求的 token 统计 | 缩短历史;优化系统提示词减少多余调用;为工具调用加条件约束 |
这里最容易被忽略的其实是“工具结果回填格式”。很多初学者会漏掉tool_call_id,或者把工具结果写成role=user,模型无法正确关联工具调用和工具结果,于是出现“工具明明执行了,模型却说没看到结果”的诡异问题。遇到这类问题,不要猜,直接把最终发出去的 messages 列表完整打印出来,人工检查一遍消息格式是否符合协议。
10. 最佳实践与工程建议
基于前面完整的开发流程,我整理了一份可以直接用于团队评审的工程建议清单。
第一,提示词版本化。企业级 Agent 的提示词需要像代码一样管理。建议把系统提示词抽成单独配置文件,每次修改记录版本号,并在可观测平台中关联 trace,这样效果变差时能快速回滚。
第二,工具权限最小化。Agent 能调用的工具集合要遵循最小权限原则。客服机器人不需要有删除工单的权限,数据分析 Agent 不需要有写入权限。工具函数的失败处理也要友好,把错误信息格式化成模型能理解的 JSON,而不是直接抛 500 异常。
第三,记忆要分层。短期记忆放 Redis,业务记忆放业务库,向量记忆放向量库。不要把三类记忆混在一个列表里,否则上下文越来越乱,费用越来越高。
第四,成本控制要前置。模型调用是 Agent 的主要成本来源。优化手段包括:优先使用成本更低的模型处理简单轮次;复用结果缓存;历史消息做摘要而不是无限追加;设置单用户调用频控。
第五,安全边界要清晰。Agent 涉及调用外部模型,用户输入不能直接拼接进系统提示词而不做任何过滤。使用第三方模型时,涉及敏感业务数据的场景要谨慎评估数据出境和隐私合规问题。企业内部敏感数据优先考虑私有化部署模型,或在安全的区域内调用模型服务。
第六,先跑通再上框架。如果你正在选型,我的建议是先用手写的agent_loop把业务逻辑跑通,确认价值和瓶颈。当业务链路变得复杂,需要多人协作、分支状态流转、多 Agent 协同的时候,再引入 LangGraph 或 Dify 这类工具。框架的价值在于抽象和协作,不在于炫技。
第七,做一些“防呆”设计。Agent 推理有随机性,生产系统必须假定它可能出错。因此关键操作前要加确认机制,例如用户明确说“把工单状态改成已解决”,Agent 才能执行更新,中间不能让模型自由发挥。涉及扣款、删除、权限变更等高风险动作,必须走人工确认流程。
11. 总结与后续学习方向
这篇 AI Agent 开发教程的核心内容可以浓缩成几句话:Agent 不是花哨的概念,而是“模型 + 工具 + 记忆 + 循环”的组合;开发 Agent 的重点不在提示词,而在工程设计;从零基础到企业级项目实战,要经过模型调用、工具调用、记忆管理、评测部署四个阶段,每一步都有明确的代码实践。
建议你按照文章顺序自己动手跑一遍工单处理 Agent,然后尝试替换成你自己的业务场景。比如把“天气查询”换成“订单查询”,把“工单状态”换成“库存查询”,代码骨架完全不用变。这一步能帮你真正把 Agent 开发思路内化成自己的工程能力。
后续可以继续深入的方向包括:LangGraph 状态机编排、多 Agent 协作、RAG 与 Agent 的深度融合、基于 MCP 的标准工具接入、Agent 自动化评测平台建设,以及 Java 生态下基于 Spring Boot 的 Agent 客户端设计。无论你选择哪个方向,都建议先掌握本文中的底层循环原理,再扩展学习,这样学框架时你会更容易理解它的设计动机和适用边界。收藏本文,按章节逐步实践,应该能帮你少踩不少坑。