LangGraph框架研究-基础篇

引言深受业界信赖的底层框架包括 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_TRACINGtrue并配置你的 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默认就会安装langgraphpip install -U langgraph #或者 uv add langgraph使用 LangGraph 时你通常需要访问大语言模型并定义工具。具体怎么做完全取决于你。文档中我们会采用的一种方式是使用 LangChain。请使用以下命令安装 LangChainpip 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 使用图API2.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, temperature0 ) # 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( contentYou 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(contentobservation, tool_call_idtool_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(xrayTrue).draw_mermaid_png())) # Invoke from langchain.messages import HumanMessage messages [HumanMessage(contentAdd 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 使用函数式API2.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, temperature0 ) # 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( contentYou 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(contentAdd 3 and 4.)] for chunk in agent.stream(messages, stream_modeupdates): 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_KEYlsv2...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/?baseUrlhttp://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/?baseUrlhttp://127.0.0.1:2024对于运行在自定义主机或端口的 Agent Server请更新 URL 中的baseUrl查询参数。例如如果您的服务器运行在http://myhost:3000上https://smith.langchain.com/studio/?baseUrlhttp://myhost:30002.3.7 测试接口1 安装python sdkpip install langgraph-sdk2 向助手发送一条消息无线程运行#异步 from langgraph_sdk import get_client import asyncio client get_client(urlhttp://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(fReceiving 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(urlhttp://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_modemessages-tuple, ): print(fReceiving 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 构建智能体时您首先会将其拆解为称为“节点”的独立步骤。接着您将描述每个节点的决策逻辑和流转路径。最后通过一个各节点均可读写的“共享状态”将这些节点连接起来。在本指南中我们将引导您通过构建一个客服邮件智能体的全过程来梳理这一开发思路2.4.1从您想要自动化的业务流程入手2.4.2总结与后续步骤2.5 工作流智能体