在 AI 应用开发中，单纯的大模型调用已无法满足复杂场景需求 —— 我们需要智能体（Agent）具备工具调用、记忆管理、异常处理等能力，而 LangGraph 凭借其灵活的图结构，成为构建这类智能体的理想框架。本文将基于实际代码案例，从基础的智能体创建，到工具集成与错误处理，一步步带你掌握 LangGraph 智能体的核心开发技能。

一、LangGraph 智能体：不止于对话的 "AI 助手"

在 LangGraph 的语境中，智能体（Agent） 并非简单的对话机器人，而是具备 "自主决策 + 能力扩展" 的 AI 实体。它能像人类助手一样：接收任务需求，判断是否需要调用工具（如查日期、算数据），记录交互历史以理解上下文，甚至在出错时自我修正。

与传统大模型调用相比，LangGraph 智能体的核心优势在于流程封装—— 无需手动管理工具调用逻辑、对话记忆或异常处理，只需配置核心参数，智能体即可自主完成任务流转。接下来，我们从最基础的智能体开发开始，逐步叠加高级功能。

二、基础实战：创建你的第一个 LangGraph 智能体

要构建 LangGraph 智能体，首先需要完成环境准备与依赖安装，然后通过create_agent函数快速初始化智能体。我们以 DeepSeek 和 Qwen 两个主流大模型为例，演示基础的对话与流式输出功能。

1. 环境准备：安装核心依赖

首先安装 LangGraph 及大模型集成包（以 DeepSeek 和 Qwen 为例）：

# 安装LangGraph核心包与DeepSeek集成
pip install -U langgraph langchain-deepseek
# 若使用Qwen模型，需额外安装阿里云DashScope依赖
pip install -U dashscope langchain-qwq

2. 案例 1：基础智能体（非流式输出）

此案例实现一个能回答 "自我简介" 的基础智能体，支持 DeepSeek 和 Qwen 双模型切换：

from langchain_deepseek import ChatDeepSeek
from langchain_qwq import ChatQwen

# 初始化DeepSeek模型（替换api_key为你的实际密钥）
deepseek_llm = ChatDeepSeek(
    model="deepseek-chat",
    api_key="your-api-key",
    temperature=0.7,  # 控制回答随机性：0-1，值越高越灵活
    max_tokens=1024,  # 最大输出长度
    timeout=60,       # 请求超时时间（秒）
    max_retries=3     # 失败重试次数
)

# 初始化Qwen模型（需配置阿里云DashScope的base_url）
qwen_llm = ChatQwen(
    model="qwen3-max",
    api_key="your-api-key",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    temperature=0.7,
    timeout=60,
    max_retries=3,
    max_tokens=1024
)

# 调用DeepSeek智能体
deepseek_res = deepseek_llm.invoke("你是谁？能帮我解决什么问题？")
print("DeepSeek回答：", deepseek_res.content)

# 调用Qwen智能体
qwen_res = qwen_llm.invoke("你是谁？能帮我解决什么问题？")
print("Qwen回答：", qwen_res.content)

关键参数说明：

  • temperature：控制生成结果的随机性，0 适合需要确定答案的场景（如计算），0.7 适合对话类场景。
  • max_retries：网络波动时自动重试，提升服务稳定性。
  • base_url：Qwen 等模型需指定厂商 API 地址，确保请求正确路由。

3. 案例 2：流式输出智能体

在对话场景中，流式输出能提升用户体验（无需等待完整回答加载）。LangGraph 的stream方法支持多种输出模式，其中stream_mode="messages"最适合对话场景：

from langchain.agents import create_agent
from langchain_deepseek import ChatDeepSeek

# 初始化DeepSeek模型
deepseek_llm = ChatDeepSeek(
    model="deepseek-chat",
    api_key="your-api-key",
    temperature=0.7,
    max_tokens=1024,
    timeout=60,
    max_retries=3
)

# 创建智能体（暂不集成工具，仅对话）
agent = create_agent(
    model=deepseek_llm,
    tools=[],  # 工具列表为空，仅基础对话
    system_prompt="You are a helpful assistant. 用中文回答用户问题，解释需清晰易懂。"
)

# 流式调用智能体（查询单智能体与多智能体区别）
print("智能体流式回答：")
for chunk in agent.stream(
    input={
        "messages": [{"role": "user", "content": "帮我介绍一下单智能体和多智能体的区别"}]
    },
    stream_mode="messages"
):
    # 逐块打印，flush=True确保实时输出
    print(chunk[0].content, end="", flush=True)

