LangGraph 实战完全指南

原文: https://mp.weixin.qq.com/s/hWG1hV0mUxJKXQyR4_TWcg

最近跟几个朋友聊AI Agent开发，发现一个很有意思的现象：

“你们的智能体是多智能体实现的吗？” “用的什么多智能体框架？” “多智能体之间怎么协作的？”

一上来就是多智能体。

说实话，我当时就想问一句：单智能体你搞清楚了吗？

就像学编程，连if-else都没弄明白，就想上手分布式系统。基础不牢，地动山摇啊。

今天咱们就从最基础的单智能体讲起，把原理掰开了揉碎了说清楚，配上完整可运行的代码。看完这篇文章，你不仅能理解原理，还能直接上手实践。

一、为什么要从单智能体开始？

先说个真实场景。上周有个朋友找我，说他们团队要做一个客服系统，领导要求上多智能体。我问他：

“你们现在的需求是什么？”

“就是能回答用户问题，能联网搜索，能查数据库。”

“那你要多智能体干什么？”

“不是说多智能体更智能吗？”

这就是典型的概念混淆。其实他们的需求，一个设计良好的单智能体就完全够用了。

单智能体的核心能力：

理解用户意图

选择合适的工具

执行具体操作

生成最终回答

这四步走下来，90%的业务场景都能覆盖。

二、路由代理：教AI学会”判断”

2.1 核心原理

想象你是个客服，接到用户消息后第一件事是什么？判断用户想干什么。

“你好” → 打招呼，直接回复

“我叫张三，手机138xxxx” → 要登记信息，存数据库

“你们地址在哪” → 查询问题，回答地址

路由代理做的就是这件事：根据输入内容，选择不同的处理路径。

在LangGraph里，这个判断过程通过add_conditional_edges实现：

graph.add_conditional_edges(
    "判断节点",           # 从哪个节点开始判断
    routing_function,    # 判断逻辑
    {
        True: "路径A",    # 条件为真走这里
        False: "路径B"    # 条件为假走这里
    }
)

2.2 关键技术：结构化输出

这里有个核心问题：AI怎么知道该走哪条路？

答案是：让AI的输出遵循固定格式。

就像你在网站注册，必须填姓名、邮箱、手机号，不能乱填。我们也要求AI按照指定格式输出。

LangGraph支持三种方式，我最推荐Pydantic，因为它带类型检查：

from pydantic import BaseModel, Field
from typing import Optional
class UserInfo(BaseModel):
    """用户信息模型"""
    name: str = Field(description="用户姓名")
    age: Optional[int] = Field(description="用户年龄")
    email: str = Field(description="邮箱地址")
    phone: Optional[str] = Field(description="手机号")

然后用.with_structured_output()让AI按这个格式输出：

structured_llm = llm.with_structured_output(UserInfo)
result = structured_llm.invoke("我叫张三，28岁，邮箱zhangsan@qq.com")
# result会是一个UserInfo对象，字段都填好了

2.3 完整可运行代码

好，现在上干货。我们做一个智能客服，能够：

识别用户信息并存入数据库

普通对话直接回复

第一步：安装依赖

pip install langgraph langchain-openai sqlalchemy pymysql

第二步：完整代码

import os
from typing import Union, Optional, Annotated
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END, START
from langchain_core.messages import HumanMessage, AnyMessage
from typing import TypedDict
import operator
# ========== 1. 配置 ==========
os.environ["OPENAI_API_KEY"] = "your-api-key-here"
llm = ChatOpenAI(model="gpt-4o-mini")
# ========== 2. 定义数据模型 ==========
class UserInfo(BaseModel):
    """用户信息，包括姓名、年龄、邮箱和手机号"""
    name: str = Field(description="用户的姓名")
    age: Optional[int] = Field(description="用户的年龄")
    email: str = Field(description="用户的邮箱地址")
    phone: Optional[str] = Field(description="用户的手机号")
