LangGraph 专为希望构建强大、适应性强的 AI 智能体的开发者而设计。开发者选择 LangGraph 的原因是:

可靠性和可控性。通过审核检查和人工干预审批来指导智能体行为。LangGraph 可为长时间运行的工作流持久化上下文,使您的智能体保持正常运行。

低层级和可扩展性。使用完全描述性的低层级原语构建自定义智能体,不受限制自定义的僵化抽象约束。设计可扩展的多智能体系统,其中每个智能体都为您的用例量身定制特定角色。

一流的流式传输支持。通过逐令牌流式传输和中间步骤流式传输,LangGraph 让用户实时清晰地了解智能体的推理和行动过程。

LangGraph支持两种对于构建对话代理至关重要的内存类型:

短期内存:通过在会话中维护消息历史来跟踪正在进行的对话。

长期内存:在不同会话之间存储用户特定或应用程序级别的数据。

什么是Agent?

人类在复杂的模式识别任务中表现卓越,但通常需要借助工具(如书籍、搜索引擎或计算器)来补充先验知识以得出结论。同理,生成式AI模型可通过训练使用工具获取实时信息或建议的实际动作。例如:

• 模型可利用数据库检索工具获取客户购买历史以生成个性化购物推荐

• 基于用户查询,模型可通过API调用发送邮件或完成金融交易

为实现此能力,模型需具备:

1. 外部工具集访问权限

2. 自主规划与执行任务的推理能力

这种结合推理逻辑与外部信息访问的系统,即构成智能体(Agent)。

智能体的认知架构中有三个基本组件:模型(Model)、工具(Tools)和以及一个提供指令的提示

LLM 在一个循环中运行。在每次迭代中,它会选择一个要调用的工具,提供输入,接收结果(一个观察),并利用该观察来指导下一个动作。循环会一直持续,直到满足停止条件——通常是Agent已经收集到足够的信息来响应用户时。

Agent vs. Workflow(图)

Anthropic将Agent系统划分为两类:

1. 第一类是workflow。遵循预定义的工作流,编排LLM和工具,固定代码路径。

2. Agent:此类Agent被定义为完全自主的系统,这些系统在较长时间内独立运行,可以动态地指导自身流程和工具使用的系统。通过自身的推理、规划能力,自主控制,完成任务。

注意:在LangGraph中,一切都是图,Agent是图中的一个节点。使用create_react_agent创建一个Agent,并且也得到了一个图。

以下代码片段展示了如何使用create_react_agent创建上述Agent

Python
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

model = ChatOpenAI("xxx")

def tool() -> None:
"""Testing tool."""
...

agent = create_react_agent(
model,
tools=[tool],
)

agent.get_graph().draw_mermaid_png()

LangGraph本地服务

LangGraph CLI是一个多平台命令行工具,用于在本地构建和运行 LangGraph API 服务器。生成的服务器包含您的图的所有运行、线程、助手等的 API 端点,以及运行您的代理所需的其他服务,包括用于检查点和存储的托管数据库。

安装和使用步骤

一、创建Python虚拟环境

虚拟环境的安装步骤

1. 安装好python解释器: Python >= 3.11 is required.

2. 安装虚拟环境库,在cmd中输入:

Plain Text
pip install virtualenv

3. 创建虚拟环境,在cmd中切换到需要创建虚拟环境的目录下,执行:

Plain Text
virtualenv env_name

4. 激活虚拟环境,在cmd中进入到第三步创建的 env_name/Scripts 目录下,执行:

Plain Text
activate

执行成功后,在cmd中,当前输入行前面会有 (env_name) 的前缀

在当前状态下,使用 pip 就是在虚拟环境中安装第三方库了

5. 退出虚拟环境,cmd中输入:

Plain Text
deactivate

二、安装LangGraph CLI

Python
# Python >= 3.11 is required.

pip install --upgrade "langgraph-cli[inmem]"

三、创建 LangGraph 应用

new-langgraph-project-python模板new-langgraph-project-js模板创建一个新应用。此模板演示了一个单节点应用,您可以根据自己的逻辑进行扩展。

注意:如果您使用langgraph new命令时未指定模板,将显示一个交互式菜单,允许您从可用模板列表中进行选择。

Python
langgraph new path/to/your/app --template new-langgraph-project-python

