从零构建 AI Agent：基于 LangGraph 的多工具智能体实战（含完整代码）

💡 作者按：RAG 解决了"大模型不懂私有知识"的问题，但企业真正需要的往往不止是问答——自动查天气、订会议室、写周报、调内部 API、执行多步决策，这些都需要 AI Agent。本文带你用 LangGraph 从零搭建一个能自主规划、能调用工具、能处理失败重试的生产级 AI Agent，所有代码均可直接运行。

一、为什么需要 AI Agent？先理解本质

1. 没有手和脚——不能主动调用外部系统

2. 没有记忆——每次对话都是全新的

3. 不会自主规划——你问一步它答一步，不会自己拆解复杂任务

┌─────────────────────────────────────────────────────┐
│                  AI Agent 架构                       │
├─────────────────────────────────────────────────────┤
│                                                     │
│   ┌─────────┐    ┌──────────┐    ┌──────────────┐  │
│   │ Planning │───▶│  Tools   │───▶│   Memory     │  │
│   │ (规划)   │    │ (工具调用)│    │  (记忆存储)   │  │
│   └─────────┘    └──────────┘    └──────────────┘  │
│        │               │                   │        │
│        ▼               ▼                   ▼        │
│   ┌─────────────────────────────────────────────┐   │
│   │           LLM (推理引擎)                     │   │
│   │   接收输入 → 思考 → 决定动作 → 执行 → 观察   │   │
│   └─────────────────────────────────────────────┘   │
│                        │                            │
│                        ▼                            │
│              最终回答 / 执行结果                      │
└─────────────────────────────────────────────────────┘

思考（Thought）→ 行动（Action）→ 观察（Observation）→ 再思考 → ... → 最终回答

二、技术选型与架构设计

LangChain AgentExecutor:  输入 → LLM → 工具 → LLM → 输出（黑盒，难控制）
LangGraph:               定义状态图 → 节点 = 函数 → 边 = 条件 → 完全可控

三、环境准备

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

# .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 配置与客户端

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）

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 状态图定义

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 服务

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

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 推理过程可视化

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）

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 协作

┌─────────────────────────────────────────┐
│            Supervisor Agent              │
│     （理解意图，分配子任务）              │
├──────────┬──────────┬───────────────────┤
│ 搜索Agent │ 代码Agent │   数据Agent      │
│ (Tavily) │ (Python) │  (SQL/API)       │
└──────────┴──────────┴───────────────────┘

七、踩坑清单

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]}")

• ✅ 明确说明适用场景（"当用户询问天气时使用"）

• ✅ 给出参数示例（"city: '北京' 或 'London'"）

• ❌ 不要写模糊描述（"获取一些信息"）

• 只保留最近 N 轮

• 对旧消息做摘要压缩

• 工具调用的中间结果不进入长期记忆

八、项目结构

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 集成测试

九、总结

• ✅ 多工具编排：搜索、天气、计算、笔记，可无限扩展

• ✅ 自主规划：Agent 自己决定调用哪些工具、调用顺序

• ✅ 循环推理：观察 → 思考 → 行动，直到任务完成

• ✅ 会话记忆：Redis 持久化，支持多轮对话

• ✅ 安全控制：步数限制、工具白名单、输入校验

• ✅ 可观测：每步工具调用都有记录，方便调试

1. MCP 协议接入：用 Model Context Protocol 标准化工具定义，让 Agent 能接入任意第三方服务

2. 多 Agent 编排：用 LangGraph 的 Send API 实现动态子图分发

3. RAG + Agent 融合：Agent 自己决定什么时候该检索知识库

4. 评估体系：用 LangSmith 追踪每次推理链路，量化 Agent 表现

📌 老架构师的话：Agent 框架再花哨，核心还是 Prompt 质量 + 工具设计 + 状态管理。框架只是帮你把 ReAct 循环工程化，真正的智能来自于你对业务场景的理解和工具链的精心设计。别被"Agent"这个词唬住——它本质上就是一个 while 循环里套了个 LLM 调用，关键在于你怎么设计循环里的每一步。