通过 LangGraph SDK 快速创建、版本化和管理 Assistant——深度讲解如何用同一张图、不同 configurable 参数,打造多套相互隔离的业务助手。
Assistant 是 LangGraph Platform 提供的一个概念:它将"图(Graph)"与"配置(Configuration)"绑定在一起,持久化存储到 PostgreSQL,并分配一个唯一的 assistant_id。
一句话理解:同一张图 + 不同 config = 不同的 Assistant。
task_mAIstro 应用使用同一张图,通过不同 todo_category("personal" / "work")和 task_maistro_role 创建两个独立的 ToDo 助手,互不干扰。
Assistants 依赖图内部定义的 configurable 字段。task_mAIstro 图在 configuration.py 中声明了三个可配置字段,并在节点内读取它们:
# configuration.py(简化)
from dataclasses import dataclass, field
from langgraph.config import get_configurable
@dataclass
class Configuration:
user_id: str = "default_user"
todo_category: str = "general" # "personal" | "work" | ...
task_maistro_role: str = "" # 系统提示词中的角色说明
# 在图节点内读取
config = Configuration(**get_configurable())
图本身不固化业务逻辑,而是通过 configurable 字段接收运行时参数。这使得同一份代码可以在不同 Assistant 实例中产生截然不同的行为。
使用 client.assistants.create() 创建一个 Assistant,传入图名称和初始配置。
from langgraph_sdk import get_client
client = get_client(url="http://localhost:8123")
# 创建 personal 助手(version 1)
personal_assistant = await client.assistants.create(
"task_maistro", # 已部署的图名称
config={"configurable": {"todo_category": "personal"}}
)
print(personal_assistant["assistant_id"]) # uuid
print(personal_assistant["version"]) # 1
create() 返回的对象包含 assistant_id、version、config 等字段,并持久化到 PostgreSQL。
部署图时会自动生成一个默认 Assistant(version 1,空 config)。如果直接用 client.runs.stream(thread_id, "task_maistro", ...) 而不传 assistant_id,使用的就是这个默认 Assistant。
使用 client.assistants.update() 为已有 Assistant 添加新配置,每次调用都会生成一个新版本,旧版本自动保留。
task_maistro_role = """You are a friendly and organized personal task assistant.
- When providing a 'todo summary':
1. List tasks grouped by deadline (overdue, today, this week, future)
2. Highlight tasks missing deadlines
- Proactively ask for deadlines when new tasks lack them
- Tone: encouraging and supportive"""
configurations = {
"todo_category": "personal",
"user_id": "lance",
"task_maistro_role": task_maistro_role,
}
personal_assistant = await client.assistants.update(
personal_assistant["assistant_id"], # 传入已有 ID
config={"configurable": configurations}
)
print(personal_assistant["version"]) # 2
每次 update() 都不会覆盖旧版本,而是追加新版本。这为实验性配置提供了安全网——如果新的 task_maistro_role 效果不理想,可以随时切回历史版本。
task_mAIstro 的典型用例:用两个 Assistant 分别管理个人任务和工作任务,相互隔离。
todo_category: "personal"
鼓励式语气,提醒截止日期,关注生活事务(游泳课、家务、旅行等)。
todo_category: "work"
专业高效,根据任务类型建议合理工期(DevRel 1天、文档冲刺 3天等)。
user_input = "Create another ToDo: Finalize set of report generation tutorials."
thread = await client.threads.create()
async for chunk in client.runs.stream(
thread["thread_id"],
work_assistant_id, # 使用 work assistant 的 ID
input={"messages": [HumanMessage(content=user_input)]},
stream_mode="values"
):
if chunk.event == 'values':
convert_to_messages(chunk.data["messages"])[-1].pretty_print()
# 用户在同一线程回复截止日期
user_input = "OK, for this task let's get it done by next Tuesday."
async for chunk in client.runs.stream(
thread["thread_id"], # 复用同一线程,保持上下文
work_assistant_id,
input={"messages": [HumanMessage(content=user_input)]},
stream_mode="values"
):
if chunk.event == 'values':
convert_to_messages(chunk.data["messages"])[-1].pretty_print()
user_input = "Give me a todo summary."
thread = await client.threads.create()
async for chunk in client.runs.stream(
thread["thread_id"],
personal_assistant_id, # 切换到 personal assistant
input={"messages": [HumanMessage(content=user_input)]},
stream_mode="values"
):
if chunk.event == 'values':
convert_to_messages(chunk.data["messages"])[-1].pretty_print()
Personal 和 Work 两个 Assistant 各自使用不同的 todo_category,因此写入 Store 时的键不同,各自的 ToDo 列表完全独立,不会相互污染。
所有 Assistant 持久化在 PostgreSQL 中,可以随时通过 SDK 查询和删除。
assistants = await client.assistants.search()
for assistant in assistants:
print({
'assistant_id': assistant['assistant_id'],
'version': assistant['version'],
'config': assistant['config']
})
# 先创建一个临时 assistant
temp_assistant = await client.assistants.create(
"task_maistro",
config={"configurable": configurations}
)
# 删除它
await client.assistants.delete(temp_assistant["assistant_id"])
# 确认已删除
assistants = await client.assistants.search()
print(len(assistants)) # 回到删除前的数量
删除 Assistant 不会删除与其关联的线程(Thread)和会话历史。线程是独立存储的。但删除后 assistant_id 失效,无法再用它创建新 run。
# 按返回顺序分配(通常按创建时间倒序)
work_assistant_id = assistants[0]['assistant_id']
personal_assistant_id = assistants[1]['assistant_id']
在生产环境中,建议将 assistant_id 持久化到应用数据库,而不是每次从 search() 推断索引——多人协作时顺序可能不稳定。
| 维度 | 默认 Assistant | 自定义 Assistant |
|---|---|---|
| 创建方式 | 部署图时自动生成 | client.assistants.create() |
| 配置 | 空 config(使用图默认值) | 自定义 configurable 字段 |
| 版本化 | 固定 version 1 | 每次 update() 递增版本 |
| 数据隔离 | 所有调用共用一个上下文 | 通过不同 config 键隔离数据 |
| 适用场景 | 开发测试、单一场景 | 多用户、多业务线、多语言等 |
当你需要同一套图逻辑服务于多种配置时使用 Assistants:多租户 SaaS(每个客户独立配置)、角色分离(客服 vs 销售 vs 技术)、A/B 测试不同 system prompt、多语言助手(中文版 vs 英文版)。Assistants 让这些场景无需重复部署代码。
6-1 Creating:将图代码打包为 Docker 镜像,用 docker-compose 启动三服务(LangGraph Server + PostgreSQL + Redis)。
6-2 Connecting:通过 LangGraph SDK 连接服务,管理线程与运行,实现流式输出。
6-3 Double Texting:用 multitask_strategy 处理并发消息——Reject / Enqueue / Interrupt / Rollback。
6-4 Assistants:用同一张图、不同 configurable 创建多套隔离助手,支持版本化管理。