使用线程

本指南展示如何创建、查看和检查线程。线程与助手协作，以有状态方式执行你已部署的图。

理解线程

线程是一个持久的对话容器，它在多次运行之间维护状态。每当你在线程上执行一次运行，图会用线程的当前状态处理输入，并用新信息更新该状态。

线程通过在运行之间保留对话历史和上下文来实现有状态的交互。没有线程，每次运行都将无状态，对先前的交互没有任何记忆。线程对于以下场景特别有用：

助手需要记住讨论过内容的多轮对话。
需要在多个步骤之间维持上下文的长时间运行任务。
每个用户有自己对话历史的用户级状态管理。

该图例说明一个线程如何在两次运行之间维持状态。第二次运行可以访问第一次运行的消息，从而让助手理解 "What about tomorrow?" 的上下文指的是第一次运行的天气查询：

sequenceDiagram
    participant User
    participant Thread
    participant Assistant
    participant Graph

    Note over Thread: Thread ID: abc-123<br/>Persistent conversation

    User->>Thread: Run 1: "What's the weather?"
    Thread->>Assistant: Use Assistant Config
    Assistant->>Graph: Execute with context
    Graph-->>Thread: Update State<br/>{messages: [user_msg, ai_response]}
    Thread-->>User: Response

    Note over Thread: State persisted ✓

    User->>Thread: Run 2: "What about tomorrow?"
    Note over Thread: Previous messages<br/>still in state
    Thread->>Assistant: Use Assistant Config
    Assistant->>Graph: Execute with full history
    Graph-->>Thread: Update State<br/>{messages: [...prev, new_msgs]}
    Thread-->>User: Response with context

线程以唯一的线程 ID 维持一段持久对话。
每次运行都将助手的配置应用到图执行。
状态在每次运行后更新，并为后续运行持久保留。
较晚的运行可以访问完整的对话历史。

NOTE

助手定义你的图如何执行的配置（模型、提示、工具）。创建运行时，你可以指定一个图 ID（例如 "agent"）以使用默认助手，或指定一个助手 ID（UUID）以使用特定配置。
线程维持状态和对话历史。
运行将一个助手和一个线程结合起来，以特定配置和状态执行你的图。

TIP

最佳实践：在线程（对话）中追踪运行时，确保在所有运行上都设置了 thread_id——父运行和子运行都要。这对于线程过滤、token 计数和线程级评估正确工作是必需的。

创建一个线程

要以状态持久化运行你的图，你必须首先创建一个线程：

空线程

要创建一个新线程，使用以下之一：

PythonJavaScriptcURL

python

from langgraph_sdk import get_client

# Initialize the client with your deployment URL
client = get_client(url=<DEPLOYMENT_URL>)

# Create an empty thread
# This creates a new thread with no initial state
thread = await client.threads.create()

print(thread)

javascript

import { Client } from "@langchain/langgraph-sdk";

// Initialize the client with your deployment URL
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });

// Create an empty thread
// This creates a new thread with no initial state
const thread = await client.threads.create();

console.log(thread);

bash

curl --request POST \
    --url <DEPLOYMENT_URL>/threads \
    --header 'Content-Type: application/json' \
    --data '{}'

更多信息，参阅 Python 和 JS SDK 文档，或 REST API 参考。

输出：

json

{
  "thread_id": "123e4567-e89b-12d3-a456-426614174000",
  "created_at": "2025-05-12T14:04:08.268Z",
  "updated_at": "2025-05-12T14:04:08.268Z",
  "metadata": {},
  "status": "idle",
  "values": {}
}

复制线程

或者，如果你已有一个线程且你希望复制其状态，你可以使用 copy 方法。这会创建一个独立线程，其历史在操作发生时与原线程完全相同：

PythonJavaScriptcURL

python

# Copy an existing thread
# The new thread will have the same state as the original at the time of copying
copied_thread = await client.threads.copy(thread["thread_id"])

javascript

// Copy an existing thread
// The new thread will have the same state as the original at the time of copying
const copiedThread = await client.threads.copy(thread["thread_id"]);