流式模式解析：

  • stream_mode="messages"：按消息片段返回，适合直接展示给用户。
  • end=""与flush=True：避免换行并强制刷新缓冲区，实现 "打字机" 效果。

三、能力扩展：为智能体添加工具调用

基础对话仅能处理常识性问题，要让智能体具备 "查日期"" 算数据 ""查天气" 等实际能力，需集成工具（Tools）。LangGraph 通过@tool装饰器快速定义工具，并自动管理调用流程。

案例 3：集成 "日期查询" 工具

此案例实现一个能回答 "今天日期" 的智能体，当用户询问日期时，智能体自动调用get_current_date工具：

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_deepseek import ChatDeepSeek
import datetime

# 1. 定义工具：获取当前日期（@tool装饰器标记为工具）
@tool
def get_current_date(input: str = "") -> str:
    """
    获取今天的日期
    Returns:
        str: 日期格式为 YYYY-MM-DD（如2025-10-01）
    """
    return datetime.datetime.today().strftime("%Y-%m-%d")

# 2. 初始化模型与智能体
deepseek_llm = ChatDeepSeek(
    model="deepseek-chat",
    api_key="your-api-key",
    temperature=0.7,
    max_tokens=1024,
    timeout=60,
    max_retries=3
)

agent = create_agent(
    model=deepseek_llm,
    tools=[get_current_date],  # 集成日期查询工具
    # 系统提示词：明确工具使用场景
    system_prompt="你是一个乐于助人的助手。当用户查询「今天几号」「当前日期」等问题时，必须使用get_current_date工具获取最新日期，不能凭记忆回答。"
)

# 3. 调用智能体查询日期
print("智能体工具调用结果：")
s = agent.invoke(
    {"messages": [{"role": "user", "content": "今天几月几号啊"}] }
)

# 提取并打印最终回答（messages最后一条为结果）
res = s["messages"][-1].content
print("用户问题：今天几月几号啊")
print("智能体回答：", res)

工具开发关键要点：

  1. @tool装饰器：自动将 Python 函数转为 LangGraph 可识别的工具，无需手动封装调用逻辑。
  2. 工具描述（docstring）：需清晰说明工具功能与返回格式，大模型会根据描述判断是否调用该工具。
  3. 系统提示词引导：明确 "何时使用工具"，避免智能体凭记忆回答（如日期这类实时性数据）。

四、稳定性保障：为工具添加错误处理

工具调用可能出现异常（如除法运算中除数为 0），若不处理会导致智能体崩溃。LangGraph 提供ToolNode与自定义错误处理函数，确保智能体在出错时能友好反馈。

案例 4：带错误处理的 "除法计算" 工具

此案例实现一个除法计算工具，处理 "除数为 0"" 除数为 1（无意义）" 等异常场景：

from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode
from langchain.agents import create_agent
from langchain_deepseek import ChatDeepSeek
from langgraph.graph import StateGraph, MessagesState
from langchain_core.messages import HumanMessage

# 1. 定义除法工具（带基础异常处理）
@tool("devide_tool", return_direct=True)
def devide(a: int, b: int) -> str:
    """
    计算两个整数的除法
    Args:
        a (int): 除数（被除数 ÷ 除数 = 结果）
        b (int): 被除数
    """
    try:
        if b == 1:
            return "除数为1没有意义，请重新输个除数和被除数。"
        elif b == 0:
            return "除数不能为0，请重新输个除数和被除数。"
        return f"计算结果：{a} ÷ {b} = {a / b}"
    except Exception as e:
        return f"工具调用错误：{str(e)}"

# 2. 定义全局错误处理函数（兜底处理未捕获的异常）
def handle_tool_error(error: Exception) -> str:
    """工具调用全局错误处理"""
    if isinstance(error, ZeroDivisionError):
        return "除数不能为0，请重新输入。"
    elif isinstance(error, ValueError):
        return "输入不是有效整数，请重新输入数字。"
    return f"工具执行失败：{str(error)}"

# 3. 创建带错误处理的工具节点
tool_node = ToolNode(
    tools=[devide],
    handle_tool_errors=True  # 启用错误处理
)

# 4. 初始化模型与智能体
llm = ChatDeepSeek(
    model="deepseek-chat",
    api_key="your-api-key",
    temperature=0.7,
    timeout=60,
    max_retries=2
)

