💡 作者按：RAG 解决了"大模型不懂私有知识"的问题，但企业真正需要的往往不止是问答——自动查天气、订会议室、写周报、调内部 API、执行多步决策，这些都需要 AI Agent。本文带你用 LangGraph 从零搭建一个能自主规划、能调用工具、能处理失败重试的生产级 AI Agent，所有代码均可直接运行。
一、为什么需要 AI Agent？先理解本质
大模型本身是一个"超级大脑"，但它有三大局限：
- 没有手和脚——不能主动调用外部系统
- 没有记忆——每次对话都是全新的
- 不会自主规划——你问一步它答一步，不会自己拆解复杂任务
Agent 的核心思想就是给大模型装上三样东西：
┌─────────────────────────────────────────────────────┐ │ AI Agent 架构 │ ├─────────────────────────────────────────────────────┤ │ │ │ ┌─────────┐ ┌──────────┐ ┌──────────────┐ │ │ │ Planning │───▶│ Tools │───▶│ Memory │ │ │ │ (规划) │ │ (工具调用)│ │ (记忆存储) │ │ │ └─────────┘ └──────────┘ └──────────────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌─────────────────────────────────────────────┐ │ │ │ LLM (推理引擎) │ │ │ │ 接收输入 → 思考 → 决定动作 → 执行 → 观察 │ │ │ └─────────────────────────────────────────────┘ │ │ │ │ │ ▼ │ │ 最终回答 / 执行结果 │ └─────────────────────────────────────────────────────┘
Agent 的推理循环（ReAct 范式）可以概括为：
思考（Thought）→ 行动（Action）→ 观察（Observation）→ 再思考 → ... → 最终回答
二、技术选型与架构设计
组件 |
选型 |
理由 |
Agent 框架 |
LangGraph 0.2+ |
相比 LangChain 的 AgentExecutor，LangGraph 提供状态图级别的精细控制，支持循环、条件分支、人工介入 |
LLM |
DeepSeek / GPT-4o |
需要较强的推理和工具调用能力 |
工具层 |
自定义 Tool + Tavily Search |
搜索 + 自定义业务工具 |
记忆 |
Redis |
跨会话持久化，支持 TTL 过期 |
Web 框架 |
FastAPI |
异步、高性能 |
LangGraph vs LangChain AgentExecutor 的核心区别：
LangChain AgentExecutor: 输入 → LLM → 工具 → LLM → 输出（黑盒，难控制） LangGraph: 定义状态图 → 节点 = 函数 → 边 = 条件 → 完全可控
LangGraph 让你能精确控制 Agent 的每一步行为——什么时候调用工具、什么时候该停下来问人、什么时候该重试。
三、环境准备
mkdir ai-agent-langgraph && cd ai-agent-langgraph python -m venv .venvsource .venv/bin/activate pip install fastapi==0.115.0 uvicorn==0.30.0 pip install langgraph==0.2.0 langchain==0.3.0 langchain-core==0.3.0 pip install langchain-openai==0.2.0 pip install tavily-python==0.3.0 pip install redis==5.0.0 python-dotenv==1.0.0 pip install pydantic==2.9.0
.env 文件：
# .envLLM_API_KEY=sk-your-key LLM_BASE_URL=https://api.deepseek.com/v1 LLM_MODEL=deepseek-chat TAVILY_API_KEY=tvly-your-tavily-key REDIS_HOST=localhost REDIS_PORT=6379 REDIS_DB=0 MAX_ITERATIONS=10 # Agent 最大推理步数（防无限循环）
💡 Tavily 是专为 AI Agent 设计的搜索 API，比直接用 Bing/Google 更适合 Agent 场景（返回结构化的、去广告的搜索结果）。免费额度足够开发测试。
四、核心代码实现
4.1 配置与客户端
config.py：
import osfrom dotenv import load_dotenvfrom openai import OpenAIimport redis load_dotenv()# LLMLLM_API_KEY = os.getenv("LLM_API_KEY", "") LLM_BASE_URL = os.getenv("LLM_BASE_URL", "https://api.openai.com/v1") LLM_MODEL = os.getenv("LLM_MODEL", "gpt-4o")# Tavily SearchTAVILY_API_KEY = os.getenv("TAVILY_API_KEY", "")# RedisREDIS_HOST = os.getenv("REDIS_HOST", "localhost") REDIS_PORT = int(os.getenv("REDIS_PORT", 6379)) REDIS_DB = int(os.getenv("REDIS_DB", 0)) MAX_ITERATIONS = int(os.getenv("MAX_ITERATIONS", 10))# 全局客户端llm_client = OpenAI(api_key=LLM_API_KEY, base_url=LLM_BASE_URL) redis_client = redis.Redis( host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB, decode_responses=True)
4.2 工具定义（Tools）
tools.py：
import requestsfrom typing import List, Dict, Anyfrom tavily import TavilyClientfrom config import TAVILY_API_KEY tavily = TavilyClient(api_key=TAVILY_API_KEY)def search_web(query: str) -> str: """ 搜索互联网获取最新信息。 适用于：新闻、实时数据、常识性问题、需要验证的信息。 """ try: results = tavily.search( query=query, max_results=5, search_depth="advanced" ) # 格式化搜索结果 formatted = [] for i, r in enumerate(results.get("results", []), 1): formatted.append( f"[{i}] {r['title']}\n" f" URL: {r['url']}\n" f" 摘要: {r.get('content', '')[:300]}..." ) return "\n\n".join(formatted) if formatted else "未找到相关结果" except Exception as e: return f"搜索失败: {str(e)}"def get_weather(city: str) -> str: """ 获取指定城市的当前天气信息。 适用于：用户询问天气、出行建议等场景。 """ # 使用 wttr.in 免费天气 API（无需 key） try: resp = requests.get( f"https://wttr.in/{city}?format=j1", timeout=10 ) data = resp.json() current = data["current_condition"][0] desc = current["weatherDesc"][0]["value"] temp_c = current["temp_C"] feels_like = current["FeelsLikeC"] humidity = current["humidity"] wind_kmph = current["windspeedKmph"] return ( f"{city}当前天气：{desc}，气温 {temp_c}°C，" f"体感 {feels_like}°C，湿度 {humidity}%，" f"风速 {wind_kmph} km/h" ) except Exception as e: return f"获取天气失败: {str(e)}"def calculate(expression: str) -> str: """ 计算数学表达式。支持 + - * / 以及括号。 适用于：数学计算、单位换算等。 """ try: # 安全计算：只允许数学运算 allowed_chars = set("0123456789+-*/.() ") if not all(c in allowed_chars for c in expression): return "表达式包含非法字符，仅支持数字和 + - * / ()" result = eval(expression, {"__builtins__": {}}, {}) return f"{expression} = {result}" except Exception as e: return f"计算失败: {str(e)}"def save_note(title: str, content: str) -> str: """ 保存一条笔记到知识库。 适用于：用户要求记录信息、写备忘、保存重要内容。 """ from config import redis_client import json import time note_id = f"note:{int(time.time())}" note = { "title": title, "content": content, "created_at": time.strftime("%Y-%m-%d %H:%M:%S") } redis_client.set(note_id, json.dumps(note, ensure_ascii=False)) redis_client.expire(note_id, 7 * 24 * 3600) # 7天过期 return f"笔记已保存：{title}（ID: {note_id}）"# 工具注册表：Agent 能看到的所有工具TOOLS = { "search_web": { "func": search_web, "description": "搜索互联网获取最新信息。输入：搜索关键词（字符串）", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "搜索关键词"} }, "required": ["query"] } }, "get_weather": { "func": get_weather, "description": "获取指定城市的当前天气。输入：城市名称（字符串，如'北京'、'London'）", "parameters": { "type": "object", "properties": { "city": {"type": "string", "description": "城市名称"} }, "required": ["city"] } }, "calculate": { "func": calculate, "description": "计算数学表达式。输入：表达式字符串（如 '123 * 456'）", "parameters": { "type": "object", "properties": { "expression": {"type": "string", "description": "数学表达式"} }, "required": ["expression"] } }, "save_note": { "func": save_note, "description": "保存笔记。输入：标题和内容", "parameters": { "type": "object", "properties": { "title": {"type": "string", "description": "笔记标题"}, "content": {"type": "string", "description": "笔记内容"} }, "required": ["title", "content"] } } }
4.3 LangGraph 状态图定义
agent_graph.py：
import jsonfrom typing import TypedDict, List, Optional, Annotatedfrom langgraph.graph import StateGraph, ENDfrom langgraph.prebuilt import ToolNodefrom langchain_core.messages import HumanMessage, AIMessage, SystemMessagefrom config import llm_client, LLM_MODEL, MAX_ITERATIONSfrom tools import TOOLS# ========== 状态定义 ==========class AgentState(TypedDict): """Agent 的完整状态，在图的节点间传递""" messages: List[dict] # 对话历史 iterations: int # 当前推理步数 max_iterations: int # 最大步数限制 final_answer: Optional[str] # 最终答案（有值则结束）# ========== 系统提示词 ==========SYSTEM_PROMPT = """你是一个智能助手，可以调用工具来完成任务。 ## 可用工具 {tools_description} ## 工作规则 1. 分析用户需求，决定是否需要调用工具 2. 如果需要工具，输出 JSON 格式：{{"tool": "工具名", "args": {{"参数名": "参数值"}}}} 3. 如果工具返回了结果，分析结果并决定下一步（继续调用工具 or 给出最终答案） 4. 如果信息已足够回答用户，输出最终答案（纯文本，不要 JSON） ## 重要 - 每次只调用一个工具 - 工具调用结果会在下一步以"Observation"形式提供给你 - 最终答案要简洁、准确、有结构 """def format_tools_description() -> str: """格式化工具描述，注入到系统提示词""" descriptions = [] for name, info in TOOLS.items(): descriptions.append(f"- **{name}**: {info['description']}") return "\n".join(descriptions)# ========== 节点函数 ==========def agent_node(state: AgentState) -> AgentState: """ Agent 推理节点：接收当前状态，决定下一步动作。 这是 Agent 的"大脑"——思考、规划、决策。 """ messages = state["messages"] iterations = state.get("iterations", 0) # 步数安全检查 if iterations >= state.get("max_iterations", MAX_ITERATIONS): return { **state, "final_answer": "抱歉，任务过于复杂，已达到最大推理步数限制。请尝试简化问题。", "iterations": iterations } # 构造消息列表 system_msg = SystemMessage( content=SYSTEM_PROMPT.format(tools_description=format_tools_description()) ) # 转换历史消息为 LangChain 格式 lc_messages = [system_msg] for msg in messages: if msg["role"] == "user": lc_messages.append(HumanMessage(content=msg["content"])) elif msg["role"] == "assistant": lc_messages.append(AIMessage(content=msg["content"])) elif msg["role"] == "observation": # 工具返回结果作为 observation 注入 lc_messages.append(HumanMessage( content=f"Observation: {msg['content']}" )) # 调用 LLM resp = llm_client.chat.completions.create( model=LLM_MODEL, messages=[ {"role": "system", "content": system_msg.content}, *[{"role": m.type, "content": m.content} for m in lc_messages[1:]] ], temperature=0.1, response_format={"type": "text"} ) content = resp.choices[0].message.content.strip() # 判断是工具调用还是最终答案 try: # 尝试解析为 JSON（工具调用） parsed = json.loads(content) if "tool" in parsed and "args" in parsed: tool_name = parsed["tool"] tool_args = parsed["args"] if tool_name not in TOOLS: # 工具不存在，告诉 Agent new_messages = messages + [ {"role": "assistant", "content": content}, {"role": "observation", "content": f"错误：工具 '{tool_name}' 不存在"} ] else: # 执行工具 tool_func = TOOLS[tool_name]["func"] try: result = tool_func(**tool_args) except Exception as e: result = f"工具执行失败: {str(e)}" new_messages = messages + [ {"role": "assistant", "content": content}, {"role": "observation", "content": result} ] return { **state, "messages": new_messages, "iterations": iterations + 1, "final_answer": None } except json.JSONDecodeError: pass # 不是 JSON，当作最终答案 # 最终答案 new_messages = messages + [{"role": "assistant", "content": content}] return { **state, "messages": new_messages, "iterations": iterations + 1, "final_answer": content }def should_continue(state: AgentState) -> str: """ 条件边：判断是继续推理还是结束。 """ if state.get("final_answer"): return "end" return "continue"# ========== 构建图 ==========def build_graph(): """构建 LangGraph 状态图""" graph = StateGraph(AgentState) # 添加节点 graph.add_node("agent", agent_node) # 设置入口 graph.set_entry_point("agent") # 添加条件边 graph.add_conditional_edges( "agent", should_continue, { "continue": "agent", # 循环回 agent 节点（继续推理） "end": END # 结束 } ) return graph.compile()
4.4 FastAPI 服务
main.py：
import uuidfrom typing import List, Optionalfrom fastapi import FastAPI, HTTPExceptionfrom fastapi.middleware.cors import CORSMiddlewarefrom pydantic import BaseModelimport loggingfrom config import redis_clientfrom agent_graph import build_graph logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI( title="AI Agent 服务", description="基于 LangGraph 的多工具智能体 API", version="1.0.0") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"] )# 编译图（全局单例）graph = build_graph()# ========== 数据模型 ==========class ChatRequest(BaseModel): message: str session_id: Optional[str] = Noneclass ChatResponse(BaseModel): answer: str session_id: str iterations: int tool_calls: List[dict]# ========== 会话管理 ==========def get_session(session_id: str) -> dict: """从 Redis 获取会话状态""" data = redis_client.get(f"session:{session_id}") if data: import json return json.loads(data) return {"messages": []}def save_session(session_id: str, state: dict): """保存会话状态到 Redis（30分钟过期）""" import json redis_client.setex( f"session:{session_id}", 1800, json.dumps(state, ensure_ascii=False) )# ========== API 路由 ==========@app.post("/chat", response_model=ChatResponse)async def chat(req: ChatRequest): """ 与 Agent 对话。 支持多轮对话：传入 session_id 可保持上下文。 """ session_id = req.session_id or str(uuid.uuid4()) # 获取历史会话 session = get_session(session_id) # 添加用户消息 session["messages"].append({ "role": "user", "content": req.message }) # 初始化状态 initial_state = { "messages": session["messages"], "iterations": 0, "max_iterations": 10, "final_answer": None } try: # 执行图 final_state = graph.invoke(initial_state) # 提取最终答案 answer = final_state.get("final_answer", "抱歉，未能生成答案。") iterations = final_state.get("iterations", 0) # 提取工具调用记录 tool_calls = [] messages = final_state.get("messages", []) for msg in messages: if msg.get("role") == "assistant": try: import json parsed = json.loads(msg["content"]) if "tool" in parsed: tool_calls.append(parsed) except (json.JSONDecodeError, KeyError): pass # 更新会话（只保留最近 20 轮） updated_messages = final_state.get("messages", []) session["messages"] = updated_messages[-40:] # 每轮 user+assistant = 2条 save_session(session_id, session) logger.info(f"Session {session_id}: {iterations} iterations, {len(tool_calls)} tool calls") return ChatResponse( answer=answer, session_id=session_id, iterations=iterations, tool_calls=tool_calls ) except Exception as e: logger.error(f"Agent 执行失败: {e}") raise HTTPException(status_code=500, detail=f"Agent 执行异常: {str(e)}")@app.get("/sessions/{session_id}")async def get_session_info(session_id: str): """查看会话详情""" session = get_session(session_id) return { "session_id": session_id, "message_count": len(session.get("messages", [])), "messages": session.get("messages", []) }@app.delete("/sessions/{session_id}")async def clear_session(session_id: str): """清除会话""" redis_client.delete(f"session:{session_id}") return {"status": "cleared", "session_id": session_id}@app.get("/health")async def health(): return {"status": "healthy", "service": "ai-agent"}# ========== 启动 ==========if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8001)
五、运行与测试
5.1 启动服务
# 确保 Redis 已启动redis-server# 启动 Agent 服务uvicorn main:app --host 0.0.0.0 --port 8001 --reload
访问 http://localhost:8001/docs 查看 Swagger 文档。
5.2 测试对话
# 1. 简单计算 + 天气查询（多步推理）curl -X POST "http://localhost:8001/chat" \ -H "Content-Type: application/json" \ -d '{ "message": "北京今天天气怎么样？另外帮我算一下 15 * 28 是多少" }'# 返回示例：# {# "answer": "北京今天天气：晴，气温 22°C，体感 21°C，湿度 45%，风速 12 km/h。\n\n另外，15 × 28 = 420。",# "session_id": "a1b2c3d4-...",# "iterations": 3,# "tool_calls": [# {"tool": "get_weather", "args": {"city": "北京"}},# {"tool": "calculate", "args": {"expression": "15 * 28"}}# ]# }# 2. 多轮对话（带上下文）curl -X POST "http://localhost:8001/chat" \ -H "Content-Type: application/json" \ -d '{ "message": "帮我记一下，今天下午3点要开会讨论Q4规划", "session_id": "a1b2c3d4-..." }'# 3. 搜索 + 总结curl -X POST "http://localhost:8001/chat" \ -H "Content-Type: application/json" \ -d '{ "message": "帮我搜索一下 LangGraph 和 LangChain 的最新区别，用中文总结" }'
5.3 Agent 推理过程可视化
Agent 处理上述第一个请求时的内部推理链：
Step 1: 用户 → "北京今天天气怎么样？另外帮我算一下 15 * 28" Agent 思考 → 需要调用两个工具 Agent 行动 → {"tool": "get_weather", "args": {"city": "北京"}} Step 2: Observation → "北京当前天气：晴，气温 22°C..." Agent 思考 → 天气已获取，还需要计算 Agent 行动 → {"tool": "calculate", "args": {"expression": "15 * 28"}} Step 3: Observation → "15 * 28 = 420" Agent 思考 → 两个任务都完成了，可以给出最终答案 Agent 回答 → "北京今天天气：晴... 另外，15 × 28 = 420。"
六、生产级增强方案
6.1 流式输出（SSE）
用户不希望等 Agent 推理完才看到结果，应该实时看到每一步：
from sse_starlette.sse import EventSourceResponseimport asyncio@app.post("/chat/stream")async def chat_stream(req: ChatRequest): """流式输出 Agent 推理过程""" async def event_generator(): # 逐步 yield 每一步推理 yield {"event": "thinking", "data": "正在分析您的问题..."} # 执行图，每步 yield async for step in graph.astream(initial_state): if step.get("tool_call"): yield { "event": "tool_call", "data": json.dumps(step["tool_call"]) } if step.get("observation"): yield { "event": "observation", "data": step["observation"] } yield { "event": "final_answer", "data": final_state["final_answer"] } return EventSourceResponse(event_generator())
6.2 人工介入（Human-in-the-Loop）
某些操作（如发送邮件、删除数据）需要人工确认：
def human_approval_node(state: AgentState) -> AgentState: """人工审批节点""" pending_action = state.get("pending_action") # 暂停图执行，等待人工输入 # LangGraph 的 interrupt() 机制 ...# 在图中添加审批边graph.add_node("human_approval", human_approval_node) graph.add_edge("agent", "human_approval") graph.add_conditional_edges("human_approval", check_approval)
6.3 工具调用失败自动重试
def agent_node_with_retry(state: AgentState) -> AgentState: """带重试的 Agent 节点""" max_retries = 2 for attempt in range(max_retries + 1): try: return agent_node(state) except Exception as e: if attempt == max_retries: state["final_answer"] = f"执行失败，已重试 {max_retries} 次: {str(e)}" return state state["messages"].append({ "role": "observation", "content": f"第 {attempt+1} 次尝试失败: {e}，正在重试..." })
6.4 多 Agent 协作
复杂场景可以用"主管 Agent + 专家 Agent"模式：
┌─────────────────────────────────────────┐ │ Supervisor Agent │ │ （理解意图，分配子任务） │ ├──────────┬──────────┬───────────────────┤ │ 搜索Agent │ 代码Agent │ 数据Agent │ │ (Tavily) │ (Python) │ (SQL/API) │ └──────────┴──────────┴───────────────────┘
七、踩坑清单
坑 1：LLM 输出格式不稳定
即使你要求输出 JSON，模型偶尔还是会输出带 markdown 代码块的格式（json ...）。一定要做格式清洗：
import redef extract_json(text: str) -> dict: # 尝试直接解析 try: return json.loads(text) except json.JSONDecodeError: pass # 尝试提取代码块 match = re.search(r'```(?:json)?\s*([\s\S]*?)\s*```', text) if match: return json.loads(match.group(1)) raise ValueError(f"无法解析 JSON: {text[:100]}")
坑 2：工具描述写不好，Agent 就不会用工具
工具描述是 Agent 决定"何时用、怎么用"的唯一依据。描述要：
- ✅ 明确说明适用场景（"当用户询问天气时使用"）
- ✅ 给出参数示例（"city: '北京' 或 'London'"）
- ❌ 不要写模糊描述（"获取一些信息"）
坑 3：无限循环
Agent 可能陷入"调用工具 → 结果不满意 → 换个参数再调 → 还是不满意"的死循环。必须设置 max_iterations，并在接近上限时给 Agent 一个"强制总结"的提示。
坑 4：上下文窗口爆炸
多轮对话中，历史消息会不断累积。如果 Agent 每轮都把所有历史塞给 LLM，很快会超出上下文窗口。策略：
- 只保留最近 N 轮
- 对旧消息做摘要压缩
- 工具调用的中间结果不进入长期记忆
坑 5：并发安全问题
LangGraph 的 StateGraph 本身是线程安全的，但如果你在节点函数中使用了全局可变状态（如共享的计数器），需要加锁。
八、项目结构
ai-agent-langgraph/ ├── config.py # 配置与客户端 ├── tools.py # 工具定义与注册 ├── agent_graph.py # LangGraph 状态图 ├── main.py # FastAPI 入口 ├── requirements.txt ├── .env └── tests/ ├── test_tools.py # 工具单元测试 └── test_agent.py # Agent 集成测试
九、总结
本文实现的 AI Agent 具备以下生产级特性：
- ✅ 多工具编排：搜索、天气、计算、笔记，可无限扩展
- ✅ 自主规划：Agent 自己决定调用哪些工具、调用顺序
- ✅ 循环推理：观察 → 思考 → 行动，直到任务完成
- ✅ 会话记忆：Redis 持久化，支持多轮对话
- ✅ 安全控制：步数限制、工具白名单、输入校验
- ✅ 可观测：每步工具调用都有记录，方便调试
RAG vs Agent 怎么选？
场景 |
选 RAG |
选 Agent |
问答、知识检索 |
✅ |
|
需要实时数据 |
✅ |
|
需要执行操作 |
✅ |
|
多步复杂任务 |
✅ |
|
简单信息查找 |
✅ |
下一步演进：
- MCP 协议接入：用 Model Context Protocol 标准化工具定义，让 Agent 能接入任意第三方服务
- 多 Agent 编排：用 LangGraph 的
SendAPI 实现动态子图分发 - RAG + Agent 融合：Agent 自己决定什么时候该检索知识库
- 评估体系：用 LangSmith 追踪每次推理链路，量化 Agent 表现
📌 老架构师的话：Agent 框架再花哨，核心还是 Prompt 质量 + 工具设计 + 状态管理。框架只是帮你把 ReAct 循环工程化，真正的智能来自于你对业务场景的理解和工具链的精心设计。别被"Agent"这个词唬住——它本质上就是一个
while循环里套了个 LLM 调用，关键在于你怎么设计循环里的每一步。
本文由 摸鱼不慌 发布，转载请注明出处。