class ConversationalResponse(BaseModel):
    """普通对话回复"""
    response: str = Field(description="对用户的回复内容")
class FinalResponse(BaseModel):
    """最终输出，可能是用户信息或普通回复"""
    final_output: Union[UserInfo, ConversationalResponse]
# ========== 3. 定义图的状态 ==========
class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
# ========== 4. 定义节点函数 ==========
def chat_with_model(state):
    """第一步：让AI理解用户输入并生成结构化输出"""
    print(f"[chat_with_model] 收到消息: {state['messages'][-1].content}")

    messages = state['messages']
    structured_llm = llm.with_structured_output(FinalResponse)
    response = structured_llm.invoke(messages)

    print(f"[chat_with_model] AI判断结果: {type(response.final_output).__name__}")
    return {"messages": [response]}
def final_answer(state):
    """处理普通对话"""
    print("[final_answer] 生成普通回复")

    messages = state['messages'][-1]
    response = messages.final_output.response

    print(f"[final_answer] 回复内容: {response}")
    return {"messages": [response]}
def insert_db(state):
    """处理用户信息存储"""
    print("[insert_db] 准备存储用户信息")

    result = state['messages'][-1]
    output = result.final_output

    # 这里简化处理，实际项目需要真实的数据库操作
    print(f"[insert_db] 存储信息: 姓名={output.name}, 年龄={output.age}, "
          f"邮箱={output.email}, 手机={output.phone}")

    return {"messages": [f"✅ 已成功记录您的信息：{output.name}"]}
# ========== 5. 定义路由函数 ==========
def generate_branch(state: AgentState):
    """判断走哪条路：存储信息 or 普通回复"""
    result = state['messages'][-1]
    output = result.final_output

    if isinstance(output, UserInfo):
        print("[Router] 检测到用户信息，路由到数据库")
        return "insert_db"
    else:
        print("[Router] 检测到普通对话，路由到回复")
        return "final_answer"
# ========== 6. 构建图 ==========
graph_builder = StateGraph(AgentState)
# 添加节点
graph_builder.add_node("chat_with_model", chat_with_model)
graph_builder.add_node("final_answer", final_answer)
graph_builder.add_node("insert_db", insert_db)
# 设置起点
graph_builder.add_edge(START, "chat_with_model")
# 添加条件边（核心路由逻辑）
graph_builder.add_conditional_edges(
    "chat_with_model",
    generate_branch,
    {
        "insert_db": "insert_db",
        "final_answer": "final_answer"
    }
)
# 设置终点
graph_builder.add_edge("final_answer", END)
graph_builder.add_edge("insert_db", END)
# 编译图
graph = graph_builder.compile()
# ========== 7. 测试函数 ==========
def test_agent(query):
    print("\n" + "="*60)
    print(f"用户输入: {query}")
    print("="*60)

    input_message = {"messages": [HumanMessage(content=query)]}
    result = graph.invoke(input_message)

    print("\n最终结果:", result["messages"][-1])
    print("="*60 + "\n")
# ========== 8. 运行测试 ==========
if __name__ == "__main__":
    # 测试1：普通对话
    test_agent("你好，请介绍一下你自己")

    # 测试2：用户信息
    test_agent("我叫张三，28岁，邮箱zhangsan@qq.com，手机13812345678")

    # 测试3：另一个普通问题
    test_agent("今天天气怎么样？")

运行效果：

============================================================
用户输入: 你好，请介绍一下你自己
============================================================
[chat_with_model] 收到消息: 你好，请介绍一下你自己
[chat_with_model] AI判断结果: ConversationalResponse
[Router] 检测到普通对话，路由到回复
[final_answer] 生成普通回复
[final_answer] 回复内容: 你好！我是一个AI助手...

最终结果:

你好！我是一个AI助手...
============================================================
============================================================
用户输入: 我叫张三，28岁，邮箱zhangsan@qq.com，手机13812345678
============================================================
[chat_with_model] 收到消息: 我叫张三，28岁，邮箱zhangsan@qq.com，手机13812345678
[chat_with_model] AI判断结果: UserInfo
[Router] 检测到用户信息，路由到数据库
[insert_db] 准备存储用户信息
[insert_db] 存储信息: 姓名=张三, 年龄=28, 邮箱=zhangsan@qq.com, 手机=13812345678
✅ 已成功记录您的信息：张三
============================================================

2.4 原理拆解

看完代码，我们来捋一遍流程：

  1. 用户输入进来
    {"messages": [HumanMessage(content="我叫张三...")]}
  2. chat_with_model节点处理
  • 调用llm.with_structured_output(FinalResponse)
  • AI分析输入，返回UserInfo或ConversationalResponse对象
  1. generate_branch判断
    if isinstance(output, UserInfo):
 return "insert_db"  # 走数据库存储
else:
 return "final_answer"  # 走普通回复
  2. 执行对应节点
  • insert_db：存储用户信息
  • final_answer：生成回复
  1. 返回最终结果
    整个过程就像一个智能分拣员，根据包裹类型送到不同的处理窗口。

三、工具调用代理：让AI学会”使用工具”

3.1 路由代理的局限

上面的路由代理很好用，但有个明显问题：只能在两条路之间选。

如果我们想让AI：

能联网搜索最新消息

能查询天气

能存储用户信息

还能正常聊天

难道要写一堆if-else判断吗？那代码会变得非常臃肿。

这时候就需要工具调用代理了。

3.2 核心原理

工具调用代理的思路是：给AI一个工具箱，让它自己选工具。

就像装修师傅，面前摆着锤子、螺丝刀、电钻，他会根据任务选合适的工具。

在LangGraph里，这个过程分三步：

第一步：定义工具

from langchain_core.tools import tool
@tool
def search_web(query: str):
    """搜索互联网获取最新信息"""
    # 实际调用搜索API
    return "搜索结果..."
@tool
def get_weather(city: str):
    """查询城市天气"""
    if city == "北京":
        return "北京今天16度，晴"
    return "未知城市"

第二步：绑定到模型

tools = [search_web, get_weather]
llm_with_tools = llm.bind_tools(tools)
第三步：让AI自动调用

response = llm_with_tools.invoke("北京天气怎么样？")
# AI会自动生成：get_weather(city="北京")

3.3 完整可运行代码

这次我们做个功能更强大的助手，支持：

联网搜索

天气查询

用户信息存储

普通对话

完整代码：

import os
import json
import requests
from typing import Optional, Annotated
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph import StateGraph, END, START
from langgraph.prebuilt import ToolNode
from langchain_core.messages import HumanMessage, AnyMessage, AIMessage
from typing import TypedDict
import operator
# ========== 1. 配置 ==========
os.environ["OPENAI_API_KEY"] = "your-api-key-here"
llm = ChatOpenAI(model="gpt-4o-mini")
# ========== 2. 定义工具 ==========
@tool
def search_web(query: str):
    """搜索互联网获取最新信息

    Args:
        query: 搜索关键词
    """
    print(f"[Tool] 正在搜索: {query}")

    # 这里用一个简化的实现
    # 实际项目可以接入真实的搜索API
    return f"关于'{query}'的最新搜索结果：[这里是模拟的搜索结果]"
@tool
def get_weather(city: str):
    """查询城市天气

    Args:
        city: 城市名称
    """
    print(f"[Tool] 正在查询天气: {city}")

    weather_data = {
        "北京": "北京今天16度，天气晴朗",
        "上海": "上海今天20度，多云",
        "深圳": "深圳今天28度，有雨"
    }

    return weather_data.get(city, f"抱歉，暂时没有{city}的天气信息")