bash

curl --request POST --url <DEPLOYMENT_URL>/threads/thread["thread_id"]/copy \
--header 'Content-Type: application/json'

更多信息，参阅 Python 和 JS SDK 文档，或 REST API 参考。

预填充状态

你可以通过向 create 方法提供一个 supersteps 列表，用任意预定义状态创建一个线程。supersteps 描述一系列建立线程初始状态的更新。当你想要以下情形时这很有用：

用现有对话历史创建一个线程。
从另一个系统迁移对话。
用特定初始状态设置测试场景。
从之前的会话恢复对话。

有关检查点和状态管理的更多信息，参阅 LangGraph 持久化文档。

PythonJavaScriptcURL

python

from langgraph_sdk import get_client

# Initialize the client
client = get_client(url=<DEPLOYMENT_URL>)

# Create a thread with pre-populated conversation history
# The supersteps define a sequence of state updates that build up the initial state
thread = await client.threads.create(
  graph_id="agent",  # Specify which graph this thread is for
  supersteps=[
    {
      updates: [
        {
          values: {},
          as_node: '__input__',  # Initial input node
        },
      ],
    },
    {
      updates: [
        {
          values: {
            messages: [
              {
                type: 'human',
                content: 'hello',
              },
            ],
          },
          as_node: '__start__',  # User's first message
        },
      ],
    },
    {
      updates: [
        {
          values: {
            messages: [
              {
                content: 'Hello! How can I assist you today?',
                type: 'ai',
              },
            ],
          },
          as_node: 'call_model',  # Assistant's response
        },
      ],
    },
  ])

print(thread)

javascript

import { Client } from "@langchain/langgraph-sdk";

// Initialize the client
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });

// Create a thread with pre-populated conversation history
// The supersteps define a sequence of state updates that build up the initial state
const thread = await client.threads.create({
    graphId: 'agent',  // Specify which graph this thread is for
    supersteps: [
    {
      updates: [
        {
          values: {},
          asNode: '__input__',  // Initial input node
        },
      ],
    },
    {
      updates: [
        {
          values: {
            messages: [
              {
                type: 'human',
                content: 'hello',
              },
            ],
          },
          asNode: '__start__',  // User's first message
        },
      ],
    },
    {
      updates: [
        {
          values: {
            messages: [
              {
                content: 'Hello! How can I assist you today?',
                type: 'ai',
              },
            ],
          },
          asNode: 'call_model',  // Assistant's response
        },
      ],
    },
  ],
});

console.log(thread);

bash

curl --request POST \
    --url <DEPLOYMENT_URL>/threads \
    --header 'Content-Type: application/json' \
    --data '{"metadata":{"graph_id":"agent"},"supersteps":[{"updates":[{"values":{},"as_node":"__input__"}]},{"updates":[{"values":{"messages":[{"type":"human","content":"hello"}]},"as_node":"__start__"}]},{"updates":[{"values":{"messages":[{"content":"Hello\u0021 How can I assist you today?","type":"ai"}]},"as_node":"call_model"}]}]}'

输出：

json

{
  "thread_id": "f15d70a1-27d4-4793-a897-de5609920b7d",
  "created_at": "2025-05-12T15:37:08.935038+00:00",
  "updated_at": "2025-05-12T15:37:08.935046+00:00",
  "metadata": {
    "graph_id": "agent"
  },
  "status": "idle",
  "config": {},
  "values": {
    "messages": [
      {
        "content": "hello",
        "additional_kwargs": {},
        "response_metadata": {},
        "type": "human",
        "name": null,
        "id": "8701f3be-959c-4b7c-852f-c2160699b4ab",
        "example": false
      },
      {
        "content": "Hello! How can I assist you today?",
        "additional_kwargs": {},
        "response_metadata": {},
        "type": "ai",
        "name": null,
        "id": "4d8ea561-7ca1-409a-99f7-6b67af3e1aa3",
        "example": false,
        "tool_calls": [],
        "invalid_tool_calls": [],
        "usage_metadata": null
      }
    ]
  }
}

