跳到主要内容

文档索引

在以下地址获取完整的文档索引：https://docs.langchain.org.cn/llms.txt

在进一步探索之前，请使用此文件发现所有可用页面。

本指南向您展示如何创建、查看和检查线程。线程与 智能体 (assistants) 协同工作，以实现您 已部署图表 的 有状态 (stateful) 执行。

​
了解线程

线程是一个持久化的对话容器，可在多次运行之间维护状态。每当您在线程上执行一次运行 (run) 时，图表都会利用线程的当前状态处理输入，并用新信息更新该状态。 线程通过在多次运行之间保留对话历史记录和上下文来实现有状态的交互。如果没有线程，每次运行都将是无状态的，无法记忆之前的交互。线程特别适用于：
  • 多轮对话，其中智能体需要记住讨论的内容。
  • 长时间运行的任务，需要在多个步骤中维护上下文。
  • 特定于用户的状态管理，其中每个用户都有自己的对话历史记录。
该图表展示了线程如何在两次运行之间维护状态。第二次运行可以访问第一次运行的消息，从而使智能体能够理解“那明天呢？”这一语境是指第一次运行中的天气查询。
  • 线程通过唯一的线程 ID 维护持久化的对话。
  • 每次运行都会将智能体的配置应用于图表执行。
  • 状态在每次运行后更新，并为后续运行持久保存。
  • 后续运行可以访问完整的对话历史记录。
  • 智能体 定义了图表执行方式的配置（模型、提示词、工具）。在创建运行时，您可以指定 图表 ID（例如 "agent"）以使用默认智能体，或指定 智能体 ID (UUID) 以使用特定配置。
  • 线程 维护状态和对话历史。
  • 运行 (Runs) 将智能体和线程结合起来，以特定的配置和状态执行您的图表。
最佳实践：在线程（对话）中追踪运行时，请确保在所有运行（包括父运行和子运行）上都设置了 thread_id。这是确保线程过滤、Token 计数和线程级评估正常工作所必需的。

​
创建线程

要以状态持久化方式运行图表，必须首先创建一个线程

​
空线程

要创建新线程，请使用以下方法之一
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)
有关详细信息，请参阅 Python 和 JS SDK 文档，或 REST API 参考。输出：
{
  "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 方法。这将创建一个独立的线程，其历史记录在操作时与原始线程相同。
# 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"])
有关详细信息，请参阅 Python 和 JS SDK 文档，或 REST API 参考。

​
预填充状态

您可以通过在 create 方法中提供 supersteps 列表来创建一个具有任意预定义状态的线程。supersteps 描述了一系列状态更新，这些更新构成了线程的初始状态。当您想要执行以下操作时，这非常有用：
  • 创建具有现有对话历史记录的线程。
  • 从其他系统迁移对话。
  • 设置具有特定初始状态的测试场景。
  • 从之前的会话恢复对话。
有关检查点 (checkpoints) 和状态管理的更多信息，请参阅 LangGraph 持久化文档。
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)
输出
{
  "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 线程：
# Search for idle threads
# The status filter accepts: idle, busy, interrupted, error
print(await client.threads.search(status="idle", limit=1))
有关详细信息，请参阅 Python 和 JS SDK 文档，或 REST API 参考。输出：
[
  {
    "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 job)。
您还可以过滤在创建或更新线程时附加的任何自定义元数据。

​
按图表过滤

print(await client.threads.search(metadata={"graph_id": "agent"}, limit=1))
输出
[
  {
    "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": {}
    }
  }
]

​
按智能体过滤

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

​
按定时任务过滤

print(await client.threads.search(
    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 方法。
# 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"])))
输出
{
  "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 方法。这将返回当前值、接下来要执行的节点以及检查点信息。
# 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"])))
输出
{
  "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：
# 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
然后使用该检查点 ID 获取特定点的状态
# 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
)

​
检查完整线程历史

要查看线程的历史记录，请使用 get_history 方法。这将返回线程经历过的每一种状态的列表，使您能够追踪完整的执行路径。
# 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']}")
此方法特别适用于：
  • 通过查看状态演变来调试执行流。
  • 理解图表执行过程中的决策点。
  • 审核对话历史和状态变更。
  • 重放或分析过去的交互。
有关详细信息，请参阅 Python 和 JS SDK 文档，或 REST API 参考。
将这些文档连接到 Claude、VSCode 等，以获得实时答案。
在 GitHub 上编辑此页面 或 提交问题。
© . This site is unofficial and not affiliated with LangChain, Inc.