@tool
def save_user_info(name: str, age: int, email: str, phone: str):
    """保存用户信息到数据库

    Args:
        name: 用户姓名
        age: 用户年龄
        email: 邮箱地址
        phone: 手机号
    """
    print(f"[Tool] 正在保存用户信息: {name}")

    # 实际项目这里应该是真实的数据库操作
    print(f"  - 姓名: {name}")
    print(f"  - 年龄: {age}")
    print(f"  - 邮箱: {email}")
    print(f"  - 手机: {phone}")

    return f"✅ 已成功保存 {name} 的信息"
# ========== 3. 创建工具节点 ==========
tools = [search_web, get_weather, save_user_info]
tool_node = ToolNode(tools)
# 绑定工具到模型
llm_with_tools = llm.bind_tools(tools)
# ========== 4. 定义图的状态 ==========
class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
# ========== 5. 定义节点函数 ==========
def call_model(state):
    """调用大模型，让它决定要不要用工具"""
    print(f"\n[call_model] 收到消息: {state['messages'][-1].content}")

    messages = state['messages']
    response = llm_with_tools.invoke(messages)

    # 检查AI是否要调用工具
    if response.tool_calls:
        print(f"[call_model] AI决定调用工具: {[tc['name'] for tc in response.tool_calls]}")
    else:
        print("[call_model] AI决定直接回答")

    return {"messages": [response]}
# ========== 6. 定义路由函数 ==========
def should_continue(state: AgentState):
    """判断是否需要调用工具"""
    messages = state["messages"]
    last_message = messages[-1]

    # 如果AI生成了tool_calls，就去执行工具
    if last_message.tool_calls:
        return "tools"
    # 否则直接结束
    return END
# ========== 7. 构建图 ==========
workflow = StateGraph(AgentState)
# 添加节点
workflow.add_node("agent", call_model)  # AI决策节点
workflow.add_node("tools", tool_node)    # 工具执行节点
# 设置入口
workflow.add_edge(START, "agent")
# 添加条件边：AI决定后，要么调用工具，要么结束
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools",
        END: END
    }
)
# 工具执行完后，回到AI节点让它总结结果
workflow.add_edge("tools", "agent")
# 编译
graph = workflow.compile()
# ========== 8. 测试函数 ==========
def test_agent(query):
    print("\n" + "="*70)
    print(f"👤 用户: {query}")
    print("="*70)

    result = graph.invoke(
        {"messages": [HumanMessage(content=query)]},
        {"recursion_limit": 10}  # 防止无限循环
    )

    final_answer = result["messages"][-1].content
    print(f"\n🤖 助手: {final_answer}")
    print("="*70 + "\n")
# ========== 9. 运行测试 ==========
if __name__ == "__main__":
    # 测试1：普通对话
    test_agent("你好，请介绍一下你自己")

    # 测试2：天气查询
    test_agent("北京今天天气怎么样？")

    # 测试3：联网搜索
    test_agent("Claude 4.5 Sonnet有什么新功能？")

    # 测试4：用户信息
    test_agent("我叫李四，25岁，邮箱lisi@example.com，手机13987654321")

    # 测试5：复杂任务（可能调用多个工具）
    test_agent("帮我查一下上海的天气，然后搜索一下最近的AI新闻")

运行效果：

======================================================================
👤 用户: 北京今天天气怎么样？
======================================================================
[call_model] 收到消息: 北京今天天气怎么样？
[call_model] AI决定调用工具: ['get_weather']
[Tool] 正在查询天气: 北京
[call_model] 收到消息: ToolMessage(content='北京今天16度，天气晴朗')
[call_model] AI决定直接回答
🤖 助手: 北京今天的天气是16度，天气晴朗。
======================================================================
======================================================================
👤 用户: 我叫李四，25岁，邮箱lisi@example.com，手机13987654321
======================================================================
[call_model] 收到消息: 我叫李四，25岁，邮箱lisi@example.com，手机13987654321
[call_model] AI决定调用工具: ['save_user_info']
[Tool] 正在保存用户信息: 李四
  - 姓名: 李四
  - 年龄: 25
  - 邮箱: lisi@example.com
  - 手机: 13987654321
