LangGraph

概述

LangGraph 为任何长时间运行的有状态工作流或智能体提供底层基础设施支持。LangGraph 不会对提示词（Prompts）或架构进行抽象，并提供以下核心优势：

• 持久化执行：构建能够从故障中恢复并可长时间运行的智能体，从中断处继续执行。
• 人机回环：通过在任何时间点检查和修改智能体状态，引入人工监管。
• 全面的记忆能力：创建具有状态的智能体，既包含用于当前推理的短期工作记忆，也包含跨会话的长期记忆。
• 使用 LangSmith 进行调试：通过可视化工具深入了解复杂的智能体行为，追踪执行路径、捕获状态转换并提供详细的运行时指标。
• 生产级部署：凭借专为处理有状态、长运行工作流挑战而设计的可扩展基础设施，自信地部署复杂的智能体系统。

# 一个简单的程序
from langgraph.graph import StateGraph, MessagesState, START, END

def mock_llm(state: MessagesState):
    return {"messages": [{"role": "ai", "content": "hello world"}]}

graph = StateGraph(MessagesState)
graph.add_node(mock_llm)
graph.add_edge(START, "mock_llm")
graph.add_edge("mock_llm", END)
graph = graph.compile()

response = graph.invoke({"messages": [{"role": "user", "content": "hi!"}]})
print(response)
# {'messages': 
#   [HumanMessage(content='hi!', additional_kwargs={}, response_metadata={}, id='bfdf4c2a-2f21-4097-bc49-71342a26fefa'), 
#    AIMessage(content='hello world', additional_kwargs={}, response_metadata={}, id='0bb9448e-b4b1-4763-8e86-ff57ac37c311')]}

快速入门

• 将智能体定义为由节点（Nodes）和边（Edges）组成的图，使用 Graph API

• 如果你偏好将智能体定义为单个函数，使用 Functional API

graph API定义

# 1. 定义模型与工具
from langchain.chat_models import init_chat_model
from langchain.tools import tool

model = init_chat_model("openai:qwen-flash",
    api_key="sk-XXXXXXX",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    extra_body={"enable_thinking": False}
)

@tool
def multiply(a: int, b: int) -> int:
    """将 a 和 b 相乘。"""
    return a * b

@tool
def add(a: int, b: int) -> int:
    """将 a 和 b 相加。"""
    return a + b

@tool
def divide(a: int, b: int) -> float:
    """将 a 除以 b。"""
    return a / b

tools = [add, multiply, divide]
tools_by_name = {tool.name: tool for tool in tools}
model_with_tools = model.bind_tools(tools)

# 2.定义状态
from langchain.messages import AnyMessage
from typing_extensions import TypedDict, Annotated
import operator

class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    llm_calls: int

[!NOTE]

Annotated 是一种增强型类型注解。它允许你在声明变量类型的同时，附加额外的元数据。

• 语法结构：Annotated[基本类型, 附加元数据1, 附加元数据2, ...]
• 普通注解 vs Annotated：
  • x: list：只告诉编辑器 x 是个列表。
  • x: Annotated[list, "read-only"]：告诉编辑器 x 是列表，同时通过元数据标记它是“只读”的。

operator 是 Python 的内置模块，它将 Python 的内置运算符（如 +, -, *）封装成了可调用的函数。

• 等价关系：operator.add(a, b) 实际上执行的是 a + b。
• 针对列表的行为：在 Python 中，[1, 2] + [3, 4] 的结果是 [1, 2, 3, 4]（即列表拼接）。

在 LangGraph 中，Annotated 被赋予了特殊的使命：定义状态（State）的归约逻辑（Reducer）。

1. 无 Annotated：如果状态字段只是简单的 messages: list，LangGraph 默认执行覆盖（Overwrite）操作。新节点返回的消息会替换掉旧消息。
2. 带 Annotated：Annotated[list, operator.add] 告诉 LangGraph，当有新数据写入该字段时，请调用 operator.add(旧值, 新值)。

在构建 Agent 时，我们必须保留对话历史，模型才能理解上下文。

# 3.定义模型节点
# 在 LangGraph 中，每个节点本质上都是一个 Python 函数
from langchain.messages import SystemMessage
def llm_call(state: dict) -> dict:
    """
    Args:
        state (dict): 当前图的全局状态，预期包含:
            - "messages": list, 包含 Human, AI, Tool 类型的历史消息序列。
            - "llm_calls": int, 可选，用于记录该会话累计的模型调用次数。

    Returns:
        dict: 状态更新集，包含:
            - "messages": 包装在 list 中的新生成消息对象（AI 响应）。
            - "llm_calls": 更新后的累计调用计数，用于配额监控或死循环熔断。
    """
    # 系统提示词
    system_prompt = SystemMessage(content="你是一个负责执行算术运算的助手。")
    # 构造完整上下文
    full_context = [system_prompt] + state["messages"]
    # 获取AI响应
    response = model_with_tools.invoke(full_context)
    # 增加模型调用次数
    current_calls = state.get('llm_calls', 0) + 1
    return {
        "messages": [response],
        "llm_calls": current_calls
    }