agent = create_agent(
    model=llm,
    tools=[devide],
    system_prompt="你是计算助手，用户需要除法计算时调用devide工具，确保输入为整数。"
)

# 5. 构建LangGraph工作流（控制工具调用逻辑）
def should_continue(state: MessagesState) -> str:
    """判断是否需要继续调用工具（有tool_calls则继续）"""
    last_msg = state["messages"][-1]
    return "continue" if (hasattr(last_msg, 'tool_calls') and last_msg.tool_calls) else "end"

def call_model(state: MessagesState):
    """调用大模型生成决策或回答"""
    response = agent.invoke({"messages": state["messages"]})
    return {"messages": [response["messages"][-1]]}

# 6. 组装工作流并编译
workflow = StateGraph(MessagesState)
workflow.add_node("agent", call_model)    # 智能体决策节点
workflow.add_node("tools", tool_node)    # 工具执行节点
workflow.set_entry_point("agent")        # 入口：先调用智能体决策
# 条件边：根据是否需要工具调用，决定下一步
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {"continue": "tools", "end": "__end__"}  # continue→工具节点，end→结束
)
workflow.add_edge("tools", "agent")      # 工具执行后，返回智能体整理结果
app = workflow.compile()                 # 编译工作流

# 7. 测试错误处理能力
print("测试1：10除以0")
result = app.invoke({"messages": [HumanMessage(content="10除以0等于多少？")]})
print(result["messages"][-1].content)

print("\n测试2：10除以1")
result = app.invoke({"messages": [HumanMessage(content="10除以1等于多少？")]})
print(result["messages"][-1].content)

print("\n测试3：10除以5")
result = app.invoke({"messages": [HumanMessage(content="10除以5等于多少？")]})
print(result["messages"][-1].content)

错误处理设计思路：

  1. 工具内基础校验：先处理明确的异常（如 b=0、b=1），减少错误流转。
  2. 全局错误处理：通过handle_tool_error兜底处理未捕获的异常（如输入非整数）。
  3. 工作流控制：should_continue函数判断是否需要调用工具，避免无效循环。

五、总结与后续方向

本文通过 4 个实际案例，带你掌握了 LangGraph 智能体的核心开发技能：从基础的对话智能体，到流式输出、工具集成，再到错误处理。这些能力已能满足多数中等复杂度的 AI 应用需求，如智能客服、数据查询助手等。

后续可进一步探索的方向包括：

  1. 记忆管理：为智能体添加短期记忆（会话内上下文）与长期记忆（用户历史数据），支持多轮对话连贯理解。
  2. 多智能体协作：构建 "协调者 + 专业智能体" 架构（如航班预订智能体 + 酒店预订智能体），处理复杂任务拆分。
  3. 人类监督（Human-in-the-loop）：在敏感操作（如支付、预订）前加入人工确认步骤，提升安全性。

LangGraph 的优势在于将复杂流程抽象为 "节点 + 边" 的图结构，开发者无需关注底层流转逻辑，只需聚焦核心功能设计。后续博客将带你深入这些高级特性，构建更强大的 AI 系统。

# 人工智能 # microsoft
Logo
ModelEngine社区

更多推荐

从零开始搭建Dify旅行助手Agent完整指南

本文介绍了使用Dify平台快速开发AI旅行助手应用的完整流程：从Docker部署Dify平台（解决国内网络问题）→初始化配置AI模型→创建智能旅行助手Agent→Postman接口测试→最终集成到Python Streamlit Web应用中。通过Dify的LLM开发平台，无需深厚AI背景即可高效构建智能应用，将开发效率提升10倍以上。教程包含详细步骤和代码示例，适合对AI应用开发感兴趣的开发者快

avatar ModelEngine社区
cover

08.大模型Function Call的应用

avatar ModelEngine社区

零代码构建企业级智能工作流：AutoAgent多模型协作与动态任务编排实战指南

你是否还在为复杂业务逻辑的自动化实现而烦恼？面对多步骤任务拆解、跨模型协作和动态流程调整时束手无策？本文将通过AutoAgent框架的数学解题工作流实例，带你掌握零代码构建智能业务流程的核心技术，无需编程基础也能打造企业级自动化解决方案。读完本文你将获得：多智能体协作设计方法、动态任务路由实现、结果聚合策略以及可视化工作流编排技巧。## 智能工作流核心架构解析AutoAgent工作流框架采

avatar ModelEngine社区