[call_model] 收到消息: ToolMessage(content='✅ 已成功保存 李四 的信息')
[call_model] AI决定直接回答
🤖 助手: 您的信息已经成功保存，李四！如果还有其他需要帮助的地方，请随时告诉我。
======================================================================

3.4 原理深度拆解

这个工具调用的流程比路由代理复杂一些，我们详细拆解：

流程图：

关键点1：AI如何知道要调用哪个工具？

当我们执行llm.bind_tools(tools)时，LangChain会把工具的信息（名称、描述、参数）告诉AI：

你现在有以下工具可用：

  1. search_web(query: str) - 搜索互联网获取最新信息
  2. get_weather(city: str) - 查询城市天气
  3. save_user_info(name, age, email, phone) - 保存用户信息

用户问题：北京今天天气怎么样？

请问你要调用哪个工具？参数是什么？

AI分析后会返回：

{
    "name": "get_weather",
    "args": {"city": "北京"}
}
关键点2：ToolNode如何执行工具？

ToolNode内部实现很简单：

def tool_node(state):
    results = []
    for tool_call in state["messages"][-1].tool_calls:
        # 找到对应的工具
        tool = tools_by_name[tool_call["name"]]
        # 执行工具
        result = tool.invoke(tool_call["args"])
        # 包装成ToolMessage
        results.append(ToolMessage(content=result, ...))
    return {"messages": results}
关键点3：为什么要回到agent节点？

因为工具的返回结果通常是原始数据，需要AI整理成人类能看懂的回答：

  • 工具返回: “北京今天16度，天气晴朗”
  • AI整理后: “北京今天的天气是16度，天气晴朗。适合外出活动哦！”

3.5 实战技巧

  1. 工具描述要清晰

AI是根据工具的描述来决定调用哪个的，所以描述一定要准确：

# ❌ 不好的描述
@tool
def func1(x):
    """一个函数"""
    pass
# ✅ 好的描述
@tool
def search_flights(departure: str, destination: str, date: str):
    """搜索航班信息

    Args:
        departure: 出发城市，如"北京"
        destination: 目的地城市，如"上海"  
        date: 日期，格式YYYY-MM-DD，如"2024-03-15"

    Returns:
        返回可用航班列表，包含时间、价格等信息
    """
    pass
  1. 处理工具调用失败

AI有时候会传错参数，要做好异常处理：

@tool
def get_weather(city: str):
    """查询天气"""
    try:
        # 实际的API调用
        result = weather_api.get(city)
        return result
    except Exception as e:
        # 返回友好的错误信息
        return f"抱歉，查询{city}的天气时出错了：{str(e)}"
  1. 设置递归限制

防止AI陷入无限循环调用工具：

result = graph.invoke(
    {"messages": [HumanMessage(content=query)]},
    {"recursion_limit": 10}  # 最多10次迭代
)

四、实际项目中的应用

4.1 客服系统

我最近用这套方案做了个客服系统，接入了：

  • 知识库搜索（RAG）
  • 工单创建
  • 订单查询
  • 物流追踪

核心代码就100多行，但能处理90%的常见问题。

4.2 数据分析助手

  • 给数据分析师用的助手：
  • SQL查询生成
  • 图表绘制
  • 数据清洗
  • 报告生成

分析师只需要说”帮我分析一下上个月的销售数据”，剩下的交给AI。

4.3 个人助理

我自己的私人助手：

  • 日程管理
  • 邮件处理
  • 待办提醒
  • 信息收集

基本上替代了我原来用的好几个工具。

五、常见问题答疑

Q1: 单智能体和多智能体到底什么区别？

单智能体：一个AI从头干到尾，就像一个全能型选手。 多智能体：多个AI协作，每个AI专注一个领域，像团队合作。

