WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-09-06 01:37:49 +02:00
Code Issues Packages Projects Releases Wiki Activity

[Docs] Update docstrings (#780)

Browse Source 
To add more color on the meaning of some arguments.
This commit is contained in:
William FH
2024-06-23 17:38:26 -07:00
committed by GitHub
parent 20f84f2041
commit d9b4d021bd
6 changed files with 178 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/_scripts/copy_notebooks.py
+5 -4
View File
@@ -42,7 +42,7 @@ _MANUAL = {
"introduction.ipynb",
"customer-support/customer-support.ipynb",
"tutorials/tnt-llm/tnt-llm.ipynb",
"tutorials/sql-agent.ipynb"
"tutorials/sql-agent.ipynb",
],
}
_MANUAL_INVERSE = {v: docs_dir / k for k, vs in _MANUAL.items() for v in vs}
@@ -103,9 +103,9 @@ def copy_notebooks():
continue
if any(path in _HOW_TOS for path in root.split(os.sep)):
dst_dir = how_tos_dir
elif 'sdk' in root.split(os.sep):
elif "sdk" in root.split(os.sep):
dst_dir = cloud_sdk_dir
elif 'cloud_examples' in root.split(os.sep):
elif "cloud_examples" in root.split(os.sep):
dst_dir = cloud_how_tos_dir
else:
dst_dir = tutorials_dir
@@ -145,7 +145,7 @@ def copy_notebooks():
with open(dst_path, "w") as f:
f.write(content)
dst_dir = dst_dir_
# Top level notebooks are "how-to's"
# for file in examples_dir.iterdir():
# if file.suffix.endswith(".ipynb") and not os.path.isdir(
@@ -155,6 +155,7 @@ def copy_notebooks():
# dst_path = os.path.join(docs_dir, "how-tos", file.name)
# shutil.copy(src_path, dst_path)
if __name__ == "__main__":
clean_notebooks()
copy_notebooks()
docs/docs/reference/graphs.md
+6 -5
View File
@@ -13,17 +13,19 @@ graph = StateGraph(MyState)
```
::: langgraph.graph.StateGraph
handler: python
handler: python
## MessageGraph
::: langgraph.graph.message.MessageGraph
## CompiledGraph
::: langgraph.graph.graph.CompiledGraph
handler: python
## StreamMode
::: langgraph.pregel.StreamMode
## Constants
@@ -63,5 +65,4 @@ builder.add_conditional_edges("my_node", my_condition)
## Send
::: langgraph.constants.Send
handler: python
::: langgraph.constants.Send
libs/langgraph/langgraph/checkpoint/memory.py
+1 -1
View File
@@ -98,7 +98,7 @@ class MemorySaver(BaseCheckpointSaver):
"""List checkpoints from the in-memory storage.
This method retrieves a list of checkpoint tuples from the in-memory storage based
on the provided config. The checkpoints are ordered by timestamp in descending order.
on the provided config. The checkpoints are ordered by timestamp in insertion order.
Args:
config (RunnableConfig): The config to use for listing the checkpoints.
libs/langgraph/langgraph/graph/graph.py
+2
View File
@@ -224,6 +224,8 @@ class Graph:
def set_entry_point(self, key: str) -> None:
"""Specifies the first node to be called in the graph.
Equivalent to calling `add_edge(START, key)`.
Parameters:
key (str): The key of the node to set as the entry point.
libs/langgraph/langgraph/graph/message.py
+2 -1
View File
@@ -103,8 +103,9 @@ class MessageGraph(StateGraph):
>>> builder.set_finish_point("chatbot")
>>> builder.compile().invoke([("user", "Hi there.")])
[HumanMessage(content="Hi there.", id='...'), AIMessage(content="Hello!", id='...')]
```
```pycon
>>> from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
>>> from langgraph.graph.message import MessageGraph
...
libs/langgraph/langgraph/pregel/__init__.py
+162 -59
View File
@@ -160,11 +160,15 @@ class Channel:
return PregelNode(
channels=cast(
Union[Mapping[None, str], Mapping[str, str]],
{key: channels}
if isinstance(channels, str) and key is not None
else [channels]
if isinstance(channels, str)
else {chan: chan for chan in channels},
(
{key: channels}
if isinstance(channels, str) and key is not None
else (
[channels]
if isinstance(channels, str)
else {chan: chan for chan in channels}
)
),
),
triggers=[channels] if isinstance(channels, str) else channels,
tags=tags,
@@ -180,15 +184,24 @@ class Channel:
return ChannelWrite(
[ChannelWriteEntry(c) for c in channels]
+ [
ChannelWriteEntry(k, skip_none=True, mapper=coerce_to_runnable(v))
if isinstance(v, Runnable) or callable(v)
else ChannelWriteEntry(k, value=v)
(
ChannelWriteEntry(k, skip_none=True, mapper=coerce_to_runnable(v))
if isinstance(v, Runnable) or callable(v)
else ChannelWriteEntry(k, value=v)
)
for k, v in kwargs.items()
]
)
StreamMode = Literal["values", "updates", "debug"]
"""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.
"""
class Pregel(
@@ -738,7 +751,73 @@ class Pregel(
interrupt_after: Optional[Union[All, Sequence[str]]] = None,
debug: Optional[bool] = None,
) -> Iterator[Union[dict[str, Any], Any]]:
"""Stream graph steps for a single input."""
"""Stream graph steps for a single input.
Args:
input: The input to the graph.
config: The configuration to use for the run.
stream_mode: The mode to stream output, defaults to 'values'.
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.
output_keys: The keys to stream, defaults to all non-context channels.
input_keys: The keys to use from the input, defaults to all input 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.
debug: Whether to print debug information during execution, defaults to False.
Yields:
The output of each step in the graph. The output shape depends on the stream_mode.
Examples:
Using different stream modes with a graph:
```pycon
>>> import operator
>>> from typing_extensions import Annotated, TypedDict
>>> from langgraph.graph import StateGraph
>>> from libs.langgraph.langgraph.constants import START
...
>>> class State(TypedDict):
... alist: Annotated[list, operator.add]
... another_list: Annotated[list, operator.add]
...
>>> builder = StateGraph(State)
>>> builder.add_node("a", lambda _state: {"another_list": ["hi"]})
>>> builder.add_node("b", lambda _state: {"alist": ["there"]})
>>> builder.add_edge("a", "b")
>>> builder.add_edge(START, "a")
>>> graph = builder.compile()
```
With stream_mode="values":
```pycon
>>> for event in graph.stream({"alist": ['Ex for stream_mode="values"']}, stream_mode="values"):
... print(event)
{'alist': ['Ex for stream_mode="values"'], 'another_list': []}
{'alist': ['Ex for stream_mode="values"'], 'another_list': ['hi']}
{'alist': ['Ex for stream_mode="values"', 'there'], 'another_list': ['hi']}
```
With stream_mode="updates":
```pycon
>>> for event in graph.stream({"alist": ['Ex for stream_mode="updates"']}, stream_mode="updates"):
... print(event)
{'a': {'another_list': ['hi']}}
{'b': {'alist': ['there']}}
```
With stream_mode="debug":
```pycon
>>> for event in graph.stream({"alist": ['Ex for stream_mode="debug"']}, stream_mode="debug"):
... print(event)
{'type': 'task', 'timestamp': '2024-06-23T...+00:00', 'step': 1, 'payload': {'id': '...', 'name': 'a', 'input': {'alist': ['Ex for stream_mode="debug"'], 'another_list': []}, 'triggers': ['start:a']}}
{'type': 'task_result', 'timestamp': '2024-06-23T...+00:00', 'step': 1, 'payload': {'id': '...', 'name': 'a', 'result': [('another_list', ['hi'])]}}
{'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'])]}}
```
"""
config = ensure_config(config)
callback_manager = get_callback_manager_for_config(config)
run_manager = callback_manager.on_chain_start(
@@ -856,18 +935,22 @@ class Pregel(
config,
-1,
for_execution=True,
get_next_version=self.checkpointer.get_next_version
if self.checkpointer
else _increment,
get_next_version=(
self.checkpointer.get_next_version
if self.checkpointer
else _increment
),
)
# apply input writes
_apply_writes(
checkpoint,
channels,
input_writes,
self.checkpointer.get_next_version
if self.checkpointer
else _increment,
(
self.checkpointer.get_next_version
if self.checkpointer
else _increment
),
)
# save input checkpoint
yield from put_checkpoint(
@@ -904,9 +987,11 @@ class Pregel(
step,
for_execution=True,
manager=run_manager,
get_next_version=self.checkpointer.get_next_version
if self.checkpointer
else _increment,
get_next_version=(
self.checkpointer.get_next_version
if self.checkpointer
else _increment
),
)
# if no more tasks, we're done
@@ -952,9 +1037,11 @@ class Pregel(
done, inflight = concurrent.futures.wait(
futures,
return_when=concurrent.futures.FIRST_COMPLETED,
timeout=max(0, end_time - time.monotonic())
if end_time
else None,
timeout=(
max(0, end_time - time.monotonic())
if end_time
else None
),
)
for fut in done:
task = futures.pop(fut)
@@ -1002,9 +1089,11 @@ class Pregel(
checkpoint,
channels,
pending_writes,
self.checkpointer.get_next_version
if self.checkpointer
else _increment,
(
self.checkpointer.get_next_version
if self.checkpointer
else _increment
),
)
# yield values output
@@ -1020,14 +1109,14 @@ class Pregel(
{
"source": "loop",
"step": step,
"writes": single(
map_output_updates(output_keys, next_tasks)
)
if self.stream_mode == "updates"
else single(
map_output_values(
output_keys, pending_writes, channels
),
"writes": (
single(map_output_updates(output_keys, next_tasks))
if self.stream_mode == "updates"
else single(
map_output_values(
output_keys, pending_writes, channels
),
)
),
}
)
@@ -1206,18 +1295,22 @@ class Pregel(
config,
-1,
for_execution=True,
get_next_version=self.checkpointer.get_next_version
if self.checkpointer
else _increment,
get_next_version=(
self.checkpointer.get_next_version
if self.checkpointer
else _increment
),
)
# apply input writes
_apply_writes(
checkpoint,
channels,
input_writes,
self.checkpointer.get_next_version
if self.checkpointer
else _increment,
(
self.checkpointer.get_next_version
if self.checkpointer
else _increment
),
)
# save input checkpoint
for chunk in put_checkpoint(
@@ -1251,9 +1344,11 @@ class Pregel(
step,
for_execution=True,
manager=run_manager,
get_next_version=self.checkpointer.get_next_version
if self.checkpointer
else _increment,
get_next_version=(
self.checkpointer.get_next_version
if self.checkpointer
else _increment
),
)
# if no more tasks, we're done
@@ -1300,9 +1395,9 @@ class Pregel(
done, inflight = await asyncio.wait(
futures,
return_when=asyncio.FIRST_COMPLETED,
timeout=max(0, end_time - loop.time())
if end_time
else None,
timeout=(
max(0, end_time - loop.time()) if end_time else None
),
)
for fut in done:
task = futures.pop(fut)
@@ -1352,9 +1447,11 @@ class Pregel(
checkpoint,
channels,
pending_writes,
self.checkpointer.get_next_version
if self.checkpointer
else _increment,
(
self.checkpointer.get_next_version
if self.checkpointer
else _increment
),
)
# yield current values
@@ -1371,12 +1468,14 @@ class Pregel(
{
"source": "loop",
"step": step,
"writes": single(
map_output_updates(output_keys, next_tasks)
)
if self.stream_mode == "updates"
else single(
map_output_values(output_keys, pending_writes, channels)
"writes": (
single(map_output_updates(output_keys, next_tasks))
if self.stream_mode == "updates"
else single(
map_output_values(
output_keys, pending_writes, channels
)
)
),
}
):
@@ -1740,9 +1839,11 @@ def _prepare_next_tasks(
},
),
run_name=packet.node,
callbacks=manager.get_child(f"graph:step:{step}")
if manager
else None,
callbacks=(
manager.get_child(f"graph:step:{step}")
if manager
else None
),
configurable={
# deque.extend is thread-safe
CONFIG_KEY_SEND: partial(
@@ -1820,9 +1921,11 @@ def _prepare_next_tasks(
},
),
run_name=name,
callbacks=manager.get_child(f"graph:step:{step}")
if manager
else None,
callbacks=(
manager.get_child(f"graph:step:{step}")
if manager
else None
),
configurable={
# deque.extend is thread-safe
CONFIG_KEY_SEND: partial(