在当今 AI 快速发展的时代，智能 Agent 已经成为构建复杂 AI 应用的重要基础。LangGraph 作为一个强大的框架，为我们提供了构建智能 Agent 的完整解决方案。本文将从基础概念开始，逐步介绍如何使用 LangGraph 构建功能完善的智能 Agent

一、什么是智能 Agent？

想象一下，您有一个虚拟员工，他不仅能够独立完成任务，还能在关键节点及时向您汇报并根据您的指示调整工作。这就是 LangGraph 为我们勾勒的智能 Agent 形象

智能Agent的核心特征：

• 自主性：能够独立完成基于大模型的任务​

• 交互性：可以与用户进行自然对话​

• 工具使用能力：能够调用外部工具扩展功能​

• 记忆能力：具备短期和长期记忆管理​

• 可监督性：关键节点允许人类干预和调整

二、从零开始构建聊天 Agent

让我们从最基础的聊天功能开始，构建第一个智能 Agent

2.1 环境准备

首先需要安装必要的依赖：

pip install -U langgraph langchain langchain-deepseek

2.2 基础聊天 Agent 实现

from langchain_deepseek import ChatDeepSeek
from langgraph.prebuilt import create_agent

# 初始化大模型客户端
llm = ChatDeepSeek(
    model="deepseek-chat",
    api_key="您的API密钥",
    temperature=0.7,
    timeout=60
)

# 创建基础聊天Agent
agent = create_agent(
    model=llm,
    tools=[],  # 暂时不使用工具
    system_prompt="你是一个乐于助人的AI助手，善于用简单易懂的语言回答问题。"
)

# 测试Agent
response = agent.invoke({
    "messages": [{"role": "user", "content": "你好，能介绍一下自己吗？"}]
})

print(response["messages"][-1]["content"])

2.3 流式输出功能

LangGraph 还支持流式输出，让对话体验更加流畅：

# 流式输出示例
for chunk in agent.stream({
    "messages": [{"role": "user", "content": "请详细介绍人工智能的发展历程。"}],
    "stream_mode": "messages"
}):
    if "messages" in chunk:
        print(chunk["messages"][-1]["content"], end="")

流式输出模式说明：

• "values": 返回状态值的完整快照​

• "updates": 返回状态值的增量更新​

• "messages": 返回消息内容（适合对话场景）​

• "tasks": 返回执行的任务信息​

• "debug": 返回调试信息

三、为 Agent 添加工具调用能力

工具调用是智能 Agent 的核心功能之一，它让 Agent 能够调用外部工具完成复杂任务

3.1 工具调用的工作原理

一个完整的工具调用流程包括：​

1、客户端定义工具类和功能​

2、将工具信息传递给大模型​

3、大模型判断是否需要调用工具​

4、返回工具调用指令​

5、执行工具并返回结果​

6、大模型根据结果生成回答

3.2 实现一个简单的工具

from langchain_core.tools import tool

# 定义一个计算除法的工具
@tool("division_tool")
def divide(a: int, b: int) -> str:
    """
    计算两个整数的除法
    
    参数:
    a (int): 被除数
    b (int): 除数
    
    返回:
    str: 计算结果
    """
    try:
        result = a / b
        return f"计算结果：{a} ÷ {b} = {result}"
    except ZeroDivisionError:
        return "错误：除数不能为零"

# 创建带工具的Agent
agent_with_tools = create_agent(
    model=llm,
    tools=[divide],
    system_prompt="你是一个聪明的助手，当遇到数学计算问题时，请使用提供的工具进行计算。"
)

# 测试工具调用
response = agent_with_tools.invoke({
    "messages": [{"role": "user", "content": "请计算 100 除以 25 的结果。"}]
})

print(response["messages"][-1]["content"])

3.3 工具调用错误处理

LangGraph 提供了完善的错误处理机制：

from langgraph.prebuilt import ToolNode

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

# 自定义错误处理函数
def handle_tool_error(error):
    if isinstance(error, ZeroDivisionError):
        return "计算错误：除数不能为零，请重新输入。"
    return f"工具执行错误：{str(error)}"

四、Agent 的记忆管理系统

记忆功能是智能 Agent 的重要特性，LangGraph 将记忆分为短期记忆和长期记忆

4.1 短期记忆（CheckPoint）

短期记忆用于当前对话的上下文管理：

from langgraph.checkpoint.memory import MemorySaver

# 创建记忆存储
memory = MemorySaver()

# 创建带记忆的Agent
agent_with_memory = create_agent(
    model=llm,
    tools=[divide],
    checkpointer=memory
)

# 使用特定的对话ID进行多轮对话
config = {"configurable": {"thread_id": "conversation_123"}}

# 第一轮对话
response1 = agent_with_memory.invoke({
    "messages": [{"role": "user", "content": "记住，我的名字是张三。"}],
}, config)

# 第二轮对话 - Agent应该记住名字
response2 = agent_with_memory.invoke({
    "messages": [{"role": "user", "content": "你还记得我叫什么名字吗？"}],
}, config)

print(response2["messages"][-1]["content"])

4.2 长期记忆（Store）

长期记忆用于持久化存储用户信息或应用数据：

from langgraph.store.memory import InMemoryStore

# 创建长期存储
long_term_store = InMemoryStore()

