由 LangChain 团队提出的低级编排框架，用于构建、管理和部署长时间运行、有状态的智能体。

LangGraph 的核心概念

1. 状态（State）

记录智能体运行过程中的所有信息，包括输入数据、中间计算结果、输出内容及节点间传递的参数等。状态是图中节点间通信的核心载体，也是智能体实现 “持久记忆” 和 “断点恢复” 的基础。

2. 节点（Node）

构成 LangGraph 的基本单元，通常由 Python 函数（同步 / 异步）实现，每个节点负责特定功能（如数据处理、工具调用、逻辑判断等）。节点通过接收状态输入、处理任务并输出新状态，推动智能体流程的执行。

3. 边（Edge）

连接节点的逻辑链路，定义了节点间的执行顺序和状态传递规则。边可以基于条件（如任务成功 / 失败、数据类型判断等）或固定流程设置，决定智能体从一个节点到下一个节点的跳转逻辑。例如：
当节点 A 处理完成后，通过边直接触发节点 B；
当节点 A 返回特定结果时，通过条件边选择执行节点 C 或节点 D。

4. 图（Graph）

由节点和边构成的完整逻辑结构，是智能体工作流程的抽象表示。图的结构决定了智能体的执行路径（如串行、并行、循环、分支等），通过节点与边的协同，实现复杂任务的自动化编排（如多轮问答、代码生成迭代、多智能体协作等）

基础使用

1. 基础工作

from typing import Literal # 用于类型限制
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool # 装饰器，用于定义工具函数

from langgraph.checkpoint.memory import MemorySaver # 内存检查点，用于保存会话状态
from langgraph.graph import START,END,StateGraph,MessagesState # 状态图的起点、终点、状态图类、消息状态类
from langgraph.prebuilt import ToolNode # 定义工具节点
# 大模型
from langchain_deepseek import ChatDeepSeek
def get_api_key():
    try:
        # 打开文件并读取内容
        with open('deepseek_api_key.txt', 'r') as f:
            # 去除可能的换行符和空格
            api_key = f.read().strip()
        return api_key
    except FileNotFoundError:
        print("错误：未找到api_key.txt文件")
        return None
    except Exception as e:
        print(f"读取文件时出错：{e}")
        return None
API_KEY = get_api_key()
chat = ChatDeepSeek(
    model="deepseek-chat",
    temperature=0,
    api_key=API_KEY,
)
# 获取天气的工具函数 使用 @tool装饰器声明此函数为工具
import requests
@tool
def get_weather(loc:str):
    """
    查询即时天气函数
    Args:
        loc(str): 必要参数，字符串类型，用于表示查询天气的具体城市名称的拼音，例如查询“北京”的天气则参数为"beijing"
    Returns:
        str:查询即时天气的结果，返回结果对象类型为解析之后的JSON格式对象，并用字符串形式进行表示，其中包含了全部重要的天气信息
    """
    url = f"https://api.seniverse.com/v3/weather/daily.json?key=nmwgA6tH7k9-&location={loc}&language=zh-Hans&unit=c&start=0&days=5"
    response = requests.get(url)
    return response.json()
# 工具集
tools = [get_weather]
# 绑定工具到模型chat，经过测试有没有此代码对结果没有影响，但建议加上
chat = chat.bind_tools(tools) 

2. 路由函数

此函数相当于选择器，判断大模型是否调用工具，从而返回不同的路径

# 此方法用于路由，判断模型是否需要调用工具
def should_use_tool(state: MessagesState) -> Literal["tools", END]:
    """决定是使用工具还是直接回答"""
    last_message = state["messages"][-1]
    # 检查是否有工具调用
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return END

3. 调用大模型

当工作流执行到需要调用大模型时则调用此函数

# 大模型调用的方法
def call_model(state: MessagesState) -> MessagesState:
    """调用模型"""
    messages = state["messages"]
    response = chat.invoke(messages)
    return {"messages": [response]}

4. 构建图

1. 构建图

先创建一个图

# 创建一个 LangGraph 的状态图工作流实例，构建 LangGraph 应用的第一步，为整个工具调用流程建立了基础框架。
workflow = StateGraph(MessagesState)

2. 构建图中的节点

1. add_node的第一次参数是节点的名称，第二个参数是此节点执行时调用的函数
2. 路由函数中返回的工具字符串必须与此处定义的节点的名称相同

# 添加节点
workflow.add_node("model", call_model) # 添加模型节点model，用于调用模型
tool_node = ToolNode(tools) # 构建工具节点tool_node，这个工具节点中的工具可以解决某一类问题
workflow.add_node("tools", tool_node) # 添加工具节点tools，用于调用工具

3. 构建边

workflow.add_edge(START, "model") # 添加起始边，从START到model节点
workflow.add_conditional_edges("model", should_use_tool ) # 条件边，根据路由的返回值确定目标节点，model节点执行后要么END，要么使用tools中的工具
workflow.add_edge("tools", "model") # 添加工具节点到模型节点的边，用于工具调用后的返回

