Double Texting

1 什么是 Double Texting 及为何重要

Double Texting（双重发送）指用户在当前 run 尚未完成时，向同一线程发送了新的消息。在真实的聊天应用中，这种情况极为常见：

• 用户发送消息后立刻补充信息（"等等，我想改一下..."）
• 用户对响应太慢感到不耐烦，重复发送同一条消息
• 用户在 AI 还在处理时发送了新的、更高优先级的请求

LangGraph Server 通过 multitask_strategy 参数提供四种策略，让开发者根据业务场景选择最合适的并发处理方式。

2 四种策略总览

3 策略一：Reject（拒绝）

最简单的策略：Run 1 运行时拒绝所有新请求，返回 HTTP 409 Conflict。

import httpx

thread = await client.threads.create()
config = {"configurable": {"user_id": "Test"}}

# 启动 Run 1
run = await client.runs.create(
    thread["thread_id"], "task_maistro",
    input={"messages": [HumanMessage(content="Add a ToDo to follow-up with DI Repairs.")]},
    config=config,
)

# Run 2 使用 reject 策略 → 触发 HTTPStatusError 409
try:
    await client.runs.create(
        thread["thread_id"], "task_maistro",
        input={"messages": [HumanMessage(content="Add a ToDo to mount dresser to the wall.")]},
        config=config,
        multitask_strategy="reject",
    )
except httpx.HTTPStatusError as e:
    print("Failed to start concurrent run", e)

Run 1 正常完成后，线程状态中只包含第一条 ToDo（mount dresser 未被添加）。

4 策略二：Enqueue（排队）

Run 2 不会被拒绝，而是进入等待队列。Run 1 完成后，服务器自动开始执行 Run 2。两个 run 都会执行，消息不丢失。

thread = await client.threads.create()

# Run 1：立即开始
first_run = await client.runs.create(
    thread["thread_id"], "task_maistro",
    input={"messages": [HumanMessage(content="Send Erik his t-shirt gift this weekend.")]},
    config=config,
)

# Run 2：进入队列，等待 Run 1 完成
second_run = await client.runs.create(
    thread["thread_id"], "task_maistro",
    input={"messages": [HumanMessage(content="Get cash and pay nanny for 2 weeks. Do this by Friday.")]},
    config=config,
    multitask_strategy="enqueue",
)

# 等待 Run 2 完成（Run 1 → Run 2 串行执行）
await client.runs.join(thread["thread_id"], second_run["run_id"])

5 策略三：Interrupt（中断）

Run 2 到来时，Run 1 在当前节点执行完毕后暂停（不会立即杀死）。Run 1 已完成的工作被保留到 PostgreSQL，状态变为 interrupted；Run 2 立即开始执行。

import asyncio

thread = await client.threads.create()

# Run 1 开始（查询 ToDo 摘要）
interrupted_run = await client.runs.create(
    thread["thread_id"], "task_maistro",
    input={"messages": [HumanMessage(content="Give me a summary of my ToDos due tomorrow.")]},
    config=config,
)

# 等 1 秒让 Run 1 执行一部分节点
await asyncio.sleep(1)

# Run 2 使用 interrupt 策略 → Run 1 在当前节点结束后停止
second_run = await client.runs.create(
    thread["thread_id"], "task_maistro",
    input={"messages": [HumanMessage(content="Never mind, create a ToDo to Order Ham for Thanksgiving by next Friday.")]},
    config=config,
    multitask_strategy="interrupt",
)

await client.runs.join(thread["thread_id"], second_run["run_id"])

# 确认 Run 1 状态
status = (await client.runs.get(thread["thread_id"], interrupted_run["run_id"]))["status"]
print(status)  # "interrupted"

6 策略四：Rollback（回滚）

Run 2 到来时，Run 1 被立即停止并从数据库中删除（包括其所有状态变更）。Run 2 从线程的上一个干净 checkpoint 开始执行，就好像 Run 1 从未发生过一样。

thread = await client.threads.create()

# Run 1：用户的原始意图
rolled_back_run = await client.runs.create(
    thread["thread_id"], "task_maistro",
    input={"messages": [HumanMessage(content="Add a ToDo to call to make appointment at Yoga.")]},
    config=config,
)

# Run 2：用户改变了主意
second_run = await client.runs.create(
    thread["thread_id"], "task_maistro",
    input={"messages": [HumanMessage(content="Actually, add a ToDo to drop by Yoga in person on Sunday.")]},
    config=config,
    multitask_strategy="rollback",
)

await client.runs.join(thread["thread_id"], second_run["run_id"])

# 验证 Run 1 已被删除
try:
    await client.runs.get(thread["thread_id"], rolled_back_run["run_id"])
except httpx.HTTPStatusError:
    print("Original run was correctly deleted")  # ✓

7 四种策略对比与选择指南