WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-08-17 21:25:46 +02:00
Code Issues Packages Projects Releases Wiki Activity

style(langgraph): docstring code format pass (#6187)

Browse Source
This commit is contained in:
Mason Daugherty
2025-09-25 13:00:21 -04:00
committed by GitHub
parent 32d66d48eb
commit 90ac06deb6
7 changed files with 86 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/langgraph/langgraph/config.py
+15 -3
View File
@@ -66,14 +66,17 @@ def get_store() -> BaseStore:
store = InMemoryStore()
store.put(("values",), "foo", {"bar": 2})
class State(TypedDict):
foo: int
def my_node(state: State):
my_store = get_store()
stored_value = my_store.get(("values",), "foo").value["bar"]
return {"foo": stored_value + 1}
graph = (
StateGraph(State)
.add_node(my_node)
@@ -85,7 +88,7 @@ def get_store() -> BaseStore:
```
```pycon
{'foo': 3}
{"foo": 3}
```
Example: Using with functional API
@@ -97,16 +100,19 @@ def get_store() -> BaseStore:
store = InMemoryStore()
store.put(("values",), "foo", {"bar": 2})
@task
def my_task(value: int):
my_store = get_store()
stored_value = my_store.get(("values",), "foo").value["bar"]
return stored_value + 1
@entrypoint(store=store)
def workflow(value: int):
return my_task(value).result()
workflow.invoke(1)
```
@@ -134,14 +140,17 @@ def get_stream_writer() -> StreamWriter:
from langgraph.graph import StateGraph, START
from langgraph.config import get_stream_writer
class State(TypedDict):
foo: int
def my_node(state: State):
my_stream_writer = get_stream_writer()
my_stream_writer({"custom_data": "Hello!"})
return {"foo": state["foo"] + 1}
graph = (
StateGraph(State)
.add_node(my_node)
@@ -154,7 +163,7 @@ def get_stream_writer() -> StreamWriter:
```
```pycon
{'custom_data': 'Hello!'}
{"custom_data": "Hello!"}
```
Example: Using with functional API
@@ -162,22 +171,25 @@ def get_stream_writer() -> StreamWriter:
from langgraph.func import entrypoint, task
from langgraph.config import get_stream_writer
@task
def my_task(value: int):
my_stream_writer = get_stream_writer()
my_stream_writer({"custom_data": "Hello!"})
return value + 1
@entrypoint(store=store)
def workflow(value: int):
return my_task(value).result()
for chunk in workflow.stream(1, stream_mode="custom"):
print(chunk)
```
```pycon
{'custom_data': 'Hello!'}
{"custom_data": "Hello!"}
```
"""
runtime = get_config()[CONF][CONFIG_KEY_RUNTIME]
libs/langgraph/langgraph/func/__init__.py
+25 -17
View File
@@ -150,16 +150,19 @@ def task(
```python
from langgraph.func import entrypoint, task
@task
def add_one(a: int) -> int:
return a + 1
@entrypoint()
def add_one(numbers: list[int]) -> list[int]:
futures = [add_one(n) for n in numbers]
results = [f.result() for f in futures]
return results
# Call the entrypoint
add_one.invoke([1, 2, 3]) # Returns [2, 3, 4]
```
@@ -169,15 +172,18 @@ def task(
import asyncio
from langgraph.func import entrypoint, task
@task
async def add_one(a: int) -> int:
return a + 1
@entrypoint()
async def add_one(numbers: list[int]) -> list[int]:
futures = [add_one(n) for n in numbers]
return asyncio.gather(*futures)
# Call the entrypoint
await add_one.ainvoke([1, 2, 3]) # Returns [2, 3, 4]
```
@@ -342,15 +348,13 @@ class entrypoint(Generic[ContextT]):
from langgraph.func import entrypoint
@entrypoint(checkpointer=InMemorySaver())
def my_workflow(input_data: str, previous: Optional[str] = None) -> str:
return "world"
config = {
"configurable": {
"thread_id": "some_thread"
}
}
config = {"configurable": {"thread_id": "some_thread"}}
my_workflow.invoke("hello", config)
```
@@ -367,19 +371,21 @@ class entrypoint(Generic[ContextT]):
from langgraph.func import entrypoint
@entrypoint(checkpointer=InMemorySaver())
def my_workflow(number: int, *, previous: Any = None) -> entrypoint.final[int, int]:
def my_workflow(
number: int,
*,
previous: Any = None,
) -> entrypoint.final[int, int]:
previous = previous or 0
# This will return the previous value to the caller, saving
# 2 * number to the checkpoint, which will be used in the next invocation
# for the `previous` parameter.
return entrypoint.final(value=previous, save=2 * number)
config = {
"configurable": {
"thread_id": "some_thread"
}
}
config = {"configurable": {"thread_id": "some_thread"}}
my_workflow.invoke(3, config) # 0 (previous was None)
my_workflow.invoke(1, config) # 6 (previous was 3 * 2 from the previous invocation)
@@ -434,19 +440,21 @@ class entrypoint(Generic[ContextT]):
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import entrypoint
@entrypoint(checkpointer=InMemorySaver())
def my_workflow(number: int, *, previous: Any = None) -> entrypoint.final[int, int]:
def my_workflow(
number: int,
*,
previous: Any = None,
) -> entrypoint.final[int, int]:
previous = previous or 0
# This will return the previous value to the caller, saving
# 2 * number to the checkpoint, which will be used in the next invocation
# for the `previous` parameter.
return entrypoint.final(value=previous, save=2 * number)
config = {
"configurable": {
"thread_id": "1"
}
}
config = {"configurable": {"thread_id": "1"}}
my_workflow.invoke(3, config) # 0 (previous was None)
my_workflow.invoke(1, config) # 6 (previous was 3 * 2 from the previous invocation)
libs/langgraph/langgraph/graph/message.py
+28 -20
View File
@@ -92,6 +92,7 @@ def add_messages(
Example:
```python title="Basic usage"
from langchain_core.messages import AIMessage, HumanMessage
msgs1 = [HumanMessage(content="Hello", id="1")]
msgs2 = [AIMessage(content="Hi there!", id="2")]
add_messages(msgs1, msgs2)
@@ -110,9 +111,11 @@ def add_messages(
from typing_extensions import TypedDict
from langgraph.graph import StateGraph
class State(TypedDict):
messages: Annotated[list, add_messages]
builder = StateGraph(State)
builder.add_node("chatbot", lambda state: {"messages": [("assistant", "Hello")]})
builder.set_entry_point("chatbot")
@@ -127,30 +130,35 @@ def add_messages(
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, add_messages
class State(TypedDict):
messages: Annotated[list, add_messages(format='langchain-openai')]
messages: Annotated[list, add_messages(format="langchain-openai")]
def chatbot_node(state: State) -> list:
return {"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": "Here's an image:",
"cache_control": {"type": "ephemeral"},
},
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/jpeg",
"data": "1234",
return {
"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": "Here's an image:",
"cache_control": {"type": "ephemeral"},
},
},
]
},
]}
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/jpeg",
"data": "1234",
},
},
],
},
]
}
builder = StateGraph(State)
builder.add_node("chatbot", chatbot_node)
libs/langgraph/langgraph/graph/state.py
+9
View File
@@ -141,25 +141,31 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
from langgraph.graph import StateGraph
from langgraph.runtime import Runtime
def reducer(a: list, b: int | None) -> list:
if b is not None:
return a + [b]
return a
class State(TypedDict):
x: Annotated[list, reducer]
class Context(TypedDict):
r: float
graph = StateGraph(state_schema=State, context_schema=Context)
def node(state: State, runtime: Runtime[Context]) -> dict:
r = runtime.context.get("r", 1.0)
x = state["x"][-1]
next_value = x * r * (1 - x)
return {"x": next_value}
graph.add_node("A", node)
graph.set_entry_point("A")
graph.set_finish_point("A")
@@ -385,12 +391,15 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
from langchain_core.runnables import RunnableConfig
from langgraph.graph import START, StateGraph
class State(TypedDict):
x: int
def my_node(state: State, config: RunnableConfig) -> State:
return {"x": state["x"] + 1}
builder = StateGraph(State)
builder.add_node(my_node) # node name will be 'my_node'
builder.add_edge(START, "my_node")
libs/langgraph/langgraph/graph/ui.py
+1 -1
View File
@@ -188,7 +188,7 @@ def ui_message_reducer(
messages = ui_message_reducer(
[{"type": "ui", "id": "1", "name": "Chat", "props": {}}],
{"type": "remove-ui", "id": "1"}
{"type": "remove-ui", "id": "1"},
)
"""
libs/langgraph/langgraph/pregel/main.py
+7 -5
View File
@@ -386,7 +386,7 @@ class Pregel(
However, for **advanced** use cases, Pregel can be used directly. If you're
not sure whether you need to use Pregel directly, then the answer is probably no
– you should use the Graph API or Functional API instead. These are higher-level
- you should use the Graph API or Functional API instead. These are higher-level
interfaces that will compile down to Pregel under the hood.
Here are some examples to give you a sense of how it works:
@@ -488,7 +488,7 @@ class Pregel(
```
```pycon
{'c': ['foofoo', 'foofoofoofoo']}
{"c": ["foofoo", "foofoofoofoo"]}
```
Example: Using a BinaryOperatorAggregate channel
@@ -516,6 +516,7 @@ class Pregel(
else:
return update
app = Pregel(
nodes={"node1": node1, "node2": node2},
channels={
@@ -524,7 +525,7 @@ class Pregel(
"c": BinaryOperatorAggregate(str, operator=reducer),
},
input_channels=["a"],
output_channels=["c"]
output_channels=["c"],
)
app.invoke({"a": "foo"})
@@ -544,7 +545,8 @@ class Pregel(
from langgraph.pregel import Pregel, NodeBuilder, ChannelWriteEntry
example_node = (
NodeBuilder().subscribe_only("value")
NodeBuilder()
.subscribe_only("value")
.do(lambda x: x + x if len(x) < 10 else None)
.write_to(ChannelWriteEntry(channel="value", skip_none=True))
)
@@ -555,7 +557,7 @@ class Pregel(
"value": EphemeralValue(str),
},
input_channels=["value"],
output_channels=["value"]
output_channels=["value"],
)
app.invoke({"value": "a"})
libs/langgraph/langgraph/types.py
+1 -3
View File
@@ -296,12 +296,10 @@ class Send:
>>> class OverallState(TypedDict):
... subjects: list[str]
... jokes: Annotated[list[str], operator.add]
...
>>> from langgraph.types import Send
>>> from langgraph.graph import END, START
>>> def continue_to_jokes(state: OverallState):
... return [Send("generate_joke", {"subject": s}) for s in state['subjects']]
...
... return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]
>>> from langgraph.graph import StateGraph
>>> builder = StateGraph(OverallState)
>>> builder.add_node("generate_joke", lambda state: {"jokes": [f"Joke about {state['subject']}"]})