列出线程

要列出线程，使用 search 方法。它会列出应用中匹配所提供过滤条件的线程：

按线程状态过滤

使用 status 字段基于其状态过滤线程。支持的值为 idle、busy、interrupted 和 error。例如，要查看 idle 线程：

PythonJavaScriptcURL

python

# Search for idle threads
# The status filter accepts: idle, busy, interrupted, error
print(await client.threads.search(status="idle", limit=1))

javascript

// Search for idle threads
// The status filter accepts: idle, busy, interrupted, error
console.log(await client.threads.search({ status: "idle", limit: 1 }));

bash

curl --request POST \
--url <DEPLOYMENT_URL>/threads/search \
--header 'Content-Type: application/json' \
--data '{"status": "idle", "limit": 1}'

更多信息，参阅 Python 和 JS SDK 文档，或 REST API 参考。

输出：

json

[
  {
    "thread_id": "cacf79bb-4248-4d01-aabc-938dbd60ed2c",
    "created_at": "2024-08-14T17:36:38.921660+00:00",
    "updated_at": "2024-08-14T17:36:38.921660+00:00",
    "metadata": {
      "graph_id": "agent"
    },
    "status": "idle",
    "config": {
      "configurable": {}
    }
  }
]

按元数据过滤

search 方法允许你按元数据过滤。这对于查找与特定图、用户或你添加到线程的自定义元数据相关联的线程很有用。

你可以过滤的常见元数据字段包括：

元数据键	描述
`graph_id`	该线程所属的图（部署）。
`assistant_id`	用于在该线程上创建运行的助手。
`langgraph_auth_user_id`	拥有该线程的已认证用户（使用自定义认证时自动设置）。
`cron_id`	在该线程上创建了运行的 cron 作业。

你也可以按你在创建或更新线程时附加的任何自定义元数据进行过滤。

按图过滤

PythonJavaScriptcURL

python

print(await client.threads.search(metadata={"graph_id": "agent"}, limit=1))

javascript

console.log(await client.threads.search({ metadata: { "graph_id": "agent" }, limit: 1 }));

bash

curl --request POST \
--url <DEPLOYMENT_URL>/threads/search \
--header 'Content-Type: application/json' \
--data '{"metadata": {"graph_id": "agent"}, "limit": 1}'

输出：

json

[
  {
    "thread_id": "cacf79bb-4248-4d01-aabc-938dbd60ed2c",
    "created_at": "2024-08-14T17:36:38.921660+00:00",
    "updated_at": "2024-08-14T17:36:38.921660+00:00",
    "metadata": {
      "graph_id": "agent"
    },
    "status": "idle",
    "config": {
      "configurable": {}
    }
  }
]

按助手过滤

PythonJavaScriptcURL

python

print(await client.threads.search(
    metadata={"assistant_id": "fe096781-5601-53d2-b2f6-0d3403f7e9ca"},
    limit=1,
))

javascript

console.log(await client.threads.search({
  metadata: { "assistant_id": "fe096781-5601-53d2-b2f6-0d3403f7e9ca" },
  limit: 1,
}));

bash

curl --request POST \
--url <DEPLOYMENT_URL>/threads/search \
--header 'Content-Type: application/json' \
--data '{"metadata": {"assistant_id": "fe096781-5601-53d2-b2f6-0d3403f7e9ca"}, "limit": 1}'

按 cron 作业过滤

PythonJavaScriptcURL

python

print(await client.threads.search(
    metadata={"cron_id": "8b98a268-e49a-4228-a0d3-1a354e3a54d0"},
    limit=10,
))

javascript

console.log(await client.threads.search({
  metadata: { "cron_id": "8b98a268-e49a-4228-a0d3-1a354e3a54d0" },
  limit: 10,
}));

bash

curl --request POST \
--url <DEPLOYMENT_URL>/threads/search \
--header 'Content-Type: application/json' \
--data '{"metadata": {"cron_id": "8b98a268-e49a-4228-a0d3-1a354e3a54d0"}, "limit": 10}'

