先说个我观察到的现象。这两年代码圈里“AI智能体”和“Agent”这两个词快被用烂了,不少团队把大模型API封装一层函数,接个工具调用,就敢叫自己做了个Agent。可真放上生产环境,跑几天就露馅:遇到稍微复杂点的任务就卡壳,该调工具的时候不调,上下文越攒越乱,更别提多个Agent协作时互相甩锅。问题不在模型不行,而在于很多人的理解还停留在“调模型”的层面,压根没搞明白Agent的本质是一个自主决策循环。
这篇文章我想把AI智能体实战这件事从头到尾捋一遍。不绕弯子,不背概念,直接从“Agent和普通API调用差在哪”讲起,先带你手写一个最小可运行的Agent,再引入LangGraph做状态管理和记忆,然后聊多Agent协作,最后把部署、调优、避坑这些生产环境才见真章的内容一并倒出来。不管你是刚入门想搞懂原理的开发者,还是已经在做Agent应用、想提升稳定性的工程师,这篇都值得你花十几分钟过一遍。
1. 先弄清Agent到底在机制上和API调用差在哪
1.1 大多数项目挂羊头卖狗肉,问题出在哪
我见过太多所谓的“Agent项目”,代码长这样:用户输入一句话 → 拼进Prompt → 调一次LLM → 把结果返回。整个链路连个循环都没有,最多加了个上下文记忆,就敢对外宣称“AI智能体”。这种实现本质上还是一个高级聊天机器人,它不是Agent。
为什么这么说?因为Agent的灵魂在于行动。它不仅能“想”,还能“做”——调用外部工具、查询数据库、操作API、读文件写文件,然后根据执行结果继续判断下一步怎么做。这是一个持续运转的环路,而不是一次性的问与答。你让一个普通聊天机器人去“帮我查一下上海今天的天气,然后对比北京,告诉我哪个适合出行”,它只能瞎编。Agent则会先去调天气API拿到真实数据,再基于数据做对比分析,最后给出结论。
这两者的差距,就是“Demo”和“能用的产品”之间的差距。
1.2 Agent的核心决策循环:感知-规划-执行-观察
业界对Agent的拆解有很多说法,我自己实战下来最认同的模型就四个环节:
- 感知:接收用户的目标,同时从记忆中检索相关信息,构成当前任务的上下文。
- 规划:基于大模型的推理能力,把大目标拆成小步骤,决定下一步要调什么工具、传入什么参数。
- 执行:调用具体的工具,比如天气API、计算器、数据库查询、内部业务接口。
- 观察:拿回工具执行的结果,把它作为新的信息注入上下文,回到“规划”环节重新判断——任务完成了吗?还是需要继续下一步?
这四步循环往复,直到模型认为目标已经达成,输出最终答案。这就是业界常说的ReAct模式(Reasoning + Acting),也是目前绝大多数生产级Agent框架的底层逻辑。
1.3 一个最简循环的可执行伪代码
在引框架之前,我先把这层循环用最朴素的伪代码写出来,让你对内部逻辑有个具体的印象:
def run_agent(user_query): messages = [system_prompt, user_query] for step in range(MAX_STEPS): # 最多循环10次,防止失控 response = llm.chat(messages) # 让模型决定下一步 if response.action == "finish": return response.answer # 模型认为任务已完成 if response.action == "call_tool": result = execute_tool(response.tool_name, response.arguments) messages.append("工具返回:" + result) # 观察结果 continue # 回到规划环节 return "达到最大循环次数,提前终止"就这么点东西,却是所有Agent产品的内核。架构可以千变万化,但循环是跑不掉的。理解了这一点,你再去用任何框架都会觉得轻车熟路。
2. 主流Agent框架横评:LangGraph、CrewAI、AutoGen怎么选
2.1 为什么先看框架而不是直接写代码
第一章节写的那个最小循环,自己写完你就会发现:真要做成产品,光有个循环远远不够。你需要状态管理、多轮记忆、工具调用的重试机制、并发的多Agent协作、日志追踪……这些基础设施全部从零手写,没个几周下不来。所以生产级项目,我强烈建议站在框架的肩膀上。
但市面上的Agent框架五花八门,选错框架比不选框架更痛苦——代码写了一半发现控制力不够,换框架等于重构。
2.2 三个框架实测对比
我先后用LangGraph、CrewAI、AutoGen做了几个不同类型的项目,说实话各有各的脾气,直接上对比表:
| 框架 | 核心设计理念 | 适用场景 | 上手难度 | 生产可用性 |
|---|---|---|---|---|
| LangGraph | 可视化状态图,节点+边,流程可控性极强 | 复杂业务流、需要精细控制的场景 | 中 | 高,状态持久化完善 |
| CrewAI | 角色化协作,定义Agent角色与任务,开箱即用 | 多Agent分工明确的流水线业务 | 低 | 中,灵活度略低 |
| AutoGen | 多Agent对话驱动,自动编排会话 | 研究探索、对话式协作场景 | 中 | 中,消息流控制有难度 |
你可能会问,那些很火的LangChain不也是Agent框架吗?LangChain更像一个工具箱,提供了模型封装、Prompt管理、工具调用等基础能力,但Agent编排它不是专长。LangGraph是LangChain团队后面专门为Agent编排设计的图结构框架,要玩Agent,我更推荐直接上LangGraph。
2.3 按业务类型选型的建议
选型这事没有银弹,按需求对号入座:
- 如果你的业务是固定流程、若干步骤串联,比如“先检索资料、再生成文案、最后人工审核”,CrewAI是非常快的选择,代码量少,团队协作也自然。
- 如果你的业务有分支、条件、循环、人工介入点,比如客服系统里要根据用户情绪走不同话术、失败要重试、特定情况要转人工,LangGraph是唯一靠谱的选择。
- 如果你是搞研究、做实验,想探索多Agent的涌现行为,AutoGen有意思,但生产环境慎用。
我自己目前的默认选择是LangGraph,不是因为CrewAI不好,而是因为我在生产上更在意“流程可见、异常可控”,LangGraph的状态图模型天生适合这件事。
3. 不依赖重型框架,用工具调用手写一个最小Agent
在你被框架牵着鼻子走之前,我强烈建议你手动实现一遍Agent循环。这一步能帮你彻底摆脱“框架黑盒焦虑”,后面排障时你会感谢自己写过这段代码。
3.1 环境准备与模型接入
我做实战时用的模型是DeepSeek的deepseek-chat,因为它走OpenAI兼容接口,成本低、中文能力扎实,用来学Agent完全够。你也可以换成通义、智谱或者OpenAI,代码基本不用改。
用Python的话,创建虚拟环境并安装SDK:
mkdir minimal-agent cd minimal-agent python3 -m venv venv source venv/bin/activate pip install openai python-dotenv然后把API Key放到项目根目录的.env文件里,方便管理:
DEEPSEEK_API_KEY=sk-你的密钥接下来封装一个模型客户端,这里要注意的是,DeepSeek的接口地址和OpenAI不同,一定要指定base_url:
import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() client = OpenAI( api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com/v1" ) def chat(messages, tools=None): response = client.chat.completions.create( model="deepseek-chat", messages=messages, tools=tools, temperature=0.3 ) return response.choices[0].message3.2 定义工具描述:让Agent“看得见”能做什么
模型并不知道你给它准备了什么工具,你必须用一套结构化的方式把工具的能力描述给它。这就是OpenAI的Function Calling机制——把你每个工具的名称、功能描述、参数结构告诉模型,模型在需要时会返回一个结构化的“调用请求”,而不是自由文本。
我给你写两个实用工具,一个查天气,一个做计算:
def get_weather(city: str) -> str: """模拟天气查询,实际项目里可以改成真实天气API""" weather_map = { "北京": "晴,22°C,微风", "上海": "多云,25°C,东南风3级", "广州": "小雨,28°C,湿气较重" } return weather_map.get(city, f"{city}暂无天气数据") def calculator(expression: str) -> str: """简单计算器,仅支持四则运算""" try: # 白名单过滤,只允许数字和运算符,防止任意代码执行 safe = all(c in "0123456789+-*/(). " for c in expression) if not safe: return "包含非法字符" return str(eval(expression)) except Exception as e: return f"计算错误: {e}"对应的工具描述定义成这样:
tools = [ { "type": "function", "function": { "name": "get_weather", "description": "查询指定城市的实时天气情况", "parameters": { "type": "object", "properties": { "city": {"type": "string", "description": "城市名称,例如:北京"} }, "required": ["city"] } } }, { "type": "function", "function": { "name": "calculator", "description": "执行数学四则运算,例如:1+2*3", "parameters": { "type": "object", "properties": { "expression": {"type": "string", "description": "数学表达式"} }, "required": ["expression"] } } } ]这里有个实操要点:工具的description写得越具体,模型选错的概率越低。如果你只写“计算器”,模型可能把“比较两个数字大小”这种任务也丢给你这个工具,结果就翻车了。
3.3 实现循环主控:调用模型、解析工具调用、执行、回填
现在到了最核心的部分。要维护一个消息列表,循环地调模型;模型返回工具调用请求时,执行工具并把结果作为一条新消息回填;模型返回最终答案时,结束循环。
def run_agent(user_query: str, max_steps: int = 5): messages = [{"role": "system", "content": "你是一个能调用工具的智能助手,请务必使用工具获取真实数据。"}] messages.append({"role": "user", "content": user_query}) for step in range(max_steps): print(f"\n===== Step {step + 1} =====") ai_message = chat(messages, tools=tools) # 情况1:模型没有请求调用工具,直接给出最终答案 if not ai_message.tool_calls: print("Assistant:", ai_message.content) return ai_message.content # 情况2:模型请求调用工具 messages.append(ai_message) # 把带tool_calls的消息记入历史 for tool_call in ai_message.tool_calls: fn_name = tool_call.function.name args = eval(tool_call.function.arguments) # 解析参数JSON print(f"调用工具: {fn_name}({tool_call.function.arguments})") # 工具分发 if fn_name == "get_weather": result = get_weather(args["city"]) elif fn_name == "calculator": result = calculator(args["expression"]) else: result = f"未知工具: {fn_name}" # 把工具执行结果返回给模型 messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": result }) print(f"工具返回: {result}") print("达到最大步数,强制停止") return None if __name__ == "__main__": run_agent("北京和上海今天天气怎么样?哪个适合出门跑步?顺便算一下25*4+10等于多少。")3.4 跑通一个带工具调用的完整任务,效果如何
跑一下上面那个混合任务,模型会经历这样的过程:
- 第一轮:模型认为需要先查北京和上海的天气,发出两个
get_weather工具调用请求。 - 执行工具,返回天气数据。
- 第二轮:模型觉得还得算一下25*4+10,调用
calculator。 - 第三轮:模型拿到所有结果,整合成答案:“北京晴22°C适合跑步,上海多云25°C也可以但东南风3级略有影响,25*4+10=110。”
整个过程模型没有瞎编天气,没有跳过工具,完全按照工具返回的真实数据作答。这就是Agent最基本也最正确的行为模式。
顺嘴提一句:实际开发时不要用eval解析函数的arguments,更不要直接执行eval(expression),我上面只是为了保持代码精简,生产环境一定要用json.loads解析参数,计算器部分要改成安全的AST解析或者直接调decimal模块做运算。
4. 从Demo到生产:用LangGraph强化状态管理与记忆
手写的循环能帮你理解原理,但到了生产环境,你会发现三个痛点是手写代码很难优雅解决的:状态怎么持久化、并发怎么控制、流程怎么可视化。这就是LangGraph登场的时机。
4.1 为什么要从手写循环迁到状态图
LangGraph的核心思想是把Agent流程定义成一张有向图。每个节点做一件事,比如“大模型决策”是节点A,“执行工具”是节点B,节点A输出决定了下一步去B还是结束。全部状态集中在一个State对象里,跑完每一步你都可以把整个状态存到数据库,下次从断点继续跑。
这个设计对你意味着什么?意味着线上出问题时,你能把某次会话的完整状态导出来复盘,而不是对着一个残缺的对话日志猜来猜去。
以LangGraph 0.2.x版本为例,安装:
pip install langgraph langchain-openai4.2 定义AgentState和节点,搭建最小图
先定义状态结构。我习惯把状态设计成“用户问题+消息历史+中间变量”:
from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END from langgraph.graph.message import add_message class AgentState(TypedDict): query: str messages: Annotated[list, add_message] # 自动追加消息 intermediate_steps: list # 记录中间过程,方便追踪然后定义两个节点。第一个节点是Agent决策节点,负责调LLM决定下一步;第二个节点是工具执行节点:
def agent_node(state: AgentState): # 把系统提示词和消息历史一起发给LLM messages = [{"role": "system", "content": SYSTEM_PROMPT}] for msg in state["messages"]: messages.append(msg) response = chat(messages, tools=tools) return {"messages": [response]} def tool_node(state: AgentState): last_message = state["messages"][-1] results = [] for tool_call in last_message.tool_calls: result = dispatcher(tool_call.function.name, tool_call.function.arguments) results.append({ "role": "tool", "tool_call_id": tool_call.id, "content": result }) return {"messages": results}把节点和边串起来,中间加一个条件判断——模型想要调工具就走工具节点,不想调工具就结束:
def should_continue(state: AgentState): last_message = state["messages"][-1] if last_message.tool_calls: return "continue" return "end" graph = StateGraph(AgentState) graph.add_node("agent", agent_node) graph.add_node("tools", tool_node) graph.add_edge("tools", "agent") # 工具执行完,回到Agent继续决策 graph.add_conditional_edge( "agent", should_continue, {"continue": "tools", "end": END} ) graph.set_entry_point("agent") app = graph.compile()这段代码的优雅之处在于:你肉眼就能看出流程结构,第一轮是agent → 判断 → tools → agent → 判断 → end。哪里断了、哪里循环了,一清二楚。你还可以用graph.get_graph().draw_mermaid_png()把图导出来看,不过这属于开发期便利,生产环境用处不大。
4.3 短期记忆的裁剪与长期记忆的向量检索
Agent跑久了,消息历史会越来越长。我见过最离谱的一次,一个Agent对话到第30轮,光消息历史就占了12000个token,一次请求光历史就烧掉好几毛钱。所以必须对短期记忆做裁剪。
常用的策略是只保留最近N轮消息,以及把过长的工具返回结果做摘要压缩:
def trim_messages(state: AgentState, max_messages: int = 10): messages = state["messages"] if len(messages) > max_messages: # 保留系统提示语,截断最新的max_messages条 state["messages"] = messages[-max_messages:] return state短记忆管的是“当前对话还热乎的内容”,长记忆管的是“跨会话的业务知识”。长记忆的实现思路很固定:把关键信息切成块,用Embedding模型向量化,存入向量数据库,每次任务开始前检索相关内容注入上下文。
LangChain生态里有个MemorySaver可以快速做短期记忆持久化,但长期记忆我更建议直接用Chroma或者Milvus。Chroma轻量,适合中小项目;Milvus重一些,但支撑千万级向量没问题。
import chromadb from chromadb.utils import embedding_functions client = chromadb.Client() collection = client.get_or_create_collection( name="agent_memory", embedding_function=embedding_functions.DefaultEmbeddingFunction() ) # 写入一条长期记忆 collection.add( documents=["用户在使用本系统时偏好简洁的回复风格"], ids=["memory_001"] ) # 任务开始时检索相关记忆 results = collection.query( query_texts=["用户喜欢什么风格的回复?"], n_results=3 )把这些检索结果塞进System Prompt,Agent在每次决策时就带上了“历史经验”,这比单纯堆对话历史要高效得多。
4.4 给Agent加一个工具调用的重试与兜底
生产环境里工具调用失败太常见了:第三方API超时、参数传错、返回格式不符合预期。我踩过最典型的一个坑是:模型生成了合法的工具调用,但工具执行抛异常,如果不处理,整个Agent循环就中断了,用户只看到一行“Error”。
解决办法是在工具执行节点做异常捕获,并把错误信息作为正常的工具返回结果交回给模型,让模型自己决定怎么办。这对模型来说只是“观察到了一个新的结果”,它可能换参数重试,也可能放弃这个工具走别的路径。
def tool_node(state: AgentState): last_message = state["messages"][-1] results = [] for tool_call in last_message.tool_calls: try: result = dispatcher(tool_call.function.name, tool_call.function.arguments) except Exception as e: # 把异常包装成工具返回值,模型可以基于错误信息进行纠错 result = f"工具执行发生错误: {str(e)}" results.append({ "role": "tool", "tool_call_id": tool_call.id, "content": result }) return {"messages": results}别小看这一层兜底,它能把Agent从“遇到异常屏幕一黑”救成“自己试错重来”。我实测下来,加了异常回填之后,工具链路的整体成功率能从70%提升到95%以上。
5. 多Agent协作:把单兵变队伍,但别被通信拖垮
单Agent能解决的问题是有限的。真实业务里,一个Agent又要检索资料又要写文案又要查质量,容易要么上下文爆炸,要么顾此失彼。多Agent协作把职责拆给不同角色,各管一段。
5.1 三种协作拓扑:顺序、并行、层级
我自己归纳多Agent协作的拓扑就三种,够用:
- 顺序流水线:A的输出喂给B,B的输出喂给C。适合流程固定的任务,比如“研究员Agent查资料 → 写手Agent写文章 → 审核Agent检查质量”。
- 并行分解:一个大任务拆成若干子任务,多个Agent同时处理,最后汇总。适合数据收集类任务,比如让三个Agent同时调研三个竞品。
- 层级编排:一个主管Agent负责任务分发、结果汇总,下面挂多个执行Agent。这种模式控制力最强,适合复杂场景,但主管Agent很容易成为瓶颈,要注意它的上下文长度。
5.2 用CrewAI快速搭建一个流水线型多Agent
CrewAI把多Agent协作写成了“配置式”代码,角色、目标、任务、协作模式一目了然。先安装:
pip install crewai然后定义一个“研究-写作”流水线:
from crewai import Agent, Task, Crew, Process researcher = Agent( role="行业研究员", goal="收集和整理指定主题的最新信息", backstory="你是一名资深行业分析师,擅长从公开信息中提炼要点", llm="deepseek/deepseek-chat" # 可指定底层LLM ) writer = Agent( role="内容作者", goal="基于研究员提供的信息撰写结构化文章", backstory="你是一名技术博主,文字简洁有干货", llm="deepseek/deepseek-chat" ) research_task = Task( description="调研AI智能体在客服领域的实际落地案例,整理3个要点", agent=researcher, expected_output="3个案例简述,每个不超过200字" ) writing_task = Task( description="基于调研结果撰写出800字左右的科普文章", agent=writer, expected_output="一篇完整的Markdown格式文章" ) crew = Crew( agents=[researcher, writer], tasks=[research_task, writing_task], process=Process.sequential # 顺序执行 ) result = crew.kickoff() print(result)CrewAI的文档废话不多,这个配置十几行就能跑通,非常适合快速验证多Agent协作的可行性。
5.3 多Agent最容易翻车的三个细节
第一个坑是子Agent的输出不稳定。同一个任务,上游Agent今天输出三段式结构,明天输出大长段,下游Agent解析就会出乱。我的做法是要求每个Agent的expected_output里明确指定输出格式,比如“必须用JSON返回,字段为title和content”。最好再让下游Agent不要依赖上游的格式,而是直接读取结构化字段。
第二个坑是Token消耗比单Agent多一个量级。每个Agent都要持有完整上下文,三个Agent跑一个任务,Token消耗可能是一个Agent的三到五倍。成本敏感的场景,要给每个Agent设max_tokens上限,并尽量精简System Prompt。
第三个坑是错误传播被放大。单Agent错一步还能自愈,多Agent流水线里上游一步错,下游连错三环。我现在的做法是在每个任务之间加一个“校验节点”——不用LLM,用简单的规则检查结果是否符合预期格式,不合格就触发重跑。这个“规则+LLM”的混合思路,比让LLM去检查另一个LLM的输出靠谱得多。
6. 部署、调优与避坑:实测中反复踩过的雷
手写循环理解了原理、框架管理了状态、多Agent解决了分工,但产品要上线,还得跨过最后几道坎。这一章我直接把我踩过的雷和沉淀的方法一次性倒给你们。
6.1 上下文膨胀与循环失控的防御
上下文膨胀是Agent项目第一个拦路虎。顺着第4.3节的思路,我再补充一个策略:针对工具返回做结果压缩。比如查询数据库返回了200行,别把200行全塞进上下文,先截断前20行,然后用一行话告诉模型“本次共返回200条记录,以下是前20条摘要”。
循环失控是另一个大坑。模型在一个决策点上反复调用同一个工具,就是不收尾。LangGraph里我建议三层防御:第一层,给should_continue加步数计数器,超过N步强制跳到END;第二层,在Prompt里明确“如果同一步骤重复超过3次,请给出当前已获得的信息作为最终结论”;第三层,把工具调用次数统计进State,用代码检测重复调用模式并主动终止。
def should_continue(state: AgentState): last_message = state["messages"][-1] steps = len(state.get("intermediate_steps", [])) if steps > 8: # 超过8步强制结束 return "end" if last_message.tool_calls: return "continue" return "end"6.2 工具设计的原则:命名、参数校验、返回精简
工具是Agent的手脚,工具设计质量直接决定Agent的表现上限。我总结了三句话:
工具命名要说人话。get_user_order_info比qryord好一百倍,模型从名字就能推断用途,减少误调用。描述里要把边界条件写清楚,比如“查询某个日期之后创建的用户订单”,别让模型自由发挥。
参数必须做白名单校验。模型返回的参数值本质上是模型“猜”的,不可信。尤其是涉及路径、URL、命令的字段,必须严格校验合法字符和取值空间。第3.2节里那个calculator因为我做了字符白名单才敢跑,原理是一样的。
返回结果要精简但保留关键信息。给模型的信息过多,它抓不住重点;过少,它没法做后续决策。我习惯的工具返回格式是“操作状态 + 核心结果 + 截止条数”。比如数据库查询就返回“查询成功,共120条记录,已返回前20条,字段包括订单号、金额、时间”。
6.3 成本与可观测性:日志、追踪、缓存
Agent应用的成本不像普通API调用那么透明。一个用户请求可能触发十几次模型调用,每次还在动态增长消息长度。我至少做了三件事来管理成本:
一是做语义缓存。对高频重复的问题(比如“你们的退货政策是什么”),第一次Agent完整跑完流程后,把最终结果和用到的工具结果缓存起来。第二次再来直接命中缓存,不重新跑循环。实测在客服场景,缓存能省掉40%以上的Token开销。
二是给每次会话打trace_id。从用户进来开始,给每次Agent运行分配一个唯一ID,每步模型的输入输出、工具调用、耗时、Token消耗全部记录到日志。这块可以直接用LangSmith,或者自己写个装饰器记到ClickHouse。没有追踪日志,Agent线上出问题你是没法排查的,因为每次运行都是一个多步决策过程,跟普通接口的一次性调用完全不一样。
三是模型分级。简单任务用便宜的小模型处理,复杂任务再切大模型。我现在的一个实践是,先用一个小模型做快速分类——“这个问题需要调工具吗?大概要几步?”分类结果决定后续走轻量链路还是重量级链路。成本能省,响应也更快。
最后分享一个我自己的小心得:Agent开发最难的从来不是把功能跑通,而是让它在不可控的输入面前保持稳定。框架是工具,理解是根本。如果你现在准备上手Agent项目,我的建议是先手动实现一遍最小循环,再用LangGraph做状态管理,最后才考虑多Agent拆分——顺序反了,你会被各种抽象概念绕晕。
这篇内容算是把Agent实战的主干脉络都过了一遍,你照着这条路走一遍,遇到的具体问题欢迎随时来交流,我一定知无不言。