mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-08-17 21:25:46 +02:00
more runtime details
This commit is contained in:
@@ -0,0 +1,18 @@
|
||||
# Runtime
|
||||
|
||||
::: langgraph.runtime.Runtime
|
||||
options:
|
||||
show_root_heading: true
|
||||
show_root_full_path: false
|
||||
members:
|
||||
- context
|
||||
- store
|
||||
- stream_writer
|
||||
- previous
|
||||
|
||||
::: langgraph.runtime
|
||||
options:
|
||||
members:
|
||||
- get_runtime
|
||||
|
||||
|
||||
@@ -11,9 +11,9 @@ from langgraph.store.base import BaseStore
|
||||
from langgraph.types import _DC_KWARGS, StreamWriter
|
||||
from langgraph.typing import ContextT
|
||||
|
||||
|
||||
__all__ = ("Runtime", "get_runtime")
|
||||
|
||||
|
||||
def _no_op_stream_writer(_: Any) -> None: ...
|
||||
|
||||
|
||||
@@ -26,11 +26,61 @@ class _RuntimeOverrides(TypedDict, Generic[ContextT], total=False):
|
||||
|
||||
@dataclass(**_DC_KWARGS)
|
||||
class Runtime(Generic[ContextT]):
|
||||
"""Convenience class that bundles run-scoped context and graph configuration.
|
||||
"""Convenience class that bundles run-scoped context and other runtime utilities.
|
||||
|
||||
!!! version-added "Added in version v0.6.0"
|
||||
|
||||
TODO: write a compelling example with the new API.
|
||||
Example:
|
||||
|
||||
```python
|
||||
from typing import TypedDict
|
||||
from langgraph.graph import StateGraph
|
||||
from dataclasses import dataclass
|
||||
from langgraph.runtime import Runtime
|
||||
from langgraph.store.memory import InMemoryStore
|
||||
|
||||
|
||||
@dataclass
|
||||
class Context: # (1)!
|
||||
user_id: str
|
||||
|
||||
|
||||
class State(TypedDict, total=False):
|
||||
response: str
|
||||
|
||||
|
||||
store = InMemoryStore() # (2)!
|
||||
store.put(("users",), "user_123", {"name": "Alice"})
|
||||
|
||||
|
||||
def personalized_greeting(state: State, runtime: Runtime[Context]) -> State:
|
||||
'''Generate personalized greeting using runtime context and store.'''
|
||||
user_id = runtime.context.user_id # (3)!
|
||||
name = "unknown_user"
|
||||
if runtime.store:
|
||||
if memory := runtime.store.get(("users",), user_id):
|
||||
name = memory.value["name"]
|
||||
|
||||
response = f"Hello {name}! Nice to see you again."
|
||||
return {"response": response}
|
||||
|
||||
|
||||
graph = (
|
||||
StateGraph(state_schema=State, context_schema=Context)
|
||||
.add_node("personalized_greeting", personalized_greeting)
|
||||
.set_entry_point("personalized_greeting")
|
||||
.set_finish_point("personalized_greeting")
|
||||
.compile(store=store)
|
||||
)
|
||||
|
||||
result = graph.invoke({}, context=Context(user_id="user_123"))
|
||||
print(result)
|
||||
# > {'response': 'Hello Alice! Nice to see you again.'}
|
||||
```
|
||||
|
||||
1. Define a schema for the runtime context.
|
||||
2. Create a store to persist memories and other information.
|
||||
3. Use the runtime context to access the user_id.
|
||||
"""
|
||||
|
||||
context: ContextT = field(default=None) # type: ignore[assignment]
|
||||
@@ -81,7 +131,7 @@ DEFAULT_RUNTIME = Runtime(
|
||||
|
||||
def get_runtime(context_schema: type[ContextT] | None = None) -> Runtime[ContextT]:
|
||||
"""Get the runtime for the current graph run.
|
||||
|
||||
|
||||
Args:
|
||||
context_schema: Optional schema used for type hinting the return type of the runtime.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user