4. 保存会话

checkpointer = MemorySaver() # 创建内存检查点对象memory，用于存储对话状态

5. 将图编译成实例

# 编译有上面的边和节点组成的图，创建可执行的应用实例
app = workflow.compile(checkpointer=checkpointer) 

5. 执行

# 使用编译好的应用执行查询
result = app.invoke(
    {"messages": [HumanMessage(content="北京最近天气怎么样？")]},
    config={"configurable": {"thread_id": "123"}}  # 会话ID，用于保持上下文
)
print(result['messages'][-1].content)

6. 将构建好的图保存为文件

graph_png = app.get_graph().draw_mermaid_png()
with open('graph.png','wb') as fp:
    fp.write(graph_png)

较复杂的应用

难点：

1. 使用代理创建Agent，传入不同的system_message和名称构建了不同的Agent
2. 自定义“状态”数据类型

1. 基础工作

大模型

from langchain_core.messages import HumanMessage
from langchain_core.tools import tool # 装饰器，用于定义工具函数
# 大模型
from langchain_deepseek import ChatDeepSeek

API_KEY = get_api_key('deepseek_api_key.txt')
llm = ChatDeepSeek(
    model="deepseek-chat",
    temperature=0,
    api_key=API_KEY,
)

工具

# 工具
import requests
import json
# 互联网搜索函数
@tool
def get_search_result(question:str):
    """
    互联网搜索函数
    Args：
        question (str)：必要参数，字符串类型，用于表示在互联网上进行搜素的关键词或者搜索内容的简短描述。    
    Returns：
        str：根据参数question进行互联网搜索后的结果，其中包含了全部重要的搜索结果内容。
    """
    result = "人工智能软件市场有多大?在新的研究中,Forrester 预测,到2025 年,人工智能软件市场的规模将接近 370 亿美元。 这一数字比许多投资者和其他分析师预测的要小,Forrester 得出这一预测结果的原因包括两点。首先,大多数业务应用程序都在添加 AI 功能而没有从中获利。其次,企业为自己使用而创建的定制 AI 应用程序不会产生市场收入。 Forrester 将AI 软件划分为四大类: 1.用于通用 AI 算法和数据集的AI制造商平台; 2.用于特定 AI 功能(例如计算机视觉)的AI 促进者平台; 3.围绕 AI 构建的以 AI 为中心的应用程序和中间件工具,用于医学诊断等特殊任务; 4.注入 AI 的应用程序和中间件工具,它们通过在现有应用程序或工具类别中高级使用 AI 来区分。 最大的细分市场将是 AI 制造商平台,到2025 年将增长到 130 亿美元。基于 AI 功能的新 AI 应用(如医疗诊断和风险检测解决方案)将成为第二大市场,到2025 年将接近 100 亿美元。 人工智能软件的增长在 2020 年由于大规模的经济衰退而放缓,并在2023年之后再次放缓。与整个软件市场一样,2020 年人工智能平台、应用程序和工具的需求放缓。需求将在 2021 年恢复增长,并在 2023 年前保持高位。 但是,随着人工智能被所有软件产品所吸收,收入增长将在 2024 年趋于平缓,不再能够成为供应商之间的差异或者获得溢价。实际上,Forrester 预计,高端定价的人工智能应用程序和工具的收入在 2023 年后将开始缩水。 人工智能正在迅速成为软件的基础,就像软件成为商业的基础一样,并将继续嵌入到软件中。 因此,人工智能软件将越来越多地被现有的软件供应商嵌入到现有的软件产品中。企业将会发现,通过这些软件供应商获取人工智能功能,比使用人工智能平台和辅助工具定制自己的人工"
    return result
@tool
def send_email(query:str):
    """
    发送邮件函数
    Args：
        question (str)： 邮件发送的内容。    
    Returns：
        str： 邮件发送的结果。
    """
    return f"邮件发送成功，内容为：{query}"
    
tools = [get_search_result, send_email]

2. 状态

#定义状态：我们⾸先定义图的状态。这只是⼀个消息列表，以及⼀个⽤于跟踪最新发送者的键
# 导⼊操作符和类型注解
import operator
from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add] # 消息列表，其中保存了为了解决用户问题，所有的智能体生成的消息
    sender: str # 发送者，是哪个节点发送的

3. Agent

代理