四、安装项目依赖

在您的新 LangGraph 应用的根目录下,以编辑模式安装依赖项,以便服务器使用您的本地更改。

LangGraph中,pyproject.toml 代传统的setup.py和requirements.txt.可能包含以下扩展配置:

依赖分组:如[project.optional-dependencies]定义dev(开发工具)和test(测试框架)依赖。

动态版本控制:通过requires-python = ">=3.9"指定 Python 版本兼容性。

CI/CD 集成:通过[tool.*]配置与 GitHub Actions 或 GitLab CI 的交互

Python
cd path/to/your/app
pip install -e .

五、修改graph.py的代码

Python
# 本地私有化部署的大模型
llm = ChatOpenAI(
model='qwen3-8b',
temperature=0.8,
api_key='xx',
base_url="http://localhost:6006/v1",
extra_body={'chat_template_kwargs': {'enable_thinking': False}},
)

def get_weather(city: str) -> str:
"""Get weather for a given city."""
return f"It's always sunny in {city}!"

graph = create_react_agent(
llm,
tools=[get_weather],
prompt="You are a helpful assistant"
)

六、启动LangGraph服务器

langgraph dev命令以内存模式启动 LangGraph 服务器。此模式适用于开发和测试目的。对于生产用途,请部署 LangGraph 服务器并使其能够访问持久存储后端。

Python
命令: langgraph dev

选项

点击图片可查看完整电子表格

七、测试和访问agent的API

1. LangGraph Studio 是一个专用 UI,您可以将其连接到 LangGraph API 服务器,以在本地可视化、交互和调试您的应用。通过访问langgraph dev命令输出中提供的 URL,在 LangGraph Studio 中测试您的Agent和图。

2. PythonSDK测试

Shell
pip install langgraph-sdk

 

异步测试:

Python
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())

同步测试:

Python
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")

3. JavaScript SDK测试

安装 LangGraph JS SDK

Shell
npm install @langchain/langgraph-sdk

向LangGraph服务区发送消息:

JavaScript
const { Client } = await import("@langchain/langgraph-sdk");

// only set the apiUrl if you changed the default port when calling langgraph dev
const client = new Client({ apiUrl: "http://localhost:2024"});

const streamResponse = client.runs.stream(
null, // Threadless run
"agent", // Assistant ID
{
input: {
"messages": [
{ "role": "user", "content": "What is LangGraph?"}
]
},
streamMode: "messages-tuple",
}
);

for await (const chunk of streamResponse) {
console.log(`Receiving new event of type: ${chunk.event}...`);
console.log(JSON.stringify(chunk.data));
console.log("\n\n");
}

4. REST API测试:

Shell
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\"
}"

Tool工具的定义

在构建Agent 时,您需要为其提供一个它可以使用的工具列表。除了实际调用的函数之外,工具还包括几个组件:

注意:如果工具具有精心选择的名称、描述和args_schema,模型将表现得更好。

LangChain 支持从以下对象创建工具

1. 函数;

2. LangChain Runnables;

3. 通过从 BaseTool 子类化 -- 这是最灵活的方法,它提供了最大的控制程度,但代价是需要付出更多的努力和编写更多的代码。

一、从函数创建工具

这个@tool装饰器是定义自定义工具的最简单方法。默认情况下,装饰器使用函数名称作为工具名称,但可以通过将字符串作为第一个参数传递来覆盖。此外,装饰器将使用函数的文档字符串作为工具的描述 - 因此必须提供文档字符串。请注意,@tool支持解析注释、嵌套模式和其他特性:

sglang部署Qwen-3-8B模型

SQL
# 安装sglang
pip install "sglang[all]>=0.4.6.post1"

# 启动大模型服务
python -m sglang.launch_server \
--model-path /hy-tmp/models/Qwen/Qwen3-8B \
--served-model-name qwen3-8b \
--context-length 8192 \
--trust-remote-code \
--host 0.0.0.0 \
--port 8080 \
--reasoning-parser qwen3 \
--tool-call-parser qwen25

二、从可运行对象(Runnable)创建工具

接受字符串或dict输入的 LangChain Runnables可以使用as_tool方法转换为工具,该方法允许为参数指定名称、描述和其他模式信息。