排序

SDK 还支持使用 sort_by 和 sort_order 参数按 thread_id、status、created_at 和 updated_at 对线程排序。

检查线程

获取线程

要查看某个给定 thread_id 的特定线程，使用 get 方法：

PythonJavaScriptcURL

python

# Retrieve a specific thread by its ID
# Returns the thread metadata including status, creation time, and metadata
print((await client.threads.get(thread["thread_id"])))

javascript

// Retrieve a specific thread by its ID
// Returns the thread metadata including status, creation time, and metadata
console.log((await client.threads.get(thread["thread_id"])));

bash

curl --request GET \
--url <DEPLOYMENT_URL>/threads/thread["thread_id"] \
--header 'Content-Type: application/json'

输出：

json

{
  "thread_id": "cacf79bb-4248-4d01-aabc-938dbd60ed2c",
  "created_at": "2024-08-14T17:36:38.921660+00:00",
  "updated_at": "2024-08-14T17:36:38.921660+00:00",
  "metadata": {
    "graph_id": "agent"
  },
  "status": "idle",
  "config": {
    "configurable": {}
  }
}

更多信息，参阅 Python 和 JS SDK 文档，或 REST API 参考。

检查线程状态

要查看某个线程的当前状态，使用 get_state 方法。它返回当前值、接下来要执行的节点和检查点信息：

PythonJavaScriptcURL

python

# Get the current state of a thread
# Returns values, next nodes, tasks, checkpoint info, and metadata
print((await client.threads.get_state(thread["thread_id"])))

javascript

// Get the current state of a thread
// Returns values, next nodes, tasks, checkpoint info, and metadata
console.log((await client.threads.getState(thread["thread_id"])));

bash

curl --request GET \
--url <DEPLOYMENT_URL>/threads/thread["thread_id"]/state \
--header 'Content-Type: application/json'

输出：

json

{
  "values": {
    "messages": [
      {
        "content": "hello",
        "additional_kwargs": {},
        "response_metadata": {},
        "type": "human",
        "name": null,
        "id": "8701f3be-959c-4b7c-852f-c2160699b4ab",
        "example": false
      },
      {
        "content": "Hello! How can I assist you today?",
        "additional_kwargs": {},
        "response_metadata": {},
        "type": "ai",
        "name": null,
        "id": "4d8ea561-7ca1-409a-99f7-6b67af3e1aa3",
        "example": false,
        "tool_calls": [],
        "invalid_tool_calls": [],
        "usage_metadata": null
      }
    ]
  },
  "next": [],
  "tasks": [],
  "metadata": {
    "thread_id": "f15d70a1-27d4-4793-a897-de5609920b7d",
    "checkpoint_id": "1f02f46f-7308-616c-8000-1b158a9a6955",
    "graph_id": "agent_with_quite_a_long_name",
    "source": "update",
    "step": 1,
    "writes": {
      "call_model": {
        "messages": [
          {
            "content": "Hello! How can I assist you today?",
            "type": "ai"
          }
        ]
      }
    },
    "parents": {}
  },
  "created_at": "2025-05-12T15:37:09.008055+00:00",
  "checkpoint": {
    "checkpoint_id": "1f02f46f-733f-6b58-8001-ea90dcabb1bd",
    "thread_id": "f15d70a1-27d4-4793-a897-de5609920b7d",
    "checkpoint_ns": ""
  },
  "parent_checkpoint": {
    "checkpoint_id": "1f02f46f-7308-616c-8000-1b158a9a6955",
    "thread_id": "f15d70a1-27d4-4793-a897-de5609920b7d",
    "checkpoint_ns": ""
  },
  "checkpoint_id": "1f02f46f-733f-6b58-8001-ea90dcabb1bd",
  "parent_checkpoint_id": "1f02f46f-7308-616c-8000-1b158a9a6955"
}

更多信息，参阅 Python 和 JS SDK 文档，或 REST API 参考。