# 4.定义工具节点
from langchain.messages import ToolMessage
def tool_node(state: dict[str, any]) -> dict[str, list[ToolMessage]]:
    """
    Args:
        state (dict): 图的当前全局状态，需包含消息历史。

    Returns:
        dict: 状态增量字典，包含本次执行生成的 ToolMessage 序列。
    """
    # 存储所有工具执行结果的消息列表
    tool_outputs = []
    
    # 提取决策信息：从对话历史的最后一条消息中获取工具调用指令
    last_ai_message = state["messages"][-1]
    instructions = last_ai_message.tool_calls
    
    # 顺序分发与执行：遍历每一个工具调用请求
    for call in instructions:
        # 根据名称从注册表查找具象工具函数
        target_tool = tools_by_name[call["name"]]
        # 将 AI 提取的参数注入工具并获得返回值（Observation）
        execution_result = target_tool.invoke(call["args"])
        # 构建 ToolMessage，并使用 tool_call_id 实现请求与结果的强绑定
        observation_message = ToolMessage(
            content=str(execution_result), 
            tool_call_id=call["id"]
        )
        
        tool_outputs.append(observation_message)
    # 返回字典以触发全局状态的增量更新
    return {"messages": tool_outputs}

# 5.定义结束逻辑
from typing import Literal
from langgraph.graph import END

def should_continue(state: MessagesState) -> Literal["tool_node", END]:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        # 存在未处理的指令 -> 路由至工具执行节点
        return "tool_node"
    #   若无工具调用，说明模型已生成最终回复 -> 路由至全局结束符 END
    return END

# 6.编译并构建智能体
from langgraph.graph import StateGraph, START

agent_builder = StateGraph(MessagesState)

agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)

agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges(
    "llm_call",
    should_continue,
    ["tool_node", END]
)
agent_builder.add_edge("tool_node", "llm_call")

agent = agent_builder.compile()

# 7.调用
from langchain.messages import HumanMessage
messages = agent.invoke({"messages": [HumanMessage(content="3 加 4 等于多少？")]})
for m in messages["messages"]:
    m.pretty_print()

================================ Human Message =================================
3 加 4 等于多少？
================================== Ai Message ==================================
Tool Calls:
  add (call_0bcbdbed88f54014baee7e)
 Call ID: call_0bcbdbed88f54014baee7e
  Args:
    a: 3
    b: 4
================================= Tool Message =================================
7
================================== Ai Message ==================================
3 加 4 等于 7。

# 7.使用 stream 模式查看每一个步骤的状态变更
from langchain.messages import HumanMessage
input_data = {"messages": [HumanMessage(content="3乘以4再加10等于多少？")], "llm_calls": 0}

print("开始执行图任务...")
for output in agent.stream(input_data, stream_mode="updates"):
    # output 是一个字典，键是节点名称，值是该节点返回的更新
    for node_name, updated_state in output.items():
        print(f"\n--- 节点 {node_name} 执行完毕 ---")
        if "messages" in updated_state:
            new_msg = updated_state["messages"][-1]
            print(f"新增消息类型: {type(new_msg).__name__}")
            print(f"内容: {new_msg.content}")
            if hasattr(new_msg, 'tool_calls') and new_msg.tool_calls:
                print(f"包含工具调用: {new_msg.tool_calls}")

开始执行图任务...

--- 节点 llm_call 执行完毕 ---
新增消息类型: AIMessage
内容: 
包含工具调用: [{'name': 'multiply', 'args': {'a': 3, 'b': 4}, 'id': 'call_fc1c5d3fd7b34d0a9c3dae', 'type': 'tool_call'}, {'name': 'add', 'args': {'a': 12, 'b': 10}, 'id': 'call_401020e2aaf04914bc85e2', 'type': 'tool_call'}]

--- 节点 tool_node 执行完毕 ---
新增消息类型: ToolMessage
内容: 22

--- 节点 llm_call 执行完毕 ---
新增消息类型: AIMessage
内容: 3乘以4再加10等于22。

[!CAUTION]

需要补充 Function API定义