大部分场景下，单智能体完全够用。只有当：

  • 任务特别复杂，需要不同专业知识
  • 需要并行处理多个子任务
  • 需要不同的AI角色扮演

这时候才考虑多智能体。

Q2: 工具调用会不会很慢？

确实会比纯文本对话慢一点，因为多了：

  • AI判断要调用哪个工具
  • 实际执行工具
  • AI整理结果

但通常都在可接受范围（1-3秒）。如果对速度要求特别高，可以：

  • 优化工具执行速度
  • 使用更快的模型（如GPT-4o mini）
  • 缓存常见查询结果

Q3: 怎么保证AI不会乱调用工具？

几个方法：

  1. 清晰的工具描述：让AI明确知道每个工具的用途
  2. 参数验证：在工具内部验证参数合法性
  3. 权限控制：敏感操作需要用户确认
  4. 日志记录：记录所有工具调用，方便追踪

Q4: 出错了怎么办？

最佳实践是在每个节点加try-except：

def call_tool(state):
    try:
        result = tool.invoke(args)
        return {"messages": [result]}
    except Exception as e:
        error_msg = f"工具调用失败：{str(e)}"
        print(f"[Error] {error_msg}")
        return {"messages": [error_msg]}

六、下一步学习建议

如果你看到这里，说明你已经理解了单智能体的核心原理。接下来可以：

  1. 实践项目
  • 从简单的对话助手开始
  • 逐步添加工具功能
  • 处理各种边界情况
  1. 深入学习
  • 研究LangGraph的状态管理
  • 学习流式输出
  • 了解持久化存储
  1. 关注进阶话题
  • 自主循环代理（让AI自己决定迭代次数）
  • 人机协作（Human-in-the-loop）
  • 多智能体协作

但记住：**基础但记住：基础不牢，地动山摇。把单智能体玩透了，再去碰多智能体。

七、性能优化技巧

做完基础功能后，我们来聊聊怎么优化性能。

7.1 缓存常见查询

对于天气、搜索这类查询，可以加缓存：

from functools import lru_cache
from datetime import datetime, timedelta
# 缓存装饰器，5分钟过期
CACHE = {}
CACHE_EXPIRE = timedelta(minutes=5)
def cached_tool(func):
    """工具缓存装饰器"""
    def wrapper(*args, **kwargs):
        # 生成缓存key
        cache_key = f"{func.__name__}_{args}_{kwargs}"

        # 检查缓存
        if cache_key in CACHE:
            cached_time, cached_result = CACHE[cache_key]
            if datetime.now() - cached_time < CACHE_EXPIRE:
                print(f"[Cache] 使用缓存结果")
                return cached_result

        # 执行函数
        result = func(*args, **kwargs)

        # 存入缓存
        CACHE[cache_key] = (datetime.now(), result)
        return result

    return wrapper
@tool
@cached_tool
def get_weather(city: str):
    """查询天气（带缓存）"""
    # 实际的API调用
    pass

7.2 并行执行工具

如果AI同时调用多个工具，可以并行执行：

from concurrent.futures import ThreadPoolExecutor
def execute_tools_parallel(tool_calls):
    """并行执行多个工具"""
    with ThreadPoolExecutor(max_workers=5) as executor:
        futures = []
        for tool_call in tool_calls:
            future = executor.submit(
                execute_single_tool,
                tool_call
            )
            futures.append(future)

        results = [f.result() for f in futures]

    return results

7.3 使用更快的模型

对于简单任务，用mini模型就够了：

# 根据任务复杂度选择模型
def get_model(task_type):
    if task_type in ["simple_chat", "weather", "todo"]:
        return ChatOpenAI(model="gpt-4o-mini")  # 快且便宜
    else:
        return ChatOpenAI(model="gpt-4o")  # 复杂任务用完整版

7.4 流式输出

让用户更快看到响应：

