LANGGRAPH MODULE 6

1 两种连接方式：SDK vs RemoteGraph

LangGraph Platform 部署运行后，可通过两种方式从客户端连接：

方式一：LangGraph SDK（推荐）

from langgraph_sdk import get_client

# 连接到本地部署（CLI 方式）
url_for_cli_deployment = "http://localhost:8123"
client = get_client(url=url_for_cli_deployment)

# 或连接到 LangGraph Cloud 托管部署
url_for_cloud = "https://langchain-academy-xxx.langgraph.app"
client = get_client(url=url_for_cloud)

SDK 提供完整的 Runs、Threads、Store 管理接口，是与 LangGraph Server API 交互的标准方式。

方式二：RemoteGraph（适合 LangGraph 内部调用）

from langgraph.pregel.remote import RemoteGraph

# RemoteGraph 可以像本地图一样调用，适合在另一个图中嵌套远程图
graph_name = "task_maistro"
remote_graph = RemoteGraph(graph_name, url="http://localhost:8123")

# 用法与本地图完全相同：invoke / stream / get_state 等
result = await remote_graph.ainvoke({"messages": [...]}, config={...})

SDK vs RemoteGraph 选择原则

用 SDK 管理基础设施（创建线程、查询运行状态、操作 Store）；用 RemoteGraph 当你需要在另一个 LangGraph 图中调用远程图，保持调用接口的一致性。

2 三大核心概念：Runs、Threads、Store

⚡ Runs（运行）

图的单次执行。每次客户端发起请求，服务器分配唯一 run_id，结果持久化到 PostgreSQL。支持后台运行、阻塞等待、流式输出三种模式。

🧵 Threads（线程）

多轮交互的容器。相同 thread_id 的多个 Run 共享同一对话历史（checkpoints），支持状态查询、分叉复制和时间旅行。

🗄️ Store（存储）

跨线程长期记忆。与 Module 5 的 InMemoryStore 接口相同，但由 PostgreSQL 后端持久化，支持 search / put / delete。

3 Runs：三种执行模式

模式一：后台运行（Fire and Forget）

# 创建线程
thread = await client.threads.create()

# 启动后台运行（立即返回，不等待结果）
config = {"configurable": {"user_id": "Test"}}
run = await client.runs.create(
    thread["thread_id"],
    "task_maistro",
    input={"messages": [HumanMessage(content="Add a ToDo to book travel to HK")]},
    config=config
)
# run["status"] 为 "pending"，继续异步执行中

模式二：阻塞等待（Blocking）

# 等待 run 完成（阻塞当前协程）
await client.runs.join(thread["thread_id"], run["run_id"])

# 此时 status 变为 "success"
result = await client.runs.get(thread["thread_id"], run["run_id"])
print(result["status"])  # "success"

模式三：流式输出（Streaming）

# stream_mode="messages-tuple" → 逐 token 流式返回
async for chunk in client.runs.stream(
    thread["thread_id"],
    "task_maistro",
    input={"messages": [HumanMessage(content="What ToDo should I focus on first?")]},
    config=config,
    stream_mode="messages-tuple"
):
    if chunk.event == "messages":
        # 实时打印 AI 生成的每个 token
        print("".join(
            data_item['content'] for data_item in chunk.data
            if 'content' in data_item
        ), end="", flush=True)

流式输出的底层机制

流式 run 执行时：Queue Worker 处理图并将更新发布到 Redis；HTTP Worker 订阅 Redis 并将更新实时推送给客户端。Redis 在这里充当消息队列，实现了图执行与 HTTP 响应的解耦，是流式输出能力的核心基础设施。

模式	方法	适用场景
后台运行	`client.runs.create()`	长耗时任务，不需要立即拿到结果
阻塞等待	`client.runs.create()` + `join()`	需要等待完成再进行后续处理
流式输出	`client.runs.stream()`	用户交互场景，需要实时反馈

4 Threads：多轮交互与状态管理

获取线程状态

# 获取线程的当前状态（最新 checkpoint）
thread_state = await client.threads.get_state(thread['thread_id'])

# 访问消息历史
from langchain_core.messages import convert_to_messages
for m in convert_to_messages(thread_state['values']['messages']):
    m.pretty_print()

# 获取完整历史（所有 checkpoints）
states = await client.threads.get_history(thread['thread_id'])

复制线程（Fork）

# 复制线程：保留历史，但创建独立的新线程（不影响原线程）
copied_thread = await client.threads.copy(thread['thread_id'])