三、子类化 BaseTool

可以通过从BaseTool子类化来定义自定义工具。这提供了对工具定义的最大控制,但需要编写更多代码。

Python
class SearchArgs(BaseModel):
query: str = Field(description="需要进行网络搜索的信息。")


# 网络搜索的工具
class MySearchTool(BaseTool):
# 工具名字
name: str = "search_tool"

description: str = '搜索互联网上公开内容的工具'

return_direct: bool = False

args_schema: Type[BaseModel] = SearchArgs

def _run(self, query) -> str:
try:
print("执行我的Python中的工具,输入的参数为:", query)
response = zhipuai_client.web_search.web_search(
search_engine="search_pro",
search_query=query
)
# print(response)
if response.search_result:
return "\n\n".join([d.content for d in response.search_result])
return '没有搜索到任何内容!'
except Exception as e:
print(e)
return '没有搜索到任何内容!'

3、Agent的上下文和记忆

一、上下文

小爱AI的注册地址:https://xiaoai.plus/register?aff=3TIp

上下文包括消息列表之外的任何数据,这些数据可以影响代理行为或工具执行。这可以是:

• 运行时传入的信息,如 `user_id` 或 API 凭据。

• 多步推理过程中更新的内部状态。

• 来自先前交互的持久记忆或事实。

LangGraph 提供了三种提供上下文的主要方式:

点击图片可查看完整电子表格

您可以使用上下文来:

• 调整模型看到的系统提示

• 为工具提供必要的输入

• 在正在进行的对话中跟踪事实

Configurable

配置适用于不可变数据,如用户元数据或 API 密钥。当您有在运行期间不会更改的值时使用。

使用保留用于此目的的键"configurable"指定配置。

C++
agent.invoke(
{"messages": [{"role": "user", "content": "hi!"}]},
config={"configurable": {"user_id": "user_123"}}
)

案例:通过Configurable中掺入参数,来动态设置系统提示词

状态 AgentState(可变上下文)

状态在运行期间充当Agent的记忆,可以短期存储也可以长期存储。它保存可在执行期间演变的动态数据,例如从工具或 LLM 输出派生的值。

1、搞清楚AgentState的作用。

2、案例:(给用户发出一个祝福语句)输入username ---> config----> 工具1---> 把username修改到State中------> 工具2----->获取State的username得到最终答案。

二、记忆存储

这是一个强大的功能,允许您在多次调用中持久化代理的状态。否则,状态仅限于单次运行。

短期存储:线程级存储(会话级)

短期存储使Agent能够跟踪多轮对话。要使用它,您必须:

1. 在创建代理时提供checkpointer。checkpointer可以实现代理状态的持久性。

2. 在运行代理时在配置中提供thread_id。thread_id是对话会话的唯一标识符。

Python
# 内存: 开发环境
checkpointer = InMemorySaver()

# 在生产环境中,使用由数据库支持的检查点
DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
# 必须安装:pip install -U "psycopg[binary,pool]" langgraph-checkpoint-postgres

# 生产环境:Redis
# pip install -U langgraph-checkpoint-redis
DB_URI = "redis://:6379"
with RedisSaver.from_conn_string(DB_URI) as checkpointer:

长期存储:跨线程存储

使用长期内存来跨会话存储用户特定或应用程序特定的数据。这对于聊天机器人等应用程序非常有用,您可能希望记住用户偏好或其他信息。

要使用长期内存,您需要:

1. 配置一个存储以在调用之间持久化数据。

2. 使用get_store函数从工具或提示中访问存储。

Python
# 开发环境中: 内存
store = InMemoryStore()

# 在生产环境中,使用由数据库支持的存储
DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
with (
PostgresStore.from_conn_string(DB_URI) as store,
PostgresSaver.from_conn_string(DB_URI) as checkpointer,
):
# store.setup()
# checkpointer.setup()



DB_URI = "redis://:6379"
with (
RedisStore.from_conn_string(DB_URI) as store,
RedisSaver.from_conn_string(DB_URI) as checkpointer,
):
store.setup()
checkpointer.setup()

注意:

• 首次使用 Postgres 存储时,您需要调用store.setup(),checkpointer.setup()

• 首次使用 Redis 存储时,您需要调用store.setup(),checkpointer.setup()

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