from typing import List
from langchain.prompts.chat import ChatPromptTemplate,MessagesPlaceholder
# 从 langchain 相关模块导入聊天模型和工具基类
from langchain.chat_models.base import BaseChatModel
from langchain.tools.base import BaseTool
from langchain.agents import AgentExecutor  # 如果你需要显式指定返回类型
# 构建了一个具备工具使用能力和协作能力的 AI 代理框架
def create_agent(llm: BaseChatModel, tools: List[BaseTool], system_message: str) -> AgentExecutor:
    """创建一个代理执行器"""
    # 创建一个工具调用的提示模板
    prompt = ChatPromptTemplate.from_messages([
        ("system","""
            {system_message},
            你是一个专业的AI助手，可以与其他助手合作，
            你可以使用提供的工具来推荐或回答用户的问题。
            如果你不能完全回答用户的问题，你可以与其他助手合作来解决问题，
            它会接着你回答用户的问题，执行你能做到的，使得任务有进展。
            如果你或其他助手有最终的答案，你必须在回答中包含最终的答案，
            并在回答前加上“FINAL ANSWER”，以便团队知道任务已经完成。
            你可以使用以下工具：{tool_names}
            """),
        MessagesPlaceholder(variable_name="messages"),
    ])
    prompt = prompt.partial(tool_names=", ".join([tool.name for tool in tools]))    
    prompt = prompt.partial(system_message=system_message)
    llm = llm.bind_tools(tools)
    chain = prompt | llm
    
    return chain
   

from langchain.schema import AIMessage
# 定义智能体节点
def agent_node(state: AgentState, agent: AgentExecutor, name: str) -> AgentState:
    print('大模型接收到的数据：',state)
     # 调⽤代理
    result = agent.invoke(state)
    #将 result 转换为 AIMessage 类型，并进行进一步处理
    #使用模型的 model_dump 方法将 result 转换为字典格式，同时排除 "type" 和 "name" 字段。这通常用于序列化对象以便传输或存储。
    result = AIMessage(**result.model_dump(exclude={"type", "name"}), name=name)
    return {
        "messages": [result],
        # 由于我们有⼀个严格的⼯作流程，我们可以跟踪发送者，以便知道下⼀个传递给谁。
        "sender": name,
    }

构建节点

import functools
from langgraph.prebuilt import ToolNode # 定义工具节点
#创建搜索Agent代理对象和节点对象
research_agent = create_agent(
    llm,
    [get_search_result],
    system_message="你应该提供准确的数据供MailOpt使⽤。",
)

#创建发邮件Agent代理对象和节点对象
mail_agent = create_agent(
    llm,
    [send_email],
    system_message="你用于进行邮件发送业务实现",
)

# 创建Agent节点对象。partial：给agent_node函数的部分参数agent和name赋值，返回一个新的函数research_node
research_node = functools.partial(agent_node, agent=research_agent, name="Researcher")
# 创建Agent节点对象
mail_node = functools.partial(agent_node, agent=mail_agent, name="MailOpt")
# 添加工具节点tool_node
tool_node = ToolNode(tools) # 构建工具节点tool_node，这个工具节点中的工具可以解决某一类问题

4. 路由

#定义路由函数，决定是否继续执行
from typing import Literal
# 定义路由器函数,continue 表示代理应该继续处理消息队列中的下一条消息。
def router(state) -> Literal["call_tool", "__end__", "continue"]:
    # 这是路由器
    messages = state["messages"]
    last_message = messages[-1]
    if last_message.tool_calls:
        # 上⼀个代理正在调⽤⼯具
        return "call_tool"
    if "FINAL ANSWER" in last_message.content:
        # 任何代理决定⼯作完成
        return "__end__"
    return "continue"

5. 构建图

from langgraph.graph import START,END,StateGraph # 状态图的起点、终点、状态图类、消息状态类
from langgraph.checkpoint.memory import MemorySaver # 内存检查点，用于保存会话状态
# 创建langgraph图
workflow = StateGraph(AgentState)
# 添加节点
workflow.add_node("research_agent", research_node) 
workflow.add_node("mail_agent", mail_node) 
workflow.add_node("tools", tool_node) 
# 添加边
workflow.add_edge(START, "research_agent")
workflow.add_conditional_edges("research_agent", router,{
    "call_tool": "tools",
    "continue": "mail_agent",
    "__end__": END,
}) 
workflow.add_conditional_edges("mail_agent", router,{
    "call_tool": "tools",
    "continue": "research_agent",
    "__end__": END,
}) 
workflow.add_conditional_edges(
    "tools", 
    lambda x: x["sender"],
    {
        "Researcher": "research_agent",
        "MailOpt": "mail_agent",
    }
) 
# 持久化
checkpointer = MemorySaver() # 创建内存检查点对象memory，用于存储对话状态
# 编译⼯作流图
graph = workflow.compile()

6. 执行

#调用
events = graph.invoke(
    {
        "messages": [
            HumanMessage(
            content="2025年AI软件市场规模，归纳成100字"
            " 然后进行邮件发送。"
            " ⼀旦发送完邮件表示你完成了任务。"
            )
        ],
    }
)
events['messages'][-1].content

7. 查看中间结果

#查看中间结果
for message in events['messages']:
    print(message.content)
    print("-----------------------------------")

8. 将LangGraph保存为图

# 将⽣成的图⽚保存到⽂件
graph_png = graph.get_graph().draw_mermaid_png()
with open("collaboration.png", "wb") as f:
    f.write(graph_png)