def stream_response(self, user_input):
    """流式输出响应"""
    for chunk in self.llm.stream(user_input):
        print(chunk.content, end="", flush=True)

八、实战中的坑和解决方案

做了这么多项目，踩过的坑可以写本书了。这里分享几个最常见的。

坑1：AI不调用工具

现象：明明需要查天气，AI却自己瞎编答案。
原因：工具描述不够清晰，或者系统提示没有强调要用工具。
解决：

SYSTEM_PROMPT = """
重要：当用户询问天气、日程、待办等信息时，
你必须调用相应的工具获取准确信息，不要自己编造答案！
可用工具：
- get_weather: 查询实时天气
- list_todos: 查看待办事项
...
"""

坑2：工具参数错误

现象：AI传了错误的参数类型或格式。
原因：参数描述不够详细。
解决：

@tool
def add_schedule(date: str, time: str):
    """添加日程

    Args:
        date: 日期，必须是YYYY-MM-DD格式，例如"2024-03-15"
        time: 时间，必须是HH:MM格式，例如"14:30"

    注意：请严格按照格式要求传参！
    """
    # 加上格式验证
    import re
    if not re.match(r'\d{4}-\d{2}-\d{2}', date):
        return "❌ 日期格式错误，请使用YYYY-MM-DD格式"
    if not re.match(r'\d{2}:\d{2}', time):
        return "❌ 时间格式错误，请使用HH:MM格式"

    # 正常处理
    pass

坑3：无限循环

现象：AI不停地调用工具，停不下来。
原因：工具返回的信息AI理解不了，又重新调用。
解决：


# 1. 设置最大迭代次数
{"recursion_limit": 10}
# 2. 工具返回清晰的信息
@tool
def search_web(query):
    result = api.search(query)
    return f"搜索完成！结果：{result}\n请基于以上信息回答用户问题。"

坑4：响应太慢

现象：用户等半天才看到回复。
原因：工具执行慢，或者模型推理慢。
解决：

# 1. 异步执行
import asyncio
async def call_tool_async(tool_call):
    return await tool.ainvoke(tool_call["args"])
# 2. 添加加载提示
def chat_with_loading(user_input):
    print("🤖 小智正在思考...")
    result = self.chat(user_input)
    return result
# 3. 用流式输出
def stream_chat(user_input):
    print("🤖 小智：", end="")
    for chunk in model.stream(user_input):
        print(chunk, end="", flush=True)

坑5：工具调用失败

现象：API超时、数据库连接失败等。
原因：网络问题、配置错误、资源不足。
解决：

@tool
def robust_search(query: str):
    """带重试机制的搜索"""
    max_retries = 3

    for i in range(max_retries):
        try:
            result = search_api.get(query, timeout=5)
            return result
        except TimeoutError:
            if i < max

九、总结

还记得文章开头那个场景吗？当大家都在追逐“多智能体”这个热词时，我想起了自己第一次让智能体正确识别用户意图时的兴奋——那种“啊哈！”的时刻，比任何复杂框架都让人满足。今天这篇文章，可能就是你的那个“啊哈时刻”。我们不仅聊清楚了原理，更重要的是：

  • 你得到了两套完整可运行的代码，复制粘贴就能用
  • 理解了从路由判断到工具调用的进化路径
  • 看到了真实项目中的应用场景和避坑指南

接下来最好的学习路径：

  • 把文章里的路由代理代码跑起来，试试改判断逻辑
  • 加上一个你自己的工具（比如查快递、记笔记）
  • 遇到报错别慌，那才是你真正学会的开始

最后送大家一句话： 最好的AI应用不是用了最炫的框架，而是用最简单的技术真正解决了问题。当你把单智能体玩到得心应手时，你会发现——原来大部分场景，一个设计良好的“单智能体”真的就够了。

作者：Jeebiz  创建时间：2025-10-19 12:03
最后编辑：Jeebiz  更新时间：2025-10-19 12:19