（v2）迁移到 client.threads.stream()（v3）？)
LangGraph Python SDK 如何从 client.runs.stream()v2迁移到 client.threads.stream()v3【免费下载链接】langgraphBuild resilient agents.项目地址: https://gitcode.com/GitHub_Trending/la/langgraph如果你的 Python 应用目前通过client.runs.stream()v2 流式接口订阅 LangGraph API 的运行事件而你想改用新的client.threads.stream()v3 线程级流式接口——以获得类型化投影、共享 SSE 连接、子图流式或 WebSocket 传输——这篇基于 MIGRATION.md 的文章给出了可直接照做的迁移路径包括两个接口在行为上的全部差异、各典型场景重连已有 thread、并发消费多个投影、human-in-the-loop、同步客户端的对应写法以及用仓库自带集成脚本验证迁移结果的方法。v2 接口client.runs.stream()仍然完全受支持这不是强制升级迁移的动机是你需要 v3 独有的能力类型化投影typed projections、共享 SSE fan-out、子图流式thread.subgraphs/thread.subagents或 WebSocket 传输。准备条件安装 SDK。README.md 给出的安装方式是uv add langgraph-sdk有一个正在运行的 LangGraph API server。SDK 的作用就是连接一个运行中的 LangGraph API server如果你本地用langgraph-cli跑 serverget_client()会自动指向http://localhost:8123否则在创建客户端时显式传入 server 地址from langgraph_sdk import get_client # 连接远程 server 时用 get_client(urlREMOTE_URL) 初始化 client get_client(urlhttp://your-server:port)server 端启用 v3 流式协议。这是迁移中最容易漏掉的前置条件集成环境的 docker-compose.yml 中明确说明必须设置环境变量FF_OPTIMIZED_STREAMING: true否则 API 回退到遗留的 v2 流式接口client 的 v3 端点POST /threads/{id}/stream/events、/commands等不会被暴露FF_OPTIMIZED_STREAMING: true如果你的 API server 还没开启这个开关先在服务端配置上生效再改客户端代码否则 v3 代码会直接找不到对应端点。最小迁移v2 与 v3 的对照写法下面是最短的 before/after引自 MIGRATION.md示例中的 assistant id 为文档使用的agentv2 —client.runs.stream()from langgraph_sdk import get_client client get_client() thread await client.threads.create() async for chunk in client.runs.stream( thread[thread_id], agent, input{messages: [{role: user, content: hello}]}, stream_modemessages, ): print(chunk.event, chunk.data)v3 —client.threads.stream()from langgraph_sdk import get_client import asyncio client get_client() async with client.threads.stream(assistant_idagent) as thread: await thread.run.start(input{messages: [{role: user, content: hello}]}) async for stream in thread.messages: print(await stream.text)迁移时的两个直观变化thread 创建v2 需要显式client.threads.create()v3 是惰性的——thread_id省略时由客户端生成 UUIDv4server 在第一次run.start时懒创建 thread 行协议响应只带run_id、从不带thread_id。事件形态v2 迭代原始StreamPartchunk.event/chunk.datav3 迭代类型化投影thread.messages、thread.tool_calls、thread.values、thread.extensions[name]多个投影共享同一条底层连接。两代接口的完整行为差异如下表引自 MIGRATION.mdv2client.runs.stream()v3client.threads.stream()Thread creationExplicitclient.threads.create()Lazy (minted client-side if omitted)Connection per runYesNo — shared SSE for the sessionTyped projectionsNo (rawStreamPart)Yes (messages,tool_calls,values, …)Subgraph streamingNot supportedthread.subgraphs/thread.subagentsWebSocket transportNoYes (transportwebsocket, async only)Interrupt handlingManual pollingthread.interrupted/thread.run.respond()Terminal stateIncluded in streamawait thread.output迁移常见场景的对应写法重连到已有 threadv2 里续接会话通常意味着自己管理thread_id并重新订阅v3 直接在stream上下文里传入已有thread_idasync with client.threads.stream( thread_idexisting-thread-id, assistant_idagent, ) as thread: # If the run already completed, thread.output resolves immediately. result await thread.output如果该 run 已经完成thread.output会立即解析出终态 values如果还在运行它会在运行生命周期结束后解析。并发消费多个投影v3 的所有投影共享一条 SSE 连接。正确姿势是在任何一个投影结束之前就用asyncio.gather或asyncio.TaskGroup启动全部消费者——fan-out 任务会把事件并行路由给所有订阅者async with client.threads.stream(assistant_idagent) as thread: await thread.run.start(input{messages: [{role: user, content: hi}]}) async def collect_messages(): return [s async for s in thread.messages] async def collect_tool_calls(): return [c async for c in thread.tool_calls] messages, tool_calls await asyncio.gather( collect_messages(), collect_tool_calls(), )Human-in-the-loopinterruptv2 需要手动轮询检测中断v3 用thread.interrupted和thread.run.respond()async with client.threads.stream(assistant_idagent) as thread: await thread.run.start( input{messages: [{role: user, content: book a flight}]} ) # Wait for the run to pause at an interrupt node. # thread.interrupted becomes True when input.requested arrives. while not thread.interrupted: await asyncio.sleep(0.1) # Resume with a human response (unambiguous when only one interrupt is outstanding). await thread.run.respond(yes, confirm booking) result await thread.output同步客户端同步客户端镜像了异步 API去掉async/awaitfrom langgraph_sdk import get_sync_client client get_sync_client() with client.threads.stream(assistant_idagent) as thread: thread.run.start(input{messages: [{role: user, content: hello}]}) for stream in thread.messages: print(stream.text)注意同步客户端只走 SSEtransportwebsocket不受支持。可选分支WebSocket 传输v3 支持client.threads.stream(..., transportwebsocket)默认是transportsse。两个限制必须满足需要websockets14依赖SDK 的 pyproject.toml 中声明为websockets14,17仅异步客户端AsyncThreadStream可用同步客户端SyncThreadStream只有 SSE。验证迁移结果仓库在 libs/sdk-py/integration/ 下自带一套可跑的验证环境。脚本的说明以 test_values.py 头部注释为准是先在libs/sdk-py/integration/下执行docker compose up -d启动 API server然后运行脚本# 在 libs/sdk-py/integration/ 目录下先启动集成环境会拉起 postgres、redis 和 api 三个容器 # api 容器映射宿主机 2024 端口 docker compose up -d # 从 libs/sdk-py/ 目录运行集成脚本 uv run python integration/scripts/test_values.py副作用说明docker compose up -d会创建并启动 docker-compose.yml 定义的三个容器postgres 映射5443:5432、redis 映射6380:6379、api 映射2024:8000需要本机已安装 Docker。脚本内部的成功判定均可在 scripts/_common.py 中核对check_api_reachable()先请求{BASE_URL}/ok默认BASE_URL为http://localhost:2024可用环境变量LANGGRAPH_INTEGRATION_URL覆盖指向别的 server不通会直接报错退出脚本在集成图assistant id 为agent其ask_human节点会中途 interrupt上跑通 v3 流式全链路迭代thread.values投影、自动应答 interrupt 让运行继续到终态、读取thread.output最后断言终态items中包含sub确认 subgraph 确实执行过。脚本无断言错误地跑完即说明 v3 链路在当前 server 上工作正常。集成图配置中还要求了FF_OPTIMIZED_STREAMING: true见上文准备条件所以这套环境同时验证了 v3 端点是否被 server 正确暴露。要专门验证 WebSocket 分支可改跑 test_websocket.py它是test_values.py的等价版本仅在client.threads.stream()上多传transportwebsocket。迁移时需要知道的限制以下限制来自 README.md 的 Known Limitations与 v3 迁移直接相关thread.extensions[name]每次按名访问都会新开一个订阅。应把投影赋值给变量在同一会话内复用而不是多次迭代时重复索引同步流式在后台线程中驱动 lifecycle watcher长生命周期的同步会话会一直占用该线程直到上下文管理器退出重连尝试默认上限 5 次适用于共享 SSE fan-out 和 lifecycle watcher 两者持续的网络分区会在进行中的投影上以RuntimeError形式抛出。另外提醒一点行为差异thread.output是 v3 获取终态 values 的方式v2 中终态包含在流里thread_id省略时在客户端生成——如果旧代码依赖client.threads.create()返回的显式 thread 对象或依赖 v2 流中携带的终态事件迁移时要改成对应的 v3 调用。v2 接口在 v3 发布后保持不变且持续受支持见 CHANGELOG.md 的 Notes所以你可以按场景逐条迁移先迁只需要类型化投影的只读消费路径再把 interrupt、子图流式和 WebSocket 需求按上文对应写法处理每条路径用 integration 脚本同样的“启动环境 跑脚本”方式验证后再上线。【免费下载链接】langgraphBuild resilient agents.项目地址: https://gitcode.com/GitHub_Trending/la/langgraph创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 LangGraph Python SDK 如何从 client.runs.stream()（v2）迁移到 client.threads.stream()（v3）？
LangGraph Python SDK 如何从 client.runs.stream()v2迁移到 client.threads.stream()v3【免费下载链接】langgraphBuild resilient agents.项目地址: https://gitcode.com/GitHub_Trending/la/langgraph如果你的 Python 应用目前通过client.runs.stream()v2 流式接口订阅 LangGraph API 的运行事件而你想改用新的client.threads.stream()v3 线程级流式接口——以获得类型化投影、共享 SSE 连接、子图流式或 WebSocket 传输——这篇基于 MIGRATION.md 的文章给出了可直接照做的迁移路径包括两个接口在行为上的全部差异、各典型场景重连已有 thread、并发消费多个投影、human-in-the-loop、同步客户端的对应写法以及用仓库自带集成脚本验证迁移结果的方法。v2 接口client.runs.stream()仍然完全受支持这不是强制升级迁移的动机是你需要 v3 独有的能力类型化投影typed projections、共享 SSE fan-out、子图流式thread.subgraphs/thread.subagents或 WebSocket 传输。准备条件安装 SDK。README.md 给出的安装方式是uv add langgraph-sdk有一个正在运行的 LangGraph API server。SDK 的作用就是连接一个运行中的 LangGraph API server如果你本地用langgraph-cli跑 serverget_client()会自动指向http://localhost:8123否则在创建客户端时显式传入 server 地址from langgraph_sdk import get_client # 连接远程 server 时用 get_client(urlREMOTE_URL) 初始化 client get_client(urlhttp://your-server:port)server 端启用 v3 流式协议。这是迁移中最容易漏掉的前置条件集成环境的 docker-compose.yml 中明确说明必须设置环境变量FF_OPTIMIZED_STREAMING: true否则 API 回退到遗留的 v2 流式接口client 的 v3 端点POST /threads/{id}/stream/events、/commands等不会被暴露FF_OPTIMIZED_STREAMING: true如果你的 API server 还没开启这个开关先在服务端配置上生效再改客户端代码否则 v3 代码会直接找不到对应端点。最小迁移v2 与 v3 的对照写法下面是最短的 before/after引自 MIGRATION.md示例中的 assistant id 为文档使用的agentv2 —client.runs.stream()from langgraph_sdk import get_client client get_client() thread await client.threads.create() async for chunk in client.runs.stream( thread[thread_id], agent, input{messages: [{role: user, content: hello}]}, stream_modemessages, ): print(chunk.event, chunk.data)v3 —client.threads.stream()from langgraph_sdk import get_client import asyncio client get_client() async with client.threads.stream(assistant_idagent) as thread: await thread.run.start(input{messages: [{role: user, content: hello}]}) async for stream in thread.messages: print(await stream.text)迁移时的两个直观变化thread 创建v2 需要显式client.threads.create()v3 是惰性的——thread_id省略时由客户端生成 UUIDv4server 在第一次run.start时懒创建 thread 行协议响应只带run_id、从不带thread_id。事件形态v2 迭代原始StreamPartchunk.event/chunk.datav3 迭代类型化投影thread.messages、thread.tool_calls、thread.values、thread.extensions[name]多个投影共享同一条底层连接。两代接口的完整行为差异如下表引自 MIGRATION.mdv2client.runs.stream()v3client.threads.stream()Thread creationExplicitclient.threads.create()Lazy (minted client-side if omitted)Connection per runYesNo — shared SSE for the sessionTyped projectionsNo (rawStreamPart)Yes (messages,tool_calls,values, …)Subgraph streamingNot supportedthread.subgraphs/thread.subagentsWebSocket transportNoYes (transportwebsocket, async only)Interrupt handlingManual pollingthread.interrupted/thread.run.respond()Terminal stateIncluded in streamawait thread.output迁移常见场景的对应写法重连到已有 threadv2 里续接会话通常意味着自己管理thread_id并重新订阅v3 直接在stream上下文里传入已有thread_idasync with client.threads.stream( thread_idexisting-thread-id, assistant_idagent, ) as thread: # If the run already completed, thread.output resolves immediately. result await thread.output如果该 run 已经完成thread.output会立即解析出终态 values如果还在运行它会在运行生命周期结束后解析。并发消费多个投影v3 的所有投影共享一条 SSE 连接。正确姿势是在任何一个投影结束之前就用asyncio.gather或asyncio.TaskGroup启动全部消费者——fan-out 任务会把事件并行路由给所有订阅者async with client.threads.stream(assistant_idagent) as thread: await thread.run.start(input{messages: [{role: user, content: hi}]}) async def collect_messages(): return [s async for s in thread.messages] async def collect_tool_calls(): return [c async for c in thread.tool_calls] messages, tool_calls await asyncio.gather( collect_messages(), collect_tool_calls(), )Human-in-the-loopinterruptv2 需要手动轮询检测中断v3 用thread.interrupted和thread.run.respond()async with client.threads.stream(assistant_idagent) as thread: await thread.run.start( input{messages: [{role: user, content: book a flight}]} ) # Wait for the run to pause at an interrupt node. # thread.interrupted becomes True when input.requested arrives. while not thread.interrupted: await asyncio.sleep(0.1) # Resume with a human response (unambiguous when only one interrupt is outstanding). await thread.run.respond(yes, confirm booking) result await thread.output同步客户端同步客户端镜像了异步 API去掉async/awaitfrom langgraph_sdk import get_sync_client client get_sync_client() with client.threads.stream(assistant_idagent) as thread: thread.run.start(input{messages: [{role: user, content: hello}]}) for stream in thread.messages: print(stream.text)注意同步客户端只走 SSEtransportwebsocket不受支持。可选分支WebSocket 传输v3 支持client.threads.stream(..., transportwebsocket)默认是transportsse。两个限制必须满足需要websockets14依赖SDK 的 pyproject.toml 中声明为websockets14,17仅异步客户端AsyncThreadStream可用同步客户端SyncThreadStream只有 SSE。验证迁移结果仓库在 libs/sdk-py/integration/ 下自带一套可跑的验证环境。脚本的说明以 test_values.py 头部注释为准是先在libs/sdk-py/integration/下执行docker compose up -d启动 API server然后运行脚本# 在 libs/sdk-py/integration/ 目录下先启动集成环境会拉起 postgres、redis 和 api 三个容器 # api 容器映射宿主机 2024 端口 docker compose up -d # 从 libs/sdk-py/ 目录运行集成脚本 uv run python integration/scripts/test_values.py副作用说明docker compose up -d会创建并启动 docker-compose.yml 定义的三个容器postgres 映射5443:5432、redis 映射6380:6379、api 映射2024:8000需要本机已安装 Docker。脚本内部的成功判定均可在 scripts/_common.py 中核对check_api_reachable()先请求{BASE_URL}/ok默认BASE_URL为http://localhost:2024可用环境变量LANGGRAPH_INTEGRATION_URL覆盖指向别的 server不通会直接报错退出脚本在集成图assistant id 为agent其ask_human节点会中途 interrupt上跑通 v3 流式全链路迭代thread.values投影、自动应答 interrupt 让运行继续到终态、读取thread.output最后断言终态items中包含sub确认 subgraph 确实执行过。脚本无断言错误地跑完即说明 v3 链路在当前 server 上工作正常。集成图配置中还要求了FF_OPTIMIZED_STREAMING: true见上文准备条件所以这套环境同时验证了 v3 端点是否被 server 正确暴露。要专门验证 WebSocket 分支可改跑 test_websocket.py它是test_values.py的等价版本仅在client.threads.stream()上多传transportwebsocket。迁移时需要知道的限制以下限制来自 README.md 的 Known Limitations与 v3 迁移直接相关thread.extensions[name]每次按名访问都会新开一个订阅。应把投影赋值给变量在同一会话内复用而不是多次迭代时重复索引同步流式在后台线程中驱动 lifecycle watcher长生命周期的同步会话会一直占用该线程直到上下文管理器退出重连尝试默认上限 5 次适用于共享 SSE fan-out 和 lifecycle watcher 两者持续的网络分区会在进行中的投影上以RuntimeError形式抛出。另外提醒一点行为差异thread.output是 v3 获取终态 values 的方式v2 中终态包含在流里thread_id省略时在客户端生成——如果旧代码依赖client.threads.create()返回的显式 thread 对象或依赖 v2 流中携带的终态事件迁移时要改成对应的 v3 调用。v2 接口在 v3 发布后保持不变且持续受支持见 CHANGELOG.md 的 Notes所以你可以按场景逐条迁移先迁只需要类型化投影的只读消费路径再把 interrupt、子图流式和 WebSocket 需求按上文对应写法处理每条路径用 integration 脚本同样的“启动环境 跑脚本”方式验证后再上线。【免费下载链接】langgraphBuild resilient agents.项目地址: https://gitcode.com/GitHub_Trending/la/langgraph创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考