# 存储用户信息
long_term_store.put(
    ("users", "user_123"),  # (namespace, key)
    {"name": "张三", "age": 30, "preferences": ["科技", "阅读"]}
)

# 在Agent中使用长期记忆
def get_user_preferences(user_id: str):
    """获取用户偏好设置"""
    user_data = long_term_store.get(("users", user_id))
    if user_data:
        return f"用户{user_id}的偏好：{user_data['preferences']}"
    return "用户信息不存在"

4.3 记忆管理策略

短期记忆管理：

• 总结机制：定期对对话历史进行总结​

• 修剪机制：删除最旧的对话记录

from langgraph.prebuilt import create_react_agent
from langgraph.prebuilt.chat_agent_executor import summarization_hook

# 创建带记忆管理的Agent
agent_with_memory_management = create_react_agent(
    model=llm,
    tools=[divide],
    pre_model_hook=summarization_hook,  # 使用总结钩子
    checkpointer=memory
)

五、Human-In-Loop：人类监督机制

在关键决策点让人类进行监督和干预是生产环境中非常重要的功能

5.1 实现人类监督

# 使用流式输出配合人类监督
for chunk in agent.stream({
    "messages": [{"role": "user", "content": "请执行一个重要的操作。"}],
    "stream_mode": "messages"
}):
    if "messages" in chunk:
        message = chunk["messages"][-1]
        print(message["content"])
        
        # 在关键节点询问用户确认
        if "重要操作" in message["content"]:
            user_input = input("是否继续执行？(y/n): ")
            if user_input.lower() != 'y':
                print("操作已取消")
                break

5.2 中断和恢复机制

LangGraph 支持在执行过程中中断任务，等待用户确认后再继续：

# 中断机制示例
def human_in_the_loop_agent():
    # 执行到某个步骤时中断
    print("需要人类确认才能继续...")
    
    # 等待用户输入
    user_decision = input("请输入指令（继续/修改/取消）：")
    
    if user_decision == "继续":
        return "继续执行原有计划"
    elif user_decision == "修改":
        return "根据用户反馈调整执行计划"
    else:
        return "任务已取消"

六、构建多 Agent 协作系统

对于复杂任务，我们可以构建多个 Agent 协同工作的系统。​

6.1 多 Agent 架构模式

一种常见的多 Agent 架构是：​

• Supervisor Agent：负责任务分发和协调​

• Specialist Agent：负责具体的专业任务（如酒店预订、机票预订等）

6.2 实现多 Agent 系统

# 安装多Agent支持库
pip install langgraph-supervisor --upgrade

from langgraph_supervisor import create_supervisor

# 定义专业Agent
def hotel_agent(task):
    """酒店预订Agent"""
    return f"酒店预订Agent处理：{task}"

def flight_agent(task):
    """机票预订Agent"""  
    return f"机票预订Agent处理：{task}"

# 创建监督者Agent
supervisor = create_supervisor(
    agents={
        "hotel": hotel_agent,
        "flight": flight_agent
    },
    model=llm
)

# 测试多Agent协作
response = supervisor.invoke({
    "messages": [{"role": "user", "content": "我需要预订下周去北京的机票和酒店。"}]
})

print(response)

七、LangGraph Agent 的核心优势

7.1 完整的功能封装​

LangGraph 为 Agent 开发提供了完整的功能封装：​

• 模型集成：支持主流大语言模型​

• 工具系统：统一的工具调用接口​

• 记忆管理：短期和长期记忆机制​

• 错误处理：完善的异常处理能力

7.2 灵活的工作流定义​

使用 Graph 图的方式定义 Agent 的工作流程，让复杂逻辑变得清晰易懂：

from langgraph.graph import StateGraph, MessagesState

# 定义状态图工作流
workflow = StateGraph(MessagesState)

# 添加节点
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)

# 设置条件边
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "tools",
        "end": "__end__"
    }
)

# 设置入口点
workflow.set_entry_point("agent")

# 编译工作流
app = workflow.compile()

7.3 易于扩展和定制

LangGraph 提供了丰富的扩展点，让开发者可以根据需求定制 Agent 的行为：

# 自定义状态类
class CustomState(MessagesState):
    user_id: str
    session_id: str
    preferences: dict = {}

# 自定义工具
@tool
def personalized_recommendation(state: CustomState):
    """基于用户偏好的个性化推荐"""
    return f"为用户{state.user_id}推荐：{state.preferences.get('interests', [])}"

八、最佳实践建议

9.1 工具设计原则​

1. 单一职责：每个工具只负责一个具体功能​
2. 清晰描述：为工具提供详细的描述和参数说明​
3. 错误处理：完善的异常处理机制​
4. 返回格式：统一的返回格式便于 Agent 处理

9.2 记忆管理策略​

1. 短期记忆：定期总结和修剪，保持对话流畅​
2. 长期记忆：合理使用命名空间，避免数据混乱​
3. 隐私保护：敏感信息加密存储，合规处理

9.3 性能优化建议​

1. 模型选择：根据任务复杂度选择合适的模型​
2. 缓存机制：合理使用缓存减少重复计算​
3. 异步处理：使用异步方式处理耗时操作​
4. 负载均衡：在生产环境中考虑负载均衡