mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-08-28 10:49:56 +02:00
Compare commits
13
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
53172459bb | ||
|
|
df254a7f7f | ||
|
|
46bd949f53 | ||
|
|
2e7a6f088e | ||
|
|
f941f467d5 | ||
|
|
a7aaa81ed8 | ||
|
|
eb9e6e2a0b | ||
|
|
2d304b8a1b | ||
|
|
6b17b12ae4 | ||
|
|
8ccead9560 | ||
|
|
3a9749a0ed | ||
|
|
196fbf2631 | ||
|
|
0acd5decf8 |
@@ -151,6 +151,7 @@ class PoolConfig(TypedDict, total=False):
|
||||
"""Connection pool settings for PostgreSQL connections.
|
||||
|
||||
Controls connection lifecycle and resource utilization:
|
||||
|
||||
- Small pools (1-5) suit low-concurrency workloads
|
||||
- Larger pools handle concurrent requests but consume more resources
|
||||
- Setting max_size prevents resource exhaustion under load
|
||||
@@ -166,6 +167,7 @@ class PoolConfig(TypedDict, total=False):
|
||||
"""Additional connection arguments passed to each connection in the pool.
|
||||
|
||||
Default kwargs set automatically:
|
||||
|
||||
- autocommit: True
|
||||
- prepare_threshold: 0
|
||||
- row_factory: dict_row
|
||||
@@ -656,7 +658,8 @@ class PostgresStore(BaseStore, BasePostgresStore[_pg_internal.Conn]):
|
||||
item = store.get(("users", "123"), "prefs")
|
||||
```
|
||||
|
||||
Or using the convenient from_conn_string helper:
|
||||
Or using the convenient `from_conn_string` helper:
|
||||
|
||||
```python
|
||||
from langgraph.store.postgres import PostgresStore
|
||||
|
||||
|
||||
@@ -722,7 +722,8 @@ class SqliteStore(BaseSqliteStore, BaseStore):
|
||||
item = store.get(("users", "123"), "prefs")
|
||||
```
|
||||
|
||||
Or using the convenient from_conn_string helper:
|
||||
Or using the convenient `from_conn_string` helper:
|
||||
|
||||
```python
|
||||
from langgraph.store.sqlite import SqliteStore
|
||||
|
||||
|
||||
@@ -3,6 +3,7 @@ from __future__ import annotations
|
||||
import asyncio
|
||||
import enum
|
||||
import inspect
|
||||
import logging
|
||||
import sys
|
||||
import warnings
|
||||
from collections.abc import (
|
||||
@@ -25,6 +26,7 @@ from typing import (
|
||||
cast,
|
||||
)
|
||||
|
||||
from langchain_core.callbacks import AsyncCallbackManager, CallbackManager
|
||||
from langchain_core.runnables.base import (
|
||||
Runnable,
|
||||
RunnableConfig,
|
||||
@@ -61,6 +63,86 @@ try:
|
||||
except ImportError:
|
||||
_StreamingCallbackHandler = None # type: ignore
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _process_inputs(processor: Callable[[Any], Any] | None, inputs: Any) -> Any:
|
||||
"""Safely process trace inputs, returning error placeholder on failure."""
|
||||
if processor is None:
|
||||
return inputs
|
||||
try:
|
||||
return processor(inputs)
|
||||
except Exception:
|
||||
logger.exception("trace_inputs filter failed")
|
||||
return {"error": "<trace_inputs processing failed>"}
|
||||
|
||||
|
||||
def _process_outputs(processor: Callable[[Any], Any] | None, outputs: Any) -> Any:
|
||||
"""Safely process trace outputs, returning error placeholder on failure."""
|
||||
if processor is None:
|
||||
return outputs
|
||||
try:
|
||||
return processor(outputs)
|
||||
except Exception:
|
||||
logger.exception("trace_outputs filter failed")
|
||||
return {"error": "<trace_outputs processing failed>"}
|
||||
|
||||
|
||||
def _filter_langsmith_handlers(
|
||||
callback_manager: CallbackManager,
|
||||
) -> CallbackManager:
|
||||
"""Filter out LangChainTracer handlers from a callback manager.
|
||||
|
||||
Used when traceable config has enabled=False - we want to fire callbacks
|
||||
to custom handlers but skip LangSmith tracing.
|
||||
"""
|
||||
filtered_handlers = [
|
||||
h for h in callback_manager.handlers if not isinstance(h, LangChainTracer)
|
||||
]
|
||||
filtered_inheritable = [
|
||||
h
|
||||
for h in callback_manager.inheritable_handlers
|
||||
if not isinstance(h, LangChainTracer)
|
||||
]
|
||||
# Create a new callback manager with filtered handlers
|
||||
return CallbackManager(
|
||||
handlers=filtered_handlers,
|
||||
inheritable_handlers=filtered_inheritable,
|
||||
parent_run_id=callback_manager.parent_run_id,
|
||||
tags=callback_manager.tags,
|
||||
inheritable_tags=callback_manager.inheritable_tags,
|
||||
metadata=callback_manager.metadata,
|
||||
inheritable_metadata=callback_manager.inheritable_metadata,
|
||||
)
|
||||
|
||||
|
||||
def _filter_langsmith_handlers_async(
|
||||
callback_manager: AsyncCallbackManager,
|
||||
) -> AsyncCallbackManager:
|
||||
"""Filter out LangChainTracer handlers from an async callback manager.
|
||||
|
||||
Used when traceable config has enabled=False - we want to fire callbacks
|
||||
to custom handlers but skip LangSmith tracing.
|
||||
"""
|
||||
filtered_handlers = [
|
||||
h for h in callback_manager.handlers if not isinstance(h, LangChainTracer)
|
||||
]
|
||||
filtered_inheritable = [
|
||||
h
|
||||
for h in callback_manager.inheritable_handlers
|
||||
if not isinstance(h, LangChainTracer)
|
||||
]
|
||||
# Create a new callback manager with filtered handlers
|
||||
return AsyncCallbackManager(
|
||||
handlers=filtered_handlers,
|
||||
inheritable_handlers=filtered_inheritable,
|
||||
parent_run_id=callback_manager.parent_run_id,
|
||||
tags=callback_manager.tags,
|
||||
inheritable_tags=callback_manager.inheritable_tags,
|
||||
metadata=callback_manager.metadata,
|
||||
inheritable_metadata=callback_manager.inheritable_metadata,
|
||||
)
|
||||
|
||||
|
||||
def _set_config_context(
|
||||
config: RunnableConfig, run: Any = None
|
||||
@@ -397,7 +479,9 @@ class RunnableCallable(Runnable):
|
||||
else:
|
||||
run_manager.on_chain_end(ret)
|
||||
else:
|
||||
ret = self.func(*args, **kwargs)
|
||||
# Still need to set config context for get_config() to work
|
||||
with set_config_context(config, None) as context:
|
||||
ret = context.run(self.func, *args, **kwargs)
|
||||
if self.recurse and isinstance(ret, Runnable):
|
||||
return ret.invoke(input, config)
|
||||
return ret
|
||||
@@ -470,7 +554,13 @@ class RunnableCallable(Runnable):
|
||||
else:
|
||||
await run_manager.on_chain_end(ret)
|
||||
else:
|
||||
ret = await self.afunc(*args, **kwargs)
|
||||
# Still need to set config context for get_config() to work
|
||||
coro = cast(Coroutine[None, None, Any], self.afunc(*args, **kwargs))
|
||||
if ASYNCIO_ACCEPTS_CONTEXT:
|
||||
with set_config_context(config, None) as context:
|
||||
ret = await asyncio.create_task(coro, context=context)
|
||||
else:
|
||||
ret = await coro
|
||||
if self.recurse and isinstance(ret, Runnable):
|
||||
return await ret.ainvoke(input, config)
|
||||
return ret
|
||||
@@ -544,12 +634,20 @@ class RunnableSeq(Runnable):
|
||||
*steps: RunnableLike,
|
||||
name: str | None = None,
|
||||
trace_inputs: Callable[[Any], Any] | None = None,
|
||||
trace_outputs: Callable[[Any], Any] | None = None,
|
||||
trace: bool = True,
|
||||
skip_langsmith: bool = False,
|
||||
) -> None:
|
||||
"""Create a new RunnableSeq.
|
||||
|
||||
Args:
|
||||
steps: The steps to include in the sequence.
|
||||
name: The name of the `Runnable`.
|
||||
trace_inputs: Optional function to transform inputs before tracing.
|
||||
trace_outputs: Optional function to transform outputs before tracing.
|
||||
trace: Whether to trace this sequence. Defaults to True.
|
||||
skip_langsmith: If True, filter out LangChainTracer handlers but keep
|
||||
other callbacks. Used when traceable config has enabled=False.
|
||||
|
||||
Raises:
|
||||
ValueError: If the sequence has less than 2 steps.
|
||||
@@ -569,6 +667,9 @@ class RunnableSeq(Runnable):
|
||||
self.steps = steps_flat
|
||||
self.name = name
|
||||
self.trace_inputs = trace_inputs
|
||||
self.trace_outputs = trace_outputs
|
||||
self.trace = trace
|
||||
self.skip_langsmith = skip_langsmith
|
||||
|
||||
def __or__(
|
||||
self,
|
||||
@@ -625,43 +726,58 @@ class RunnableSeq(Runnable):
|
||||
) -> Any:
|
||||
if config is None:
|
||||
config = ensure_config()
|
||||
# setup callbacks and context
|
||||
callback_manager = get_callback_manager_for_config(config)
|
||||
# start the root run
|
||||
run_manager = callback_manager.on_chain_start(
|
||||
None,
|
||||
self.trace_inputs(input) if self.trace_inputs is not None else input,
|
||||
name=config.get("run_name") or self.get_name(),
|
||||
run_id=config.pop("run_id", None),
|
||||
)
|
||||
# invoke all steps in sequence
|
||||
try:
|
||||
for i, step in enumerate(self.steps):
|
||||
# mark each step as a child run
|
||||
config = patch_config(
|
||||
config, callbacks=run_manager.get_child(f"seq:step:{i + 1}")
|
||||
)
|
||||
# 1st step is the actual node,
|
||||
# others are writers which don't need to be run in context
|
||||
if i == 0:
|
||||
# get the run object
|
||||
for h in run_manager.handlers:
|
||||
if isinstance(h, LangChainTracer):
|
||||
run = h.run_map.get(str(run_manager.run_id))
|
||||
break
|
||||
|
||||
if self.trace:
|
||||
# setup callbacks and context
|
||||
callback_manager = get_callback_manager_for_config(config)
|
||||
# Filter out LangChainTracer if skip_langsmith is set
|
||||
if self.skip_langsmith:
|
||||
callback_manager = _filter_langsmith_handlers(callback_manager)
|
||||
# start the root run
|
||||
run_manager = callback_manager.on_chain_start(
|
||||
None,
|
||||
_process_inputs(self.trace_inputs, input),
|
||||
name=config.get("run_name") or self.get_name(),
|
||||
run_id=config.pop("run_id", None),
|
||||
)
|
||||
# invoke all steps in sequence
|
||||
try:
|
||||
for i, step in enumerate(self.steps):
|
||||
# mark each step as a child run
|
||||
config = patch_config(
|
||||
config, callbacks=run_manager.get_child(f"seq:step:{i + 1}")
|
||||
)
|
||||
# 1st step is the actual node,
|
||||
# others are writers which don't need to be run in context
|
||||
if i == 0:
|
||||
# get the run object
|
||||
for h in run_manager.handlers:
|
||||
if isinstance(h, LangChainTracer):
|
||||
run = h.run_map.get(str(run_manager.run_id))
|
||||
break
|
||||
else:
|
||||
run = None
|
||||
# run in context
|
||||
with set_config_context(config, run) as context:
|
||||
input = context.run(step.invoke, input, config, **kwargs)
|
||||
else:
|
||||
run = None
|
||||
# run in context
|
||||
with set_config_context(config, run) as context:
|
||||
input = context.run(step.invoke, input, config, **kwargs)
|
||||
else:
|
||||
input = step.invoke(input, config)
|
||||
# finish the root run
|
||||
except BaseException as e:
|
||||
run_manager.on_chain_error(e)
|
||||
raise
|
||||
input = step.invoke(input, config)
|
||||
# finish the root run
|
||||
except BaseException as e:
|
||||
run_manager.on_chain_error(e)
|
||||
raise
|
||||
else:
|
||||
run_manager.on_chain_end(_process_outputs(self.trace_outputs, input))
|
||||
return input
|
||||
else:
|
||||
run_manager.on_chain_end(input)
|
||||
# Still need to set config context for get_config() to work
|
||||
with set_config_context(config, None) as context:
|
||||
for i, step in enumerate(self.steps):
|
||||
input = (
|
||||
context.run(step.invoke, input, config, **kwargs)
|
||||
if i == 0
|
||||
else step.invoke(input, config)
|
||||
)
|
||||
return input
|
||||
|
||||
async def ainvoke(
|
||||
@@ -672,49 +788,75 @@ class RunnableSeq(Runnable):
|
||||
) -> Any:
|
||||
if config is None:
|
||||
config = ensure_config()
|
||||
# setup callbacks
|
||||
callback_manager = get_async_callback_manager_for_config(config)
|
||||
# start the root run
|
||||
run_manager = await callback_manager.on_chain_start(
|
||||
None,
|
||||
self.trace_inputs(input) if self.trace_inputs is not None else input,
|
||||
name=config.get("run_name") or self.get_name(),
|
||||
run_id=config.pop("run_id", None),
|
||||
)
|
||||
|
||||
# invoke all steps in sequence
|
||||
try:
|
||||
for i, step in enumerate(self.steps):
|
||||
# mark each step as a child run
|
||||
config = patch_config(
|
||||
config, callbacks=run_manager.get_child(f"seq:step:{i + 1}")
|
||||
)
|
||||
# 1st step is the actual node,
|
||||
# others are writers which don't need to be run in context
|
||||
if i == 0:
|
||||
if ASYNCIO_ACCEPTS_CONTEXT:
|
||||
# get the run object
|
||||
for h in run_manager.handlers:
|
||||
if isinstance(h, LangChainTracer):
|
||||
run = h.run_map.get(str(run_manager.run_id))
|
||||
break
|
||||
if self.trace:
|
||||
# setup callbacks
|
||||
callback_manager = get_async_callback_manager_for_config(config)
|
||||
# Filter out LangChainTracer if skip_langsmith is set
|
||||
if self.skip_langsmith:
|
||||
callback_manager = _filter_langsmith_handlers_async(callback_manager)
|
||||
# start the root run
|
||||
run_manager = await callback_manager.on_chain_start(
|
||||
None,
|
||||
_process_inputs(self.trace_inputs, input),
|
||||
name=config.get("run_name") or self.get_name(),
|
||||
run_id=config.pop("run_id", None),
|
||||
)
|
||||
|
||||
# invoke all steps in sequence
|
||||
try:
|
||||
for i, step in enumerate(self.steps):
|
||||
# mark each step as a child run
|
||||
config = patch_config(
|
||||
config, callbacks=run_manager.get_child(f"seq:step:{i + 1}")
|
||||
)
|
||||
# 1st step is the actual node,
|
||||
# others are writers which don't need to be run in context
|
||||
if i == 0:
|
||||
if ASYNCIO_ACCEPTS_CONTEXT:
|
||||
# get the run object
|
||||
for h in run_manager.handlers:
|
||||
if isinstance(h, LangChainTracer):
|
||||
run = h.run_map.get(str(run_manager.run_id))
|
||||
break
|
||||
else:
|
||||
run = None
|
||||
# run in context
|
||||
with set_config_context(config, run) as context:
|
||||
input = await asyncio.create_task(
|
||||
step.ainvoke(input, config, **kwargs),
|
||||
context=context,
|
||||
)
|
||||
else:
|
||||
run = None
|
||||
# run in context
|
||||
with set_config_context(config, run) as context:
|
||||
input = await step.ainvoke(input, config, **kwargs)
|
||||
else:
|
||||
input = await step.ainvoke(input, config)
|
||||
# finish the root run
|
||||
except BaseException as e:
|
||||
await run_manager.on_chain_error(e)
|
||||
raise
|
||||
else:
|
||||
await run_manager.on_chain_end(
|
||||
_process_outputs(self.trace_outputs, input)
|
||||
)
|
||||
return input
|
||||
else:
|
||||
# Still need to set config context for get_config() to work
|
||||
if ASYNCIO_ACCEPTS_CONTEXT:
|
||||
with set_config_context(config, None) as context:
|
||||
for i, step in enumerate(self.steps):
|
||||
if i == 0:
|
||||
input = await asyncio.create_task(
|
||||
step.ainvoke(input, config, **kwargs), context=context
|
||||
)
|
||||
else:
|
||||
else:
|
||||
input = await step.ainvoke(input, config)
|
||||
else:
|
||||
for i, step in enumerate(self.steps):
|
||||
if i == 0:
|
||||
input = await step.ainvoke(input, config, **kwargs)
|
||||
else:
|
||||
input = await step.ainvoke(input, config)
|
||||
# finish the root run
|
||||
except BaseException as e:
|
||||
await run_manager.on_chain_error(e)
|
||||
raise
|
||||
else:
|
||||
await run_manager.on_chain_end(input)
|
||||
else:
|
||||
input = await step.ainvoke(input, config)
|
||||
return input
|
||||
|
||||
def stream(
|
||||
@@ -725,80 +867,20 @@ class RunnableSeq(Runnable):
|
||||
) -> Iterator[Any]:
|
||||
if config is None:
|
||||
config = ensure_config()
|
||||
# setup callbacks
|
||||
callback_manager = get_callback_manager_for_config(config)
|
||||
# start the root run
|
||||
run_manager = callback_manager.on_chain_start(
|
||||
None,
|
||||
self.trace_inputs(input) if self.trace_inputs is not None else input,
|
||||
name=config.get("run_name") or self.get_name(),
|
||||
run_id=config.pop("run_id", None),
|
||||
)
|
||||
# get the run object
|
||||
for h in run_manager.handlers:
|
||||
if isinstance(h, LangChainTracer):
|
||||
run = h.run_map.get(str(run_manager.run_id))
|
||||
break
|
||||
else:
|
||||
run = None
|
||||
# create first step config
|
||||
config = patch_config(
|
||||
config,
|
||||
callbacks=run_manager.get_child(f"seq:step:{1}"),
|
||||
)
|
||||
# run all in context
|
||||
with set_config_context(config, run) as context:
|
||||
try:
|
||||
# stream the last steps
|
||||
# transform the input stream of each step with the next
|
||||
# steps that don't natively support transforming an input stream will
|
||||
# buffer input in memory until all available, and then start emitting output
|
||||
for idx, step in enumerate(self.steps):
|
||||
if idx == 0:
|
||||
iterator = step.stream(input, config, **kwargs)
|
||||
else:
|
||||
config = patch_config(
|
||||
config,
|
||||
callbacks=run_manager.get_child(f"seq:step:{idx + 1}"),
|
||||
)
|
||||
iterator = step.transform(iterator, config)
|
||||
# populates streamed_output in astream_log() output if needed
|
||||
if _StreamingCallbackHandler is not None:
|
||||
for h in run_manager.handlers:
|
||||
if isinstance(h, _StreamingCallbackHandler):
|
||||
iterator = h.tap_output_iter(run_manager.run_id, iterator)
|
||||
# consume into final output
|
||||
output = context.run(_consume_iter, iterator)
|
||||
# sequence doesn't emit output, yield to mark as generator
|
||||
yield
|
||||
except BaseException as e:
|
||||
run_manager.on_chain_error(e)
|
||||
raise
|
||||
else:
|
||||
run_manager.on_chain_end(output)
|
||||
|
||||
async def astream(
|
||||
self,
|
||||
input: Input,
|
||||
config: RunnableConfig | None = None,
|
||||
**kwargs: Any | None,
|
||||
) -> AsyncIterator[Any]:
|
||||
if config is None:
|
||||
config = ensure_config()
|
||||
# setup callbacks
|
||||
callback_manager = get_async_callback_manager_for_config(config)
|
||||
# start the root run
|
||||
run_manager = await callback_manager.on_chain_start(
|
||||
None,
|
||||
self.trace_inputs(input) if self.trace_inputs is not None else input,
|
||||
name=config.get("run_name") or self.get_name(),
|
||||
run_id=config.pop("run_id", None),
|
||||
)
|
||||
# stream the last steps
|
||||
# transform the input stream of each step with the next
|
||||
# steps that don't natively support transforming an input stream will
|
||||
# buffer input in memory until all available, and then start emitting output
|
||||
if ASYNCIO_ACCEPTS_CONTEXT:
|
||||
if self.trace:
|
||||
# setup callbacks
|
||||
callback_manager = get_callback_manager_for_config(config)
|
||||
# Filter out LangChainTracer if skip_langsmith is set
|
||||
if self.skip_langsmith:
|
||||
callback_manager = _filter_langsmith_handlers(callback_manager)
|
||||
# start the root run
|
||||
run_manager = callback_manager.on_chain_start(
|
||||
None,
|
||||
_process_inputs(self.trace_inputs, input),
|
||||
name=config.get("run_name") or self.get_name(),
|
||||
run_id=config.pop("run_id", None),
|
||||
)
|
||||
# get the run object
|
||||
for h in run_manager.handlers:
|
||||
if isinstance(h, LangChainTracer):
|
||||
@@ -813,18 +895,136 @@ class RunnableSeq(Runnable):
|
||||
)
|
||||
# run all in context
|
||||
with set_config_context(config, run) as context:
|
||||
try:
|
||||
# stream the last steps
|
||||
# transform the input stream of each step with the next
|
||||
# steps that don't natively support transforming an input stream will
|
||||
# buffer input in memory until all available, and then start emitting output
|
||||
for idx, step in enumerate(self.steps):
|
||||
if idx == 0:
|
||||
iterator = step.stream(input, config, **kwargs)
|
||||
else:
|
||||
config = patch_config(
|
||||
config,
|
||||
callbacks=run_manager.get_child(f"seq:step:{idx + 1}"),
|
||||
)
|
||||
iterator = step.transform(iterator, config)
|
||||
# populates streamed_output in astream_log() output if needed
|
||||
if _StreamingCallbackHandler is not None:
|
||||
for h in run_manager.handlers:
|
||||
if isinstance(h, _StreamingCallbackHandler):
|
||||
iterator = h.tap_output_iter(
|
||||
run_manager.run_id, iterator
|
||||
)
|
||||
# consume into final output
|
||||
output = context.run(_consume_iter, iterator)
|
||||
# sequence doesn't emit output, yield to mark as generator
|
||||
yield
|
||||
except BaseException as e:
|
||||
run_manager.on_chain_error(e)
|
||||
raise
|
||||
else:
|
||||
run_manager.on_chain_end(
|
||||
_process_outputs(self.trace_outputs, output)
|
||||
)
|
||||
else:
|
||||
# No tracing - still need to set config context for get_config() to work
|
||||
with set_config_context(config, None) as context:
|
||||
for idx, step in enumerate(self.steps):
|
||||
if idx == 0:
|
||||
iterator = step.stream(input, config, **kwargs)
|
||||
else:
|
||||
iterator = step.transform(iterator, config)
|
||||
context.run(_consume_iter, iterator)
|
||||
yield
|
||||
|
||||
async def astream(
|
||||
self,
|
||||
input: Input,
|
||||
config: RunnableConfig | None = None,
|
||||
**kwargs: Any | None,
|
||||
) -> AsyncIterator[Any]:
|
||||
if config is None:
|
||||
config = ensure_config()
|
||||
|
||||
if self.trace:
|
||||
# setup callbacks
|
||||
callback_manager = get_async_callback_manager_for_config(config)
|
||||
# Filter out LangChainTracer if skip_langsmith is set
|
||||
if self.skip_langsmith:
|
||||
callback_manager = _filter_langsmith_handlers_async(callback_manager)
|
||||
# start the root run
|
||||
run_manager = await callback_manager.on_chain_start(
|
||||
None,
|
||||
_process_inputs(self.trace_inputs, input),
|
||||
name=config.get("run_name") or self.get_name(),
|
||||
run_id=config.pop("run_id", None),
|
||||
)
|
||||
# stream the last steps
|
||||
# transform the input stream of each step with the next
|
||||
# steps that don't natively support transforming an input stream will
|
||||
# buffer input in memory until all available, and then start emitting output
|
||||
if ASYNCIO_ACCEPTS_CONTEXT:
|
||||
# get the run object
|
||||
for h in run_manager.handlers:
|
||||
if isinstance(h, LangChainTracer):
|
||||
run = h.run_map.get(str(run_manager.run_id))
|
||||
break
|
||||
else:
|
||||
run = None
|
||||
# create first step config
|
||||
config = patch_config(
|
||||
config,
|
||||
callbacks=run_manager.get_child(f"seq:step:{1}"),
|
||||
)
|
||||
# run all in context
|
||||
with set_config_context(config, run) as context:
|
||||
try:
|
||||
async with AsyncExitStack() as stack:
|
||||
for idx, step in enumerate(self.steps):
|
||||
if idx == 0:
|
||||
aiterator = step.astream(input, config, **kwargs)
|
||||
else:
|
||||
config = patch_config(
|
||||
config,
|
||||
callbacks=run_manager.get_child(
|
||||
f"seq:step:{idx + 1}"
|
||||
),
|
||||
)
|
||||
aiterator = step.atransform(aiterator, config)
|
||||
if hasattr(aiterator, "aclose"):
|
||||
stack.push_async_callback(aiterator.aclose)
|
||||
# populates streamed_output in astream_log() output if needed
|
||||
if _StreamingCallbackHandler is not None:
|
||||
for h in run_manager.handlers:
|
||||
if isinstance(h, _StreamingCallbackHandler):
|
||||
aiterator = h.tap_output_aiter(
|
||||
run_manager.run_id, aiterator
|
||||
)
|
||||
# consume into final output
|
||||
output = await asyncio.create_task(
|
||||
_consume_aiter(aiterator), context=context
|
||||
)
|
||||
# sequence doesn't emit output, yield to mark as generator
|
||||
yield
|
||||
except BaseException as e:
|
||||
await run_manager.on_chain_error(e)
|
||||
raise
|
||||
else:
|
||||
await run_manager.on_chain_end(
|
||||
_process_outputs(self.trace_outputs, output)
|
||||
)
|
||||
else:
|
||||
try:
|
||||
async with AsyncExitStack() as stack:
|
||||
for idx, step in enumerate(self.steps):
|
||||
config = patch_config(
|
||||
config,
|
||||
callbacks=run_manager.get_child(f"seq:step:{idx + 1}"),
|
||||
)
|
||||
if idx == 0:
|
||||
aiterator = step.astream(input, config, **kwargs)
|
||||
else:
|
||||
config = patch_config(
|
||||
config,
|
||||
callbacks=run_manager.get_child(
|
||||
f"seq:step:{idx + 1}"
|
||||
),
|
||||
)
|
||||
aiterator = step.atransform(aiterator, config)
|
||||
if hasattr(aiterator, "aclose"):
|
||||
stack.push_async_callback(aiterator.aclose)
|
||||
@@ -836,46 +1036,29 @@ class RunnableSeq(Runnable):
|
||||
run_manager.run_id, aiterator
|
||||
)
|
||||
# consume into final output
|
||||
output = await asyncio.create_task(
|
||||
_consume_aiter(aiterator), context=context
|
||||
)
|
||||
output = await _consume_aiter(aiterator)
|
||||
# sequence doesn't emit output, yield to mark as generator
|
||||
yield
|
||||
except BaseException as e:
|
||||
await run_manager.on_chain_error(e)
|
||||
raise
|
||||
else:
|
||||
await run_manager.on_chain_end(output)
|
||||
await run_manager.on_chain_end(
|
||||
_process_outputs(self.trace_outputs, output)
|
||||
)
|
||||
else:
|
||||
try:
|
||||
async with AsyncExitStack() as stack:
|
||||
for idx, step in enumerate(self.steps):
|
||||
config = patch_config(
|
||||
config,
|
||||
callbacks=run_manager.get_child(f"seq:step:{idx + 1}"),
|
||||
)
|
||||
if idx == 0:
|
||||
aiterator = step.astream(input, config, **kwargs)
|
||||
else:
|
||||
aiterator = step.atransform(aiterator, config)
|
||||
if hasattr(aiterator, "aclose"):
|
||||
stack.push_async_callback(aiterator.aclose)
|
||||
# populates streamed_output in astream_log() output if needed
|
||||
if _StreamingCallbackHandler is not None:
|
||||
for h in run_manager.handlers:
|
||||
if isinstance(h, _StreamingCallbackHandler):
|
||||
aiterator = h.tap_output_aiter(
|
||||
run_manager.run_id, aiterator
|
||||
)
|
||||
# consume into final output
|
||||
output = await _consume_aiter(aiterator)
|
||||
# sequence doesn't emit output, yield to mark as generator
|
||||
yield
|
||||
except BaseException as e:
|
||||
await run_manager.on_chain_error(e)
|
||||
raise
|
||||
else:
|
||||
await run_manager.on_chain_end(output)
|
||||
# No tracing - just execute the steps directly
|
||||
async with AsyncExitStack() as stack:
|
||||
for idx, step in enumerate(self.steps):
|
||||
if idx == 0:
|
||||
aiterator = step.astream(input, config, **kwargs)
|
||||
else:
|
||||
aiterator = step.atransform(aiterator, config)
|
||||
if hasattr(aiterator, "aclose"):
|
||||
stack.push_async_callback(aiterator.aclose)
|
||||
# consume into final output
|
||||
await _consume_aiter(aiterator)
|
||||
yield
|
||||
|
||||
|
||||
def _consume_iter(it: Iterator[Any]) -> Any:
|
||||
|
||||
@@ -21,6 +21,7 @@ from langgraph._internal._runnable import (
|
||||
run_in_executor,
|
||||
)
|
||||
from langgraph.config import get_config
|
||||
from langgraph.pregel._read import _validate_traceable_config
|
||||
from langgraph.pregel._write import ChannelWrite, ChannelWriteEntry
|
||||
from langgraph.types import CachePolicy, RetryPolicy
|
||||
|
||||
@@ -192,6 +193,19 @@ def get_runnable_for_entrypoint(func: Callable[..., Any]) -> Runnable:
|
||||
return CACHE.setdefault(key, run)
|
||||
|
||||
|
||||
def _compose_trace_inputs(
|
||||
base_fn: Callable[[Any], Any],
|
||||
process_fn: Callable[[Any], Any],
|
||||
) -> Callable[[Any], Any]:
|
||||
"""Compose trace_inputs with process_inputs from traceable config."""
|
||||
|
||||
def composed(inputs: Any) -> Any:
|
||||
exploded = base_fn(inputs)
|
||||
return process_fn(exploded)
|
||||
|
||||
return composed
|
||||
|
||||
|
||||
def get_runnable_for_task(func: Callable[..., Any]) -> Runnable:
|
||||
key = (func, True)
|
||||
if key in CACHE:
|
||||
@@ -206,10 +220,22 @@ def get_runnable_for_task(func: Callable[..., Any]) -> Runnable:
|
||||
else:
|
||||
name = str(func)
|
||||
|
||||
if is_async_callable(func):
|
||||
# Check for traceable config early to handle __unwrapped__
|
||||
raw_config = getattr(func, "__traceable_config__", None)
|
||||
traceable_config = _validate_traceable_config(raw_config)
|
||||
|
||||
# Use unwrapped function if available to avoid double-tracing
|
||||
# when @traceable and @task are used together
|
||||
func_to_run: Callable[..., Any] = func
|
||||
if traceable_config:
|
||||
unwrapped = traceable_config.get("__unwrapped__")
|
||||
if unwrapped is not None:
|
||||
func_to_run = unwrapped
|
||||
|
||||
if is_async_callable(func_to_run):
|
||||
run = RunnableCallable(
|
||||
None,
|
||||
func,
|
||||
func_to_run,
|
||||
explode_args=True,
|
||||
name=name,
|
||||
trace=False,
|
||||
@@ -217,20 +243,45 @@ def get_runnable_for_task(func: Callable[..., Any]) -> Runnable:
|
||||
)
|
||||
else:
|
||||
run = RunnableCallable(
|
||||
func,
|
||||
functools.wraps(func)(functools.partial(run_in_executor, None, func)),
|
||||
func_to_run,
|
||||
functools.wraps(func_to_run)(
|
||||
functools.partial(run_in_executor, None, func_to_run)
|
||||
),
|
||||
explode_args=True,
|
||||
name=name,
|
||||
trace=False,
|
||||
recurse=False,
|
||||
)
|
||||
|
||||
# Build trace_inputs - compose with process_inputs if provided
|
||||
# Use original func for signature to match expected args
|
||||
base_trace_inputs = functools.partial(
|
||||
_explode_args_trace_inputs, inspect.signature(func)
|
||||
)
|
||||
process_inputs = (
|
||||
traceable_config.get("process_inputs") if traceable_config else None
|
||||
)
|
||||
if process_inputs is not None:
|
||||
trace_inputs = _compose_trace_inputs(base_trace_inputs, process_inputs)
|
||||
else:
|
||||
trace_inputs = base_trace_inputs
|
||||
|
||||
# Build seq_kwargs
|
||||
seq_kwargs: dict[str, Any] = {
|
||||
"name": name,
|
||||
"trace_inputs": trace_inputs,
|
||||
}
|
||||
if traceable_config:
|
||||
# Filter out LangChainTracer (skip LangSmith) but keep other callbacks
|
||||
if traceable_config.get("enabled") is False:
|
||||
seq_kwargs["skip_langsmith"] = True
|
||||
if traceable_config.get("process_outputs"):
|
||||
seq_kwargs["trace_outputs"] = traceable_config["process_outputs"]
|
||||
|
||||
seq = RunnableSeq(
|
||||
run,
|
||||
ChannelWrite([ChannelWriteEntry(RETURN)]),
|
||||
name=name,
|
||||
trace_inputs=functools.partial(
|
||||
_explode_args_trace_inputs, inspect.signature(func)
|
||||
),
|
||||
**seq_kwargs,
|
||||
)
|
||||
if not _lookup_module_and_qualname(func):
|
||||
return seq
|
||||
|
||||
@@ -7,10 +7,15 @@ from typing import (
|
||||
)
|
||||
|
||||
from langchain_core.runnables import Runnable, RunnableConfig
|
||||
from typing_extensions import TypedDict
|
||||
|
||||
from langgraph._internal._config import merge_configs
|
||||
from langgraph._internal._constants import CONF, CONFIG_KEY_READ
|
||||
from langgraph._internal._runnable import RunnableCallable, RunnableSeq
|
||||
from langgraph._internal._runnable import (
|
||||
RunnableCallable,
|
||||
RunnableSeq,
|
||||
coerce_to_runnable,
|
||||
)
|
||||
from langgraph.pregel._utils import find_subgraph_pregel
|
||||
from langgraph.pregel._write import ChannelWrite
|
||||
from langgraph.pregel.protocol import PregelProtocol
|
||||
@@ -20,6 +25,32 @@ READ_TYPE = Callable[[str | Sequence[str], bool], Any | dict[str, Any]]
|
||||
INPUT_CACHE_KEY_TYPE = tuple[Callable[..., Any], tuple[str, ...]]
|
||||
|
||||
|
||||
class TraceableConfig(TypedDict, total=False):
|
||||
"""Configuration extracted from @traceable decorated functions."""
|
||||
|
||||
process_inputs: Callable[[Any], Any] | None
|
||||
process_outputs: Callable[[Any], Any] | None
|
||||
enabled: bool | None
|
||||
__unwrapped__: Callable[[Any], Any] | None
|
||||
|
||||
|
||||
def _validate_traceable_config(raw: Any) -> TraceableConfig | None:
|
||||
"""Validate __traceable_config__ has expected structure.
|
||||
|
||||
Returns validated config dict or None if invalid.
|
||||
"""
|
||||
if not isinstance(raw, dict):
|
||||
return None
|
||||
# Support both "__unwrapped__" (langsmith) and "wrapped" (legacy) keys
|
||||
unwrapped = raw.get("__unwrapped__") or raw.get("wrapped")
|
||||
return {
|
||||
"process_inputs": raw.get("process_inputs"),
|
||||
"process_outputs": raw.get("process_outputs"),
|
||||
"enabled": raw.get("enabled"), # None means use external context
|
||||
"__unwrapped__": unwrapped,
|
||||
}
|
||||
|
||||
|
||||
class ChannelRead(RunnableCallable):
|
||||
"""Implements the logic for reading state from CONFIG_KEY_READ.
|
||||
Usable both as a runnable as well as a static method to call imperatively."""
|
||||
@@ -176,6 +207,7 @@ class PregelNode:
|
||||
attrs = {**self.__dict__, **update}
|
||||
# Drop the cached properties
|
||||
attrs.pop("flat_writers", None)
|
||||
attrs.pop("_traceable_config", None)
|
||||
attrs.pop("node", None)
|
||||
attrs.pop("input_cache_key", None)
|
||||
return PregelNode(**attrs)
|
||||
@@ -197,6 +229,31 @@ class PregelNode:
|
||||
writers.pop()
|
||||
return writers
|
||||
|
||||
@cached_property
|
||||
def _traceable_config(self) -> TraceableConfig | None:
|
||||
"""Extract @traceable config if bound wraps a traceable function.
|
||||
|
||||
Returns validated TraceableConfig with:
|
||||
- process_inputs: Optional input processing function
|
||||
- process_outputs: Optional output processing function
|
||||
- enabled: Whether tracing is enabled (None = honor external context)
|
||||
- __unwrapped__: The original unwrapped function (if available)
|
||||
"""
|
||||
if not isinstance(self.bound, RunnableCallable):
|
||||
return None
|
||||
|
||||
func = self.bound.func or self.bound.afunc
|
||||
if func is None:
|
||||
return None
|
||||
|
||||
raw_config = getattr(func, "__traceable_config__", None)
|
||||
if raw_config is None:
|
||||
return None
|
||||
config = _validate_traceable_config(raw_config)
|
||||
if config is None:
|
||||
return None
|
||||
return config
|
||||
|
||||
@cached_property
|
||||
def node(self) -> Runnable[Any, Any] | None:
|
||||
"""Get a runnable that combines `bound` and `writers`."""
|
||||
@@ -207,10 +264,24 @@ class PregelNode:
|
||||
return writers[0]
|
||||
elif self.bound is DEFAULT_BOUND:
|
||||
return RunnableSeq(*writers)
|
||||
elif writers:
|
||||
return RunnableSeq(self.bound, *writers)
|
||||
else:
|
||||
elif not writers:
|
||||
return self.bound
|
||||
tc = self._traceable_config
|
||||
|
||||
seq_kwargs: dict[str, Any] = {}
|
||||
if tc:
|
||||
# Filter out LangChainTracer (skip LangSmith) but keep other callbacks
|
||||
if tc.get("enabled") is False:
|
||||
seq_kwargs["skip_langsmith"] = True
|
||||
seq_kwargs["trace_inputs"] = tc.get("process_inputs")
|
||||
seq_kwargs["trace_outputs"] = tc.get("process_outputs")
|
||||
|
||||
bound = self.bound
|
||||
if tc and (unwrapped := tc.get("__unwrapped__")):
|
||||
# We want to avoid double-tracing.
|
||||
bound = coerce_to_runnable(unwrapped, name=None, trace=False)
|
||||
|
||||
return RunnableSeq(bound, *writers, **seq_kwargs)
|
||||
|
||||
@cached_property
|
||||
def input_cache_key(self) -> INPUT_CACHE_KEY_TYPE:
|
||||
@@ -218,9 +289,11 @@ class PregelNode:
|
||||
This is used to avoid calculating the same input multiple times."""
|
||||
return (
|
||||
self.mapper,
|
||||
tuple(self.channels)
|
||||
if isinstance(self.channels, list)
|
||||
else (self.channels,),
|
||||
(
|
||||
tuple(self.channels)
|
||||
if isinstance(self.channels, list)
|
||||
else (self.channels,)
|
||||
),
|
||||
)
|
||||
|
||||
def invoke(
|
||||
|
||||
@@ -3038,7 +3038,7 @@ def test_message_graph(
|
||||
# add an extra message as if it came from "tools" node
|
||||
app_w_interrupt.update_state(config, ("ai", "an extra message"), as_node="tools")
|
||||
|
||||
# extra message is coerced BaseMessge and appended
|
||||
# extra message is coerced BaseMessage and appended
|
||||
# now the next node is "agent" per the graph edges
|
||||
assert app_w_interrupt.get_state(config) == StateSnapshot(
|
||||
values=[
|
||||
@@ -3762,7 +3762,7 @@ def test_root_graph(
|
||||
# add an extra message as if it came from "tools" node
|
||||
app_w_interrupt.update_state(config, ("ai", "an extra message"), as_node="tools")
|
||||
|
||||
# extra message is coerced BaseMessge and appended
|
||||
# extra message is coerced BaseMessage and appended
|
||||
# now the next node is "agent" per the graph edges
|
||||
assert app_w_interrupt.get_state(config) == StateSnapshot(
|
||||
values=[
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -15,12 +15,12 @@ The module implements design patterns for:
|
||||
|
||||
Key Components:
|
||||
|
||||
- `ToolNode`: Main class for executing tools in LangGraph workflows
|
||||
- `InjectedState`: Annotation for injecting graph state into tools
|
||||
- `InjectedStore`: Annotation for injecting persistent store into tools
|
||||
- `ToolRuntime`: Runtime information for tools, bundling together `state`, `context`,
|
||||
- [`ToolNode`][langgraph.prebuilt.ToolNode]: Main class for executing tools in LangGraph workflows
|
||||
- [`InjectedState`][langgraph.prebuilt.InjectedState]: Annotation for injecting graph state into tools
|
||||
- [`InjectedStore`][langgraph.prebuilt.InjectedStore]: Annotation for injecting persistent store into tools
|
||||
- [`ToolRuntime`][langgraph.prebuilt.ToolRuntime]: Runtime information for tools, bundling together `state`, `context`,
|
||||
`config`, `stream_writer`, `tool_call_id`, and `store`
|
||||
- `tools_condition`: Utility function for conditional routing based on tool calls
|
||||
- [`tools_condition`][langgraph.prebuilt.tools_condition]: Utility function for conditional routing based on tool calls
|
||||
|
||||
Typical Usage:
|
||||
```python
|
||||
@@ -614,8 +614,16 @@ class ToolNode(RunnableCallable):
|
||||
persistent storage, and control flow. Manages parallel execution,
|
||||
error handling.
|
||||
|
||||
Use `ToolNode` when building custom workflows that require fine-grained control over
|
||||
tool execution—for example, custom routing logic, specialized error handling, or
|
||||
non-standard agent architectures.
|
||||
|
||||
For standard ReAct-style agents, use [`create_agent`][langchain.agents.create_agent]
|
||||
instead. It uses `ToolNode` internally with sensible defaults for the agent loop,
|
||||
conditional routing, and error handling.
|
||||
|
||||
Input Formats:
|
||||
1. Graph state with `messages` key that has a list of messages:
|
||||
1. **Graph state** with `messages` key that has a list of messages:
|
||||
- Common representation for agentic workflows
|
||||
- Supports custom messages key via `messages_key` parameter
|
||||
|
||||
|
||||
@@ -14,7 +14,6 @@ import functools
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import ssl
|
||||
import sys
|
||||
import warnings
|
||||
from collections.abc import AsyncIterator, Callable, Iterator, Mapping, Sequence
|
||||
@@ -181,7 +180,6 @@ def get_client(
|
||||
api_key: str | None = NOT_PROVIDED,
|
||||
headers: Mapping[str, str] | None = None,
|
||||
timeout: TimeoutTypes | None = None,
|
||||
verify: VerifyTypes | None = NOT_PROVIDED,
|
||||
) -> LangGraphClient:
|
||||
"""Create and configure a LangGraphClient.
|
||||
|
||||
@@ -210,13 +208,6 @@ def get_client(
|
||||
- float (total seconds)
|
||||
- tuple `(connect, read, write, pool)` in seconds
|
||||
Defaults: connect=5, read=300, write=300, pool=5.
|
||||
verify:
|
||||
SSL certificate verification. May be:
|
||||
- `True`: verify SSL certificates using system CA bundle
|
||||
- `False`: disable SSL verification (not recommended for production)
|
||||
- `str`: path to a CA bundle file or directory
|
||||
- `ssl.SSLContext`: custom SSL context for full control
|
||||
- Not provided (default): uses httpx default
|
||||
|
||||
Returns:
|
||||
LangGraphClient:
|
||||
@@ -283,10 +274,7 @@ def get_client(
|
||||
_registered_transports.append(transport)
|
||||
|
||||
if transport is None:
|
||||
transport_kwargs: dict[str, Any] = {"retries": 5}
|
||||
if verify is not NOT_PROVIDED:
|
||||
transport_kwargs["verify"] = verify
|
||||
transport = httpx.AsyncHTTPTransport(**transport_kwargs)
|
||||
transport = httpx.AsyncHTTPTransport(retries=5)
|
||||
client = httpx.AsyncClient(
|
||||
base_url=url,
|
||||
transport=transport,
|
||||
@@ -3583,7 +3571,6 @@ def get_sync_client(
|
||||
api_key: str | None = NOT_PROVIDED,
|
||||
headers: Mapping[str, str] | None = None,
|
||||
timeout: TimeoutTypes | None = None,
|
||||
verify: VerifyTypes | None = NOT_PROVIDED,
|
||||
) -> SyncLangGraphClient:
|
||||
"""Get a synchronous LangGraphClient instance.
|
||||
|
||||
@@ -3601,12 +3588,6 @@ def get_sync_client(
|
||||
Accepts an httpx.Timeout instance, a float (seconds), or a tuple of timeouts.
|
||||
Tuple format is (connect, read, write, pool)
|
||||
If not provided, defaults to connect=5s, read=300s, write=300s, and pool=5s.
|
||||
verify: SSL certificate verification. Can be:
|
||||
- `True`: verify SSL certificates using system CA bundle
|
||||
- `False`: disable SSL verification (not recommended for production)
|
||||
- `str`: path to a CA bundle file or directory
|
||||
- `ssl.SSLContext`: custom SSL context for full control
|
||||
- Not provided (default): uses httpx default
|
||||
Returns:
|
||||
SyncLangGraphClient: The top-level synchronous client for accessing AssistantsClient,
|
||||
ThreadsClient, RunsClient, and CronClient.
|
||||
@@ -3639,10 +3620,7 @@ def get_sync_client(
|
||||
if url is None:
|
||||
url = "http://localhost:8123"
|
||||
|
||||
transport_kwargs: dict[str, Any] = {"retries": 5}
|
||||
if verify is not NOT_PROVIDED:
|
||||
transport_kwargs["verify"] = verify
|
||||
transport = httpx.HTTPTransport(**transport_kwargs)
|
||||
transport = httpx.HTTPTransport(retries=5)
|
||||
client = httpx.Client(
|
||||
base_url=url,
|
||||
transport=transport,
|
||||
@@ -6907,5 +6885,3 @@ TimeoutTypes = (
|
||||
| tuple[float | None, float | None, float | None, float | None]
|
||||
| httpx.Timeout
|
||||
)
|
||||
|
||||
VerifyTypes = bool | str | ssl.SSLContext
|
||||
|
||||
Reference in New Issue
Block a user