LangChain LangGraph
Module 1 Module 2 Module 3 Module 4 Module 5 Module 6
Module 6 · Production
1 6-1 Creating
2 6-2 Connecting
3 6-3 Double Texting
4 6-4 Assistant
HomeLangGraphModule 6 · Production6-4 Assistant
📓 Notebook 📖 Explained
LANGGRAPH MODULE 6 · LESSON 4

Assistants

通过 LangGraph SDK 快速创建、版本化和管理 Assistant——深度讲解如何用同一张图、不同 configurable 参数,打造多套相互隔离的业务助手。

1 什么是 Assistants 及其核心价值

Assistant 是 LangGraph Platform 提供的一个概念:它将"图(Graph)"与"配置(Configuration)"绑定在一起,持久化存储到 PostgreSQL,并分配一个唯一的 assistant_id

一句话理解:同一张图 + 不同 config = 不同的 Assistant

核心场景

task_mAIstro 应用使用同一张图,通过不同 todo_category("personal" / "work")和 task_maistro_role 创建两个独立的 ToDo 助手,互不干扰。

2 前提:图的 Configuration 设计

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 实例中产生截然不同的行为。

3 创建 Assistant(create)

使用 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_idversionconfig 等字段,并持久化到 PostgreSQL。

注意

部署图时会自动生成一个默认 Assistant(version 1,空 config)。如果直接用 client.runs.stream(thread_id, "task_maistro", ...) 而不传 assistant_id,使用的就是这个默认 Assistant。

4 版本化更新 Assistant(update)

使用 client.assistants.update() 为已有 Assistant 添加新配置,每次调用都会生成一个新版本,旧版本自动保留。

Version 1
todo_category: "personal"
Version 2(update 后)
+ user_id: "lance"
+ task_maistro_role: "..."
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 效果不理想,可以随时切回历史版本。

5 实战:Personal 与 Work 两个助手

task_mAIstro 的典型用例:用两个 Assistant 分别管理个人任务和工作任务,相互隔离。

Personal Assistant

个人任务助手

todo_category: "personal"
鼓励式语气,提醒截止日期,关注生活事务(游泳课、家务、旅行等)。

Work Assistant

工作任务助手

todo_category: "work"
专业高效,根据任务类型建议合理工期(DevRel 1天、文档冲刺 3天等)。

Work Assistant 示例:强制要求截止日期

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()
work assistant 回复(无截止日期时主动追问)

Human: Create another ToDo: Finalize set of report generation tutorials.
AI: I notice "Finalize set of report generation tutorials" doesn't have a deadline yet. Based on similar documentation tasks, this might take around 3 days. Would you like to set a deadline with this in mind?
# 用户在同一线程回复截止日期
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()
AI: I've added "Finalize set of report generation tutorials" to your work ToDo list with a deadline of next Tuesday.

Personal Assistant 示例:鼓励式 ToDo 汇总

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()
AI: Here's your personal ToDo summary:

📅 This Weekend
• Check on swim lessons for the baby

🔜 Future
• Check AmEx points for winter travel (no deadline set yet)

I notice "Check AmEx points" doesn't have a deadline. Would you like to add one so we can track it better?
关键点:数据天然隔离

Personal 和 Work 两个 Assistant 各自使用不同的 todo_category,因此写入 Store 时的键不同,各自的 ToDo 列表完全独立,不会相互污染。

6 搜索与管理 Assistants

所有 Assistant 持久化在 PostgreSQL 中,可以随时通过 SDK 查询和删除。

搜索所有 Assistants

assistants = await client.assistants.search()
for assistant in assistants:
    print({
        'assistant_id': assistant['assistant_id'],
        'version': assistant['version'],
        'config': assistant['config']
    })
搜索结果(示例)

{'assistant_id': 'abc-111', 'version': 1, 'config': {'configurable': {'todo_category': 'work', ...}}}
{'assistant_id': 'abc-222', 'version': 2, 'config': {'configurable': {'todo_category': 'personal', ...}}}
{'assistant_id': 'abc-000', 'version': 1, 'config': {}} ← 部署时自动创建的默认 assistant

删除 Assistant

# 先创建一个临时 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。

通过 search 获取 assistant_id

# 按返回顺序分配(通常按创建时间倒序)
work_assistant_id     = assistants[0]['assistant_id']
personal_assistant_id = assistants[1]['assistant_id']

在生产环境中,建议将 assistant_id 持久化到应用数据库,而不是每次从 search() 推断索引——多人协作时顺序可能不稳定。

7 架构总结与适用场景

维度默认 Assistant自定义 Assistant
创建方式 部署图时自动生成 client.assistants.create()
配置 空 config(使用图默认值) 自定义 configurable 字段
版本化 固定 version 1 每次 update() 递增版本
数据隔离 所有调用共用一个上下文 通过不同 config 键隔离数据
适用场景 开发测试、单一场景 多用户、多业务线、多语言等
何时使用 Assistants

当你需要同一套图逻辑服务于多种配置时使用 Assistants:多租户 SaaS(每个客户独立配置)、角色分离(客服 vs 销售 vs 技术)、A/B 测试不同 system prompt、多语言助手(中文版 vs 英文版)。Assistants 让这些场景无需重复部署代码。

Module 6 全链路回顾

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 创建多套隔离助手,支持版本化管理。