mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-09-09 19:27:54 +02:00
Mason DaughertyandGitHub
385033fd9c
Preserve LangChain package-version trace metadata when graph-bound config and invoke-time config both contribute `lc_versions`. The earlier broad nested metadata merge has been narrowed to the LangChain-owned `lc_versions` namespace, so arbitrary user metadata keeps the existing last-writer-wins behavior. ## Changes - Add a shared metadata merge path used by `merge_configs()` and `ensure_config()` so top-level metadata keys are preserved across bound and runtime configs. - Special-case only `metadata["lc_versions"]` for one-level package-version accumulation; duplicate package entries remain last-writer-wins and non-mapping values still replace. - Keep generic nested metadata maps, including user-owned `metadata["versions"]`, as replacement-only to avoid changing arbitrary metadata semantics. - Raise the `langchain-core` lower bound to `>=1.4.7` so LangGraph’s `lc_versions` handling aligns with the lc-core package-version instrumentation. - Cover both config merge helpers with tests for `lc_versions` accumulation, non-recursive replacement within the package map, generic nested metadata replacement, and defensive copying of mapping values. ## Test note The stream event assertions for `test_imp_exception` now avoid depending on leaked internal task-path metadata. With older `langchain-core`, callback metadata could be mutated by later task runs, so every task event in this test appeared to have the final task path index. That made even the `task_with_exception` start event report `metadata["langgraph_node"] == "my_task"`, which is inconsistent with the event name. `langchain-core>=1.4.6` preserves per-event metadata more accurately: the first `my_task`, `task_with_exception`, and second `my_task` report distinct task path indexes. The test now asserts the stable behavior instead: event sequence, tags, required metadata, root stream payloads, exception handling, and final outputs, without requiring the old leaked task index.
LangGraph Python SDK
This repository contains the Python SDK for interacting with the LangSmith Deployment REST API.
Quick Start
To get started with the Python SDK, install the package
pip install -U langgraph-sdk
You will need a running LangGraph API server. If you're running a server locally using langgraph-cli, SDK will automatically point at http://localhost:8123, otherwise
you would need to specify the server URL when creating a client.
from langgraph_sdk import get_client
# If you're using a remote server, initialize the client with `get_client(url=REMOTE_URL)`
client = get_client()
# List all assistants
assistants = await client.assistants.search()
# We auto-create an assistant for each graph you register in config.
agent = assistants[0]
# Start a new thread
thread = await client.threads.create()
# Start a streaming run
input = {"messages": [{"role": "human", "content": "what's the weather in la"}]}
async for chunk in client.runs.stream(thread['thread_id'], agent['assistant_id'], input=input):
print(chunk)
Known Limitations
- WebSocket transport requires
websockets>=14and is only available on the async client (AsyncThreadStream). The sync client (SyncThreadStream) uses SSE exclusively. thread.extensions[name]opens a new subscription each time the same name is accessed. Assign the projection to a variable and reuse it within a single session rather than re-indexing across multiple iterations.- Sync streaming drives the lifecycle watcher in a background thread. Long-lived sync sessions will hold that thread open until the context manager exits.
- Reconnect attempts are limited to 5 by default for both the shared SSE fan-out and the lifecycle watcher. Persistent network partitions will surface as
RuntimeErroron in-flight projections.
Thread-Centric Streaming (v3)
client.threads.stream() returns a context manager that owns the SSE session for one
thread. Typed projections — values snapshots, message streams, tool calls, custom
events — all share the same underlying connection.
from langgraph_sdk import get_client
import asyncio
client = get_client()
async with client.threads.stream(
thread_id="my-thread",
assistant_id="agent",
) as thread:
await thread.run.start(input={"messages": [{"role": "user", "content": "hi"}]})
# Start all consumers concurrently so they share one SSE connection.
async def get_messages():
return [s async for s in thread.messages]
async def get_tool_calls():
return [c async for c in thread.tool_calls]
messages, tool_calls = await asyncio.gather(get_messages(), get_tool_calls())
for stream in messages:
print(await stream.text) # accumulated text
final = await thread.output # terminal state values