可选地，要查看线程在某个检查点处的状态，请传入检查点 ID。这对于检查线程在其执行历史中某个特定时刻的状态很有用。

首先，从线程的历史中获取检查点 ID：

PythonJavaScriptcURL

python

# Get the thread history to find checkpoint IDs
history = await client.threads.get_history(thread_id=thread["thread_id"])
checkpoint_id = history[0]["checkpoint_id"]  # Get the most recent checkpoint

javascript

// Get the thread history to find checkpoint IDs
const history = await client.threads.getHistory(thread["thread_id"]);
const checkpointId = history[0].checkpoint_id;  // Get the most recent checkpoint

bash

# Get the thread history to find checkpoint IDs
curl --request POST \
--url <DEPLOYMENT_URL>/threads/thread["thread_id"]/history \
--header 'Content-Type: application/json' \
--data '{"limit": 1}'

然后使用该检查点 ID 获取那个特定时刻的状态：

PythonJavaScriptcURL

python

# Get thread state at a specific checkpoint
# Useful for inspecting historical state or debugging
thread_state = await client.threads.get_state(
  thread_id=thread["thread_id"],
  checkpoint_id=checkpoint_id
)

javascript

// Get thread state at a specific checkpoint
// Useful for inspecting historical state or debugging
const threadState = await client.threads.getState(thread["thread_id"], checkpointId);

bash

curl --request GET \
--url <DEPLOYMENT_URL>/threads/thread["thread_id"]/state/<CHECKPOINT_ID> \
--header 'Content-Type: application/json'

检查完整线程历史

要查看线程的历史，使用 get_history 方法。它返回线程所经历的每个状态的列表，让你能够追踪完整的执行路径：

PythonJavaScriptcURL

python

# Get the full history of a thread
# Returns a list of all state snapshots from the thread's execution
history = await client.threads.get_history(
  thread_id=thread["thread_id"],
  limit=10  # Optional: limit the number of states returned
)

for state in history:
    print(f"Checkpoint: {state['checkpoint_id']}")
    print(f"Step: {state['metadata']['step']}")

javascript

// Get the full history of a thread
// Returns a list of all state snapshots from the thread's execution
const history = await client.threads.getHistory(
  thread["thread_id"],
  {
    limit: 10  // Optional: limit the number of states returned
  }
);

for (const state of history) {
  console.log(`Checkpoint: ${state.checkpoint_id}`);
  console.log(`Step: ${state.metadata.step}`);
}

bash

curl --request POST \
--url <DEPLOYMENT_URL>/threads/thread["thread_id"]/history \
--header 'Content-Type: application/json' \
--data '{"limit": 10}'

此方法对于以下方面特别有用：

通过查看状态如何演进，来调试执行流程。
理解你的图执行中的决策点。
审计对话历史和状态变更。
回放或分析过去的交互。

更多信息，参阅 Python 和 JS SDK 文档，或 REST API 参考。

数据集

创建数据集

运行评估

评估技术

定义评估目标

打分方法

实验配置

多模态评估

分析实验结果

微调

教程

常见数据类型

评估器类型

UI

SDK

框架与集成

改进评估器

反馈

追踪设置

集成

手动埋点

配置与故障排除

项目与环境设置

高级追踪技术

数据与隐私

故障排除指南

查看与管理追踪

批量导出追踪数据

轨迹视图

数据类型参考

监控与告警

在线评估器

自动化

开发你的应用

设置依赖

持久化

能力

助手

运行

并发消息处理

如何构建

鉴权与访问控制

自定义鉴权教程

服务器定制

静态加密

替换内置后端

扩展 HTTP 服务器

请求头与日志

部署指南

参考

配置

参考

提示词

创建与管理提示词

连接模型

上下文中枢

教程

全栈 Web 应用

Agent Server API

控制平面 API

计费与用量

参考

按云服务商起步

用 Terraform 部署

AWS

GCP

Azure

安装实例

配置实例

连接外部服务

平台鉴权与访问控制

管理实例

自托管可观测性

混合部署

脚本

参考

组织