引言
深受业界信赖的底层框架:包括 Klarna、Uber、摩根大通等在内的众多正在塑造智能体未来的公司都信赖 LangGraph。它是一个用于构建、管理和部署长运行、有状态智能体的底层编排框架和运行时。
LangGraph 的定位非常底层,完全专注于智能体的编排。在使用 LangGraph 之前,我们建议你首先熟悉构建智能体所需的一些组件,从模型和工具开始入手。
我们在文档中通常会使用 LangChain 组件来集成模型和工具,但使用 LangGraph 并不强制要求你必须使用 LangChain。如果你是智能体领域的新手,或者希望使用更高层级的抽象,我们建议你使用 LangChain 的智能体,因为它为常见的 LLM 和工具调用循环提供了预构建的架构。
LangGraph 专注于智能体编排所需的核心底层能力:持久化执行、流式传输、人机回环等等
1 概览
1.1 安装
pip install -U langgraph后,创建一个简单的 Hello World 示例
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() result = graph.invoke({"messages": [{"role": "user", "content": "hi!"}]}) print(result)输出
{ "messages": [ { "role": "HumanMessage", "content": "hi!", "additional_kwargs": {}, "response_metadata": {}, "id": "9ee701a7-327f-412c-82ea-256023934fab" }, { "role": "AIMessage", "content": "hello world", "additional_kwargs": {}, "response_metadata": {}, "id": "7976f85b-3cf8-426e-b8b8-bd1624e562d8", "tool_calls": [], "invalid_tool_calls": [] } ] }使用 LangSmith来追踪请求、调试智能体行为以及评估输出结果。设置LANGSMITH_TRACING=true并配置你的 API 密钥即可开始使用
1.2 核心价值
LangGraph 为任何长运行、有状态的工作流或智能体提供底层基础设施支持。它不会去抽象(屏蔽)提示词或架构细节,而是提供以下核心优势:
- 持久化执行:构建能够经受故障并长期运行的智能体,它们可以中断后从上次停止的地方继续运行。
- 人机回环:通过在任何时间点检查和修改智能体的状态,将人类的监督纳入流程中。
- 全面的记忆能力:创建有状态的智能体,既具备用于持续推理的短期工作记忆,也具备跨会话的长期记忆。
- 利用 LangSmith 进行调试:通过可视化工具深入了解复杂的智能体行为,这些工具可以追踪执行路径、捕捉状态转换并提供详细的运行时指标。
- 生产级部署:利用专为应对有状态、长运行工作流的独特挑战而设计的可扩展基础设施,自信地部署复杂的智能体系统
1..3 生态系统
虽然 LangGraph 可以独立使用,但它也能与任何 LangChain 产品无缝集成,为开发者提供了一整套构建智能体的工具。为了提升你的 LLM 应用开发体验,可以将 LangGraph 与:
- LangSmith检测:在一个地方追踪请求、评估输出并监控部署情况。使用 LangGraph 在本地进行原型开发,然后通过集成的可观测性和评估功能过渡到生产环境,从而构建更可靠的智能体系统
- LangSmith部署:通过专为长运行、有状态工作流构建的部署平台,轻松部署和扩展智能体。在团队间发现、复用、配置和共享智能体,并在 Studio 中通过可视化原型快速迭代
- LangChain:提供集成和可组合的组件,以简化 LLM 应用程序的开发。其中包含构建在 LangGraph 之上的智能体抽象
1.4 致谢
LangGraph 的灵感来源于Pregel和Apache Beam。其对外公开接口的设计则借鉴了NetworkX。LangGraph 由 LangChain 的创造者——LangChain Inc 开发,但你可以完全不依赖 LangChain 独立使用它
2 入门
2.1 安装
安装langgraph包,如果已经安装过最新版langchain,默认就会安装langgraph
pip install -U langgraph #或者 uv add langgraph使用 LangGraph 时,你通常需要访问大语言模型并定义工具。具体怎么做,完全取决于你。文档中我们会采用的一种方式是使用 LangChain。请使用以下命令安装 LangChain:
pip install -U langchain # Requires Python 3.10+ #或者 uv add langchain # Requires Python 3.10+若要使用特定的大语言模型提供商(LLM Provider)的软件包,你需要单独进行安装。请前往集成页面,查看针对特定提供商的安装说明。
2.2 快速开始
本快速入门指南将演示如何使用 LangGraph 的图 API 或函数式 API 来构建一个计算器智能体
- 如果您倾向于将智能体定义为由节点和边构成的图,请使用图 API;
- 如果您更喜欢将智能体定义为单个函数,请使用函数式 API
请安装LangChain Docs MCP 服务器,以便让您的智能体能够访问最新、最准确的 LangChain 文档和代码示例。请安装LangChain Skills,以提升您的智能体在处理 LangChain 生态相关任务时的表现
关于概念性信息,请参阅图 API 概览和函数式 API 概览
2.2.1 使用图API
2.2.1.1 定义工具和模型
在本示例中,我们将使用 Claude Sonnet 4.5 模型,并定义用于加法、乘法和除法的工具
from langchain.tools import tool from langchain.chat_models import init_chat_model model = init_chat_model( "claude-sonnet-4-6", temperature=0 ) # Define tools @tool def multiply(a: int, b: int) -> int: """Multiply `a` and `b`. Args: a: First int b: Second int """ return a * b @tool def add(a: int, b: int) -> int: """Adds `a` and `b`. Args: a: First int b: Second int """ return a + b @tool def divide(a: int, b: int) -> float: """Divide `a` and `b`. Args: a: First int b: Second int """ return a / b # Augment the LLM with tools tools = [add, multiply, divide] tools_by_name = {tool.name: tool for tool in tools} #这里要注意的式model_with_tools.invoke不会真正执行工具调用,只是输出计划 model_with_tools = model.bind_tools(tools)2.2.1.2 定义状态
图的 State(状态)用于存储消息列表和 LLM 的调用次数。LangGraph 中的 State(状态)会在智能体执行的整个过程中持续存在。带有operator.add的Annotated类型可以确保新消息是追加到现有列表中,而不是将其替换掉
from langchain.messages import AnyMessage from typing_extensions import TypedDict, Annotated import operator class MessagesState(TypedDict): messages: Annotated[list[AnyMessage], operator.add] llm_calls: int2.2.1.3 定义模型节点
模型节点用于调用 LLM,并决定是否需要调用工具
from langchain.messages import SystemMessage def llm_call(state: dict): """LLM decides whether to call a tool or not""" return { "messages": [ #上面绑定了工具的模型 model_with_tools.invoke( [ SystemMessage( content="You are a helpful assistant tasked with performing arithmetic on a set of inputs." ) ] + state["messages"] ) ], "llm_calls": state.get('llm_calls', 0) + 1 }2.2.1.4 定义工具节点
工具节点用于调用工具并返回结果
from langchain.messages import ToolMessage def tool_node(state: dict): """Performs the tool call""" result = [] for tool_call in state["messages"][-1].tool_calls: tool = tools_by_name[tool_call["name"]] observation = tool.invoke(tool_call["args"]) result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"])) return {"messages": result}2.2.1.5 定义结束逻辑
条件边函数用于根据 LLM 是否发起了工具调用,将流程路由到工具节点或结束流程
from typing import Literal from langgraph.graph import StateGraph, START, END def should_continue(state: MessagesState) -> Literal["tool_node", END]: """Decide if we should continue the loop or stop based upon whether the LLM made a tool call""" messages = state["messages"] last_message = messages[-1] # If the LLM makes a tool call, then perform an action if last_message.tool_calls: return "tool_node" # Otherwise, we stop (reply to the user) return END2.2.1.6 构建编译智能体
该智能体是使用StateGraph类构建的,并通过compile方法进行编译
# Build workflow agent_builder = StateGraph(MessagesState) # Add nodes agent_builder.add_node("llm_call", llm_call) agent_builder.add_node("tool_node", tool_node) # Add edges to connect nodes 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") # Compile the agent agent = agent_builder.compile() # Show the agent from IPython.display import Image, display display(Image(agent.get_graph(xray=True).draw_mermaid_png())) # Invoke from langchain.messages import HumanMessage messages = [HumanMessage(content="Add 3 and 4.")] messages = agent.invoke({"messages": messages}) for m in messages["messages"]: m.pretty_print()输出
<IPython.core.display.Image object> ================================ Human Message ================================= Add 3 and 4. ================================== Ai Message ================================== Tool Calls: add (3a726a88-0819-4d9b-8fe6-207f9407c56e) Call ID: 3a726a88-0819-4d9b-8fe6-207f9407c56e Args: a: 3 b: 4 ================================= Tool Message ================================= 7 ================================== Ai Message ================================== The result of adding 3 and 4 is **7**.2.2.2 使用函数式API
2.2.2.1 定义工具和模型
在本示例中,我们将使用 Claude Sonnet 4.5 模型,并定义用于加法、乘法和除法的工具
from langchain.tools import tool from langchain.chat_models import init_chat_model model = init_chat_model( "claude-sonnet-4-6", temperature=0 ) # Define tools @tool def multiply(a: int, b: int) -> int: """Multiply `a` and `b`. Args: a: First int b: Second int """ return a * b @tool def add(a: int, b: int) -> int: """Adds `a` and `b`. Args: a: First int b: Second int """ return a + b @tool def divide(a: int, b: int) -> float: """Divide `a` and `b`. Args: a: First int b: Second int """ return a / b # Augment the LLM with tools tools = [add, multiply, divide] tools_by_name = {tool.name: tool for tool in tools} model_with_tools = model.bind_tools(tools) from langgraph.graph import add_messages from langchain.messages import ( SystemMessage, HumanMessage, ToolCall, ) from langchain_core.messages import BaseMessage from langgraph.func import entrypoint, task2.2.2.2 定义模型节点
模型节点用于调用 LLM,并决定是否需要调用工具。@task装饰器用于将函数标记为智能体可执行的任务。这些任务可以在入口点函数中同步或异步调用
@task def call_llm(messages: list[BaseMessage]): """LLM decides whether to call a tool or not""" return model_with_tools.invoke( [ SystemMessage( content="You are a helpful assistant tasked with performing arithmetic on a set of inputs." ) ] + messages )2.2.2.3 定义工具节点
工具节点用于调用工具并返回结果
@task def call_tool(tool_call: ToolCall): """Performs the tool call""" tool = tools_by_name[tool_call["name"]] return tool.invoke(tool_call)2.2.2.2 定义智能体
该智能体是使用@entrypoint函数构建的
在 Functional API 中,你无需显式地定义节点和边,而是直接在单个函数内编写标准的控制流逻辑(如循环、条件判断)
@entrypoint() def agent(messages: list[BaseMessage]): model_response = call_llm(messages).result() while True: if not model_response.tool_calls: break # Execute tools tool_result_futures = [ call_tool(tool_call) for tool_call in model_response.tool_calls ] tool_results = [fut.result() for fut in tool_result_futures] messages = add_messages(messages, [model_response, *tool_results]) model_response = call_llm(messages).result() messages = add_messages(messages, model_response) return messages # Invoke messages = [HumanMessage(content="Add 3 and 4.")] for chunk in agent.stream(messages, stream_mode="updates"): print(chunk) print("\n")2.3 本地服务
本指南将向您展示如何在本地运行 LangGraph 应用程序
前提条件
开始之前,请确保您具备以下条件:一个 LangSmith API 密钥 —— 注册完全免费
2.3.1 安装langgraph CLI
# Python >= 3.11 is required. pip install -U "langgraph-cli[inmem]" #或者 # Python >= 3.11 is required. uv add "langgraph-cli[inmem]"2.3.2 创建langgraph应用
基于 new-langgraph-project-python 模板创建一个新应用。该模板展示了一个单节点应用的结构,你可以在此基础上扩展自己的业务逻辑
langgraph new path/to/your/app --template new-langgraph-project-python其他模板:如果您在使用langgraph new命令时未指定具体模板,系统将会显示一个交互式菜单,允许您从可用模板列表中进行选择。
2.3.3 安装依赖
在您的新 LangGraph 应用的根目录下,以编辑模式安装依赖,以便服务器能够使用您的本地代码更改:
cd path/to/your/app pip install -e . 或者 cd path/to/your/app uv sync2.3.4 创建.env
在 LangGraph 应用中配置环境变量时,通常需要在项目根目录下创建.env文件。具体操作步骤如下:
- 复制示例文件:找到根目录下的
.env.example文件,将其内容复制到新创建的.env文件中。 - 填写 API 密钥:在
.env文件中填入必要的 API 密钥和其他配置变量。为了简化教程或开发流程,这些变量通常直接存储在.env文件中,而不是直接在应用服务中配置。 - 注意优先级:如果同时配置了系统环境变量和
.env文件,系统环境变量的优先级通常高于.env文件中的配置
LANGSMITH_API_KEY=lsv2...2.3.5 启动agent server
启动LangGraph API本地服务
langgraph dev样例输出
INFO:langgraph_api.cli: Welcome to ╦ ┌─┐┌┐┌┌─┐╔═╗┬─┐┌─┐┌─┐┬ ┬ ║ ├─┤││││ ┬║ ╦├┬┘├─┤├─┘├─┤ ╩═╝┴ ┴┘└┘└─┘╚═╝┴└─┴ ┴┴ ┴ ┴ - 🚀 API: http://127.0.0.1:2024 - 🎨 Studio UI: https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024 - 📚 API Docs: http://127.0.0.1:2024/docs This in-memory server is designed for development and testing. For production use, please use LangSmith Deployment.langgraph dev命令会以内存模式启动 Agent Server。这种模式仅适用于开发和测试目的。若要在生产环境中使用,请部署配置了持久化存储后端的 Agent Server。有关更多信息,请参阅平台设置概览。在langchain也有说明
2.3.6 在studio测试应用
Studio 是一个专用的用户界面,您可以将其连接到 LangGraph API 服务器,以便在本地可视化、交互和调试您的应用程序。请访问langgraph dev命令输出中提供的 URL,在 Studio 中测试您的图
> - LangGraph Studio Web UI: https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024对于运行在自定义主机或端口的 Agent Server,请更新 URL 中的baseUrl查询参数。例如,如果您的服务器运行在http://myhost:3000上
https://smith.langchain.com/studio/?baseUrl=http://myhost:30002.3.7 测试接口
1 安装python sdk
pip install langgraph-sdk2 向助手发送一条消息(无线程运行)
#异步 from langgraph_sdk import get_client import asyncio client = get_client(url="http://localhost:2024") async def main(): async for chunk in client.runs.stream( None, # Threadless run "agent", # Name of assistant. Defined in langgraph.json. input={ "messages": [{ "role": "human", "content": "What is LangGraph?", }], }, ): print(f"Receiving new event of type: {chunk.event}...") print(chunk.data) print("\n\n") asyncio.run(main())#同步 from langgraph_sdk import get_sync_client client = get_sync_client(url="http://localhost:2024") for chunk in client.runs.stream( None, # Threadless run "agent", # Name of assistant. Defined in langgraph.json. input={ "messages": [{ "role": "human", "content": "What is LangGraph?", }], }, stream_mode="messages-tuple", ): print(f"Receiving new event of type: {chunk.event}...") print(chunk.data) print("\n\n")#Rest API curl -s --request POST \ --url "http://localhost:2024/runs/stream" \ --header 'Content-Type: application/json' \ --data "{ \"assistant_id\": \"agent\", \"input\": { \"messages\": [ { \"role\": \"human\", \"content\": \"What is LangGraph?\" } ] }, \"stream_mode\": \"messages-tuple\" }"2.3.2 下一步
既然您已经在本地成功运行了 LangGraph 应用,不妨通过探索部署和高级功能来更进一步:
- 部署快速入门:使用 LangSmith 部署您的 LangGraph 应用。
- LangSmith:了解 LangSmith 的核心概念。
- SDK 参考:探索 SDK API 参考文
2.4LangGraph 的思维方式
在使用 LangGraph 构建智能体时,您首先会将其拆解为称为“节点”的独立步骤。接着,您将描述每个节点的决策逻辑和流转路径。最后,通过一个各节点均可读写的“共享状态”,将这些节点连接起来。在本指南中,我们将引导您通过构建一个客服邮件智能体的全过程,来梳理这一开发思路