# 复制后的线程包含完整的历史消息
copied_state = await client.threads.get_state(copied_thread['thread_id'])

线程复制的用途

Fork 线程常用于 A/B 测试（给同一历史的两个版本发送不同消息，对比结果），或在调试时保留原始对话不受实验影响。

5 Threads：分叉与 Human-in-the-Loop

通过 get_history + update_state + checkpoint_id，可以实现生产环境中的时间旅行和人工干预。

# Step 1：获取线程历史，找到要分叉的 checkpoint
states = await client.threads.get_history(thread['thread_id'])
to_fork = states[-2]  # 选择倒数第二个 checkpoint

# 查看该 checkpoint 的关键信息
to_fork['checkpoint_id']   # '1efa2c00-6609-67ff-8000-491b1dcf8129'
to_fork['next']            # ['task_mAIstro'] — 下一步执行的节点

# Step 2：修改该 checkpoint 的状态（覆写同一消息 ID，而非追加）
forked_input = {
    "messages": HumanMessage(
        content="Give me a summary of all ToDos that need to be done in the next week.",
        id=to_fork['values']['messages'][0]['id']  # 复用原消息 ID → 覆写而非追加
    )
}
forked_config = await client.threads.update_state(
    thread["thread_id"],
    forked_input,
    checkpoint_id=to_fork['checkpoint_id']
)

# Step 3：从新 checkpoint 恢复执行（input=None 表示使用 checkpoint 中的状态）
async for chunk in client.runs.stream(
    thread["thread_id"],
    graph_name,
    input=None,
    config=config,
    checkpoint_id=forked_config['checkpoint_id'],  # 指定从哪个 checkpoint 开始
    stream_mode="messages-tuple"
):
    ...

消息 ID 覆写的关键机制

LangGraph 的 add_messages reducer 规则：传入消息时，若 ID 与已有消息相同则覆写，否则追加。在 update_state 中复用原消息 ID，就能替换历史中的那条消息，而不是在末尾追加新消息，实现精确的状态编辑。

分叉后执行结果（聚焦"下周内"的任务）

AI: Here's a summary of your ToDos that need to be done in the next week:

1. Finish booking travel to Hong Kong

Status: Not started | Deadline: November 22, 2024

Estimated Time: 120 minutes

← "Call parents" 没有 deadline，不在"下周内"范围内，AI 正确过滤

6 Store：跨线程记忆的 CRUD 操作

SDK 提供 client.store 接口，允许在图之外直接操作 Store 中的长期记忆，与图内部 store.put/search/get 调用效果相同。

search_items：按命名空间检索

# task_maistro 将 ToDo 存储在 ("todo", "general", user_id) 命名空间
items = await client.store.search_items(
    ("todo", "general", "Test"),
    limit=5,
    offset=0
)

# 返回示例
# [{'value': {'task': 'Finish booking travel to Hong Kong',
#             'status': 'not started', 'deadline': '2024-11-22T23:59:59',
#             'solutions': [...], 'time_to_complete': 120},
#   'key': '18524803-c182-49de-9b10-08ccb0a06843',
#   'namespace': ['todo', 'general', 'Test'],
#   'created_at': '2024-11-14T19:37:41.664827+00:00', ...}]

put_item：直接写入记忆

from uuid import uuid4

# 在图外直接写入 Store（不经过图逻辑）
await client.store.put_item(
    ("testing", "Test"),         # namespace
    key=str(uuid4()),               # 唯一 key
    value={"todo": "Test SDK put_item"}  # value 必须是字典
)

delete_item：删除记忆

# 按 namespace + key 精确删除
await client.store.delete_item(
    ("testing", "Test"),
    key="3de441ba-8c79-4beb-8f52-00e4dcba68d4"
)

SDK Store 与图内 Store 的关系

两者操作的是同一个 PostgreSQL 后端。图内节点调用 store.put(namespace, key, value)，SDK 调用 client.store.put_item(namespace, key=..., value=...)——不同的调用接口，相同的数据源。这让你可以在不触发图执行的情况下，直接管理跨线程记忆。

命名空间规范（task_mAIstro 的约定）

数据类型	命名空间	key 格式
用户 ToDo	`("todo", "general", user_id)`	UUID（每条任务独立）
用户 Profile	`("profile", user_id)`	固定 key（单条画像）
操作指令	`("instructions", user_id)`	`"user_instructions"`