WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-09-24 18:45:11 +02:00
Code Issues Packages Projects Releases Wiki Activity

docs: update streaming docstring for Pregel and expose in api ref (#3229)

Browse Source
This commit is contained in:
Vadym Barda
2025-01-29 12:07:07 -05:00
committed by GitHub
parent ed09a77d9f
commit 82148e9bf2
5 changed files with 141 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/docs/concepts/functional_api.md
+2 -2
View File
@@ -146,7 +146,7 @@ An **entrypoint** is defined by decorating a function with the `@entrypoint` dec
The function **must accept a single positional argument**, which serves as the workflow input. If you need to pass multiple pieces of data, use a dictionary as the input type for the first argument.
Decorating a function with an `entrypoint` produces a Pregel instance which helps to manage the execution of the workflow (e.g., handles streaming, resumption, and checkpointing).
Decorating a function with an `entrypoint` produces a [`Pregel`][langgraph.pregel.Pregel.stream] instance which helps to manage the execution of the workflow (e.g., handles streaming, resumption, and checkpointing).
You will usually want to pass a **checkpointer** to the `@entrypoint` decorator to enable persistence and use features like **human-in-the-loop**.
@@ -223,7 +223,7 @@ When declaring an `entrypoint`, you can request access to additional parameters
### Executing
Using the [`@entrypoint`](#entrypoint) yields a Pregel object that can be executed using the `invoke`, `ainvoke`, `stream`, and `astream` methods.
Using the [`@entrypoint`](#entrypoint) yields a [`Pregel`][langgraph.pregel.Pregel.stream] object that can be executed using the `invoke`, `ainvoke`, `stream`, and `astream` methods.
=== "Invoke"
docs/docs/reference/pregel.md
+9
View File
@@ -0,0 +1,9 @@
::: langgraph.pregel.Pregel
options:
members:
- stream
- astream
- invoke
- ainvoke
- update_state
- aupdate_state
docs/mkdocs.yml
+1
View File
@@ -379,6 +379,7 @@ nav:
- Errors: reference/errors.md
- Types: reference/types.md
- Constants: reference/constants.md
- Pregel: reference/pregel.md
- Functional API: reference/func.md
- LangGraph Platform:
- Server API: "cloud/reference/api/api_ref.md"
libs/langgraph/langgraph/pregel/__init__.py
+122 -14
View File
@@ -1490,11 +1490,15 @@ class Pregel(PregelProtocol):
input: The input to the graph.
config: The configuration to use for the run.
stream_mode: The mode to stream output, defaults to self.stream_mode.
Options are 'values', 'updates', and 'debug'.
values: Emit the current values of the state for each step.
updates: Emit only the updates to the state for each step.
Output is a dict with the node name as key and the updated values as value.
debug: Emit debug events for each step.
Options are:
- `"values"`: Emit all values in the state after each step.
When used with functional API, values are emitted once at the end of the workflow.
- `"updates"`: Emit only the node or task names and updates returned by the nodes or tasks after each step.
If multiple updates are made in the same step (e.g. multiple nodes are run) then those updates are emitted separately.
- `"custom"`: Emit custom data using from inside nodes or tasks using `StreamWriter`.
- `"messages"`: Emit LLM messages token-by-token together with metadata for any LLM invocations inside nodes or tasks.
- `"debug"`: Emit debug events with as much information as possible for each step.
output_keys: The keys to stream, defaults to all non-context channels.
interrupt_before: Nodes to interrupt before, defaults to all nodes in the graph.
interrupt_after: Nodes to interrupt after, defaults to all nodes in the graph.
@@ -1509,8 +1513,7 @@ class Pregel(PregelProtocol):
```pycon
>>> import operator
>>> from typing_extensions import Annotated, TypedDict
>>> from langgraph.graph import StateGraph
>>> from langgraph.constants import START
>>> from langgraph.graph import StateGraph, START
...
>>> class State(TypedDict):
... alist: Annotated[list, operator.add]
@@ -1550,6 +1553,57 @@ class Pregel(PregelProtocol):
{'type': 'task', 'timestamp': '2024-06-23T...+00:00', 'step': 2, 'payload': {'id': '...', 'name': 'b', 'input': {'alist': ['Ex for stream_mode="debug"'], 'another_list': ['hi']}, 'triggers': ['a']}}
{'type': 'task_result', 'timestamp': '2024-06-23T...+00:00', 'step': 2, 'payload': {'id': '...', 'name': 'b', 'result': [('alist', ['there'])]}}
```
With stream_mode="custom":
```pycon
>>> from langgraph.types import StreamWriter
...
>>> def node_a(state: State, writer: StreamWriter):
... writer({"custom_data": "foo"})
... return {"alist": ["hi"]}
...
>>> builder = StateGraph(State)
>>> builder.add_node("a", node_a)
>>> builder.add_edge(START, "a")
>>> graph = builder.compile()
...
>>> for event in graph.stream({"alist": ['Ex for stream_mode="custom"']}, stream_mode="custom"):
... print(event)
{'custom_data': 'foo'}
```
With stream_mode="messages":
```pycon
>>> from typing_extensions import Annotated, TypedDict
>>> from langgraph.graph import StateGraph, START
>>> from langchain_openai import ChatOpenAI
...
>>> llm = ChatOpenAI(model="gpt-4o-mini")
...
>>> class State(TypedDict):
... question: str
... answer: str
...
>>> def node_a(state: State):
... response = llm.invoke(state["question"])
... return {"answer": response.content}
...
>>> builder = StateGraph(State)
>>> builder.add_node("a", node_a)
>>> builder.add_edge(START, "a")
>>> graph = builder.compile()
>>> for event in graph.stream({"question": "What is the capital of France?"}, stream_mode="messages"):
... print(event)
(AIMessageChunk(content='The', additional_kwargs={}, response_metadata={}, id='...'), {'langgraph_step': 1, 'langgraph_node': 'a', 'langgraph_triggers': ['start:a'], 'langgraph_path': ('__pregel_pull', 'a'), 'langgraph_checkpoint_ns': '...', 'checkpoint_ns': '...', 'ls_provider': 'openai', 'ls_model_name': 'gpt-4o-mini', 'ls_model_type': 'chat', 'ls_temperature': 0.7})
(AIMessageChunk(content=' capital', additional_kwargs={}, response_metadata={}, id='...'), {'langgraph_step': 1, 'langgraph_node': 'a', 'langgraph_triggers': ['start:a'], ...})
(AIMessageChunk(content=' of', additional_kwargs={}, response_metadata={}, id='...'), {...})
(AIMessageChunk(content=' France', additional_kwargs={}, response_metadata={}, id='...'), {...})
(AIMessageChunk(content=' is', additional_kwargs={}, response_metadata={}, id='...'), {...})
(AIMessageChunk(content=' Paris', additional_kwargs={}, response_metadata={}, id='...'), {...})
```
"""
stream = SyncQueue()
@@ -1712,11 +1766,15 @@ class Pregel(PregelProtocol):
input: The input to the graph.
config: The configuration to use for the run.
stream_mode: The mode to stream output, defaults to self.stream_mode.
Options are 'values', 'updates', and 'debug'.
values: Emit the current values of the state for each step.
updates: Emit only the updates to the state for each step.
Output is a dict with the node name as key and the updated values as value.
debug: Emit debug events for each step.
Options are:
- `"values"`: Emit all values in the state after each step.
When used with functional API, values are emitted once at the end of the workflow.
- `"updates"`: Emit only the node or task names and updates returned by the nodes or tasks after each step.
If multiple updates are made in the same step (e.g. multiple nodes are run) then those updates are emitted separately.
- `"custom"`: Emit custom data using from inside nodes or tasks using `StreamWriter`.
- `"messages"`: Emit LLM messages token-by-token together with metadata for any LLM invocations inside nodes or tasks.
- `"debug"`: Emit debug events with as much information as possible for each step.
output_keys: The keys to stream, defaults to all non-context channels.
interrupt_before: Nodes to interrupt before, defaults to all nodes in the graph.
interrupt_after: Nodes to interrupt after, defaults to all nodes in the graph.
@@ -1731,8 +1789,7 @@ class Pregel(PregelProtocol):
```pycon
>>> import operator
>>> from typing_extensions import Annotated, TypedDict
>>> from langgraph.graph import StateGraph
>>> from langgraph.constants import START
>>> from langgraph.graph import StateGraph, START
...
>>> class State(TypedDict):
... alist: Annotated[list, operator.add]
@@ -1772,6 +1829,57 @@ class Pregel(PregelProtocol):
{'type': 'task', 'timestamp': '2024-06-23T...+00:00', 'step': 2, 'payload': {'id': '...', 'name': 'b', 'input': {'alist': ['Ex for stream_mode="debug"'], 'another_list': ['hi']}, 'triggers': ['a']}}
{'type': 'task_result', 'timestamp': '2024-06-23T...+00:00', 'step': 2, 'payload': {'id': '...', 'name': 'b', 'result': [('alist', ['there'])]}}
```
With stream_mode="custom":
```pycon
>>> from langgraph.types import StreamWriter
...
>>> async def node_a(state: State, writer: StreamWriter):
... writer({"custom_data": "foo"})
... return {"alist": ["hi"]}
...
>>> builder = StateGraph(State)
>>> builder.add_node("a", node_a)
>>> builder.add_edge(START, "a")
>>> graph = builder.compile()
...
>>> async for event in graph.astream({"alist": ['Ex for stream_mode="custom"']}, stream_mode="custom"):
... print(event)
{'custom_data': 'foo'}
```
With stream_mode="messages":
```pycon
>>> from typing_extensions import Annotated, TypedDict
>>> from langgraph.graph import StateGraph, START
>>> from langchain_openai import ChatOpenAI
...
>>> llm = ChatOpenAI(model="gpt-4o-mini")
...
>>> class State(TypedDict):
... question: str
... answer: str
...
>>> async def node_a(state: State):
... response = await llm.ainvoke(state["question"])
... return {"answer": response.content}
...
>>> builder = StateGraph(State)
>>> builder.add_node("a", node_a)
>>> builder.add_edge(START, "a")
>>> graph = builder.compile()
>>> for event in graph.stream({"question": "What is the capital of France?"}, stream_mode="messages"):
... print(event)
(AIMessageChunk(content='The', additional_kwargs={}, response_metadata={}, id='...'), {'langgraph_step': 1, 'langgraph_node': 'a', 'langgraph_triggers': ['start:a'], 'langgraph_path': ('__pregel_pull', 'a'), 'langgraph_checkpoint_ns': '...', 'checkpoint_ns': '...', 'ls_provider': 'openai', 'ls_model_name': 'gpt-4o-mini', 'ls_model_type': 'chat', 'ls_temperature': 0.7})
(AIMessageChunk(content=' capital', additional_kwargs={}, response_metadata={}, id='...'), {'langgraph_step': 1, 'langgraph_node': 'a', 'langgraph_triggers': ['start:a'], ...})
(AIMessageChunk(content=' of', additional_kwargs={}, response_metadata={}, id='...'), {...})
(AIMessageChunk(content=' France', additional_kwargs={}, response_metadata={}, id='...'), {...})
(AIMessageChunk(content=' is', additional_kwargs={}, response_metadata={}, id='...'), {...})
(AIMessageChunk(content=' Paris', additional_kwargs={}, response_metadata={}, id='...'), {...})
```
"""
stream = AsyncQueue()
libs/langgraph/langgraph/types.py
+7 -6
View File
@@ -48,12 +48,13 @@ Checkpointer = Union[None, bool, BaseCheckpointSaver]
StreamMode = Literal["values", "updates", "debug", "messages", "custom"]
"""How the stream method should emit outputs.
- 'values': Emit all values of the state for each step.
- 'updates': Emit only the node name(s) and updates
that were returned by the node(s) **after** each step.
- 'debug': Emit debug events for each step.
- 'messages': Emit LLM messages token-by-token.
- 'custom': Emit custom output `write: StreamWriter` kwarg of each node.
- `"values"`: Emit all values in the state after each step.
When used with functional API, values are emitted once at the end of the workflow.
- `"updates"`: Emit only the node or task names and updates returned by the nodes or tasks after each step.
If multiple updates are made in the same step (e.g. multiple nodes are run) then those updates are emitted separately.
- `"custom"`: Emit custom data using from inside nodes or tasks using `StreamWriter`.
- `"messages"`: Emit LLM messages token-by-token together with metadata for any LLM invocations inside nodes or tasks.
- `"debug"`: Emit debug events with as much information as possible for each step.
"""
StreamWriter = Callable[[Any], None]