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

Merge branch 'main' into vb/update-get-state

Browse Source
This commit is contained in:
vbarda
2024-08-26 19:17:26 -04:00
parent 904a1a3471 568044171b
commit bf4dc5d5f6
39 changed files with 1048 additions and 727 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+2 -2
View File
@@ -59,7 +59,7 @@ from langchain_core.messages import HumanMessage
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, StateGraph, MessagesState
from langgraph.graph import END, START, StateGraph, MessagesState
from langgraph.prebuilt import ToolNode
@@ -107,7 +107,7 @@ workflow.add_node("tools", tool_node)
# Set the entrypoint as `agent`
# This means that this node is the first one called
workflow.set_entry_point("agent")
workflow.add_edge(START, "agent")
# We now add a conditional edge
workflow.add_conditional_edges(
docs/_scripts/copy_notebooks.py
+1
View File
@@ -56,6 +56,7 @@ _MANUAL = {
"create-react-agent-memory.ipynb",
"create-react-agent-hitl.ipynb",
"human_in_the_loop/breakpoints.ipynb",
"human_in_the_loop/dynamic_breakpoints.ipynb",
"human_in_the_loop/time-travel.ipynb",
"human_in_the_loop/edit-graph-state.ipynb",
"human_in_the_loop/wait-user-input.ipynb",
docs/docs/cloud/deployment/graph_rebuild.md
+5 -5
View File
@@ -28,7 +28,7 @@ In the standard LangGraph API configuration, the server uses the compiled graph
```python
from langchain_openai import ChatOpenAI
from langgraph.graph import END, MessageGraph
from langgraph.graph import END, START, MessageGraph
model = ChatOpenAI(temperature=0)
@@ -36,7 +36,7 @@ graph_workflow = MessageGraph()
graph_workflow.add_node("agent", model)
graph_workflow.add_edge("agent", END)
graph_workflow.set_entry_point("agent")
graph_workflow.add_edge(START, "agent")
agent = graph_workflow.compile()
```
@@ -60,7 +60,7 @@ To make your graph rebuild on each new run with custom configuration, you need t
```python
from typing import Annotated, TypedDict
from langchain_openai import ChatOpenAI
from langgraph.graph import END, MessageGraph
from langgraph.graph import END, START, MessageGraph
from langgraph.graph.state import StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
@@ -83,7 +83,7 @@ def make_default_graph():
graph_workflow.add_node("agent", call_model)
graph_workflow.add_edge("agent", END)
graph_workflow.set_entry_point("agent")
graph_workflow.add_edge(START, "agent")
agent = graph_workflow.compile()
return agent
@@ -113,7 +113,7 @@ def make_alternative_graph():
graph_workflow.add_node("agent", call_model)
graph_workflow.add_node("tools", tool_node)
graph_workflow.add_edge("tools", "agent")
graph_workflow.set_entry_point("agent")
graph_workflow.add_edge(START, "agent")
graph_workflow.add_conditional_edges("agent", should_continue)
agent = graph_workflow.compile()
docs/docs/cloud/deployment/setup.md
+24 -18
View File
@@ -1,15 +1,14 @@
# How to Set Up a LangGraph Application for Deployment
A LangGraph application must be configured with a [LangGraph API configuration file](../reference/cli.md#configuration-file) in order to be deployed to LangGraph Cloud (or to be self-hosted). This how-to guide discusses the basic steps to setup a LangGraph application for deployment using `requirements.txt` to specify project dependencies.
A LangGraph application must be configured with a [LangGraph API configuration file](../reference/cli.md#configuration-file) in order to be deployed to LangGraph Cloud (or to be self-hosted). This how-to guide discusses the basic steps to setup a LangGraph application for deployment using `requirements.txt` to specify project dependencies.
This walkthrough is based on [this repository](https://github.com/langchain-ai/langgraph-example), which you can play around with to learn more about how to setup your LangGraph application for deployment.
!!! tip "Setup with pyproject.toml"
If you prefer using poetry for dependency management, check out [this how-to guide](./setup_pyproject.md) on using `pyproject.toml` for LangGraph Cloud.
If you prefer using poetry for dependency management, check out [this how-to guide](./setup_pyproject.md) on using `pyproject.toml` for LangGraph Cloud.
!!! tip "Setup with a Monorepo"
If you are interested in deploying a graph located inside a monorepo, take a look at [this](https://github.com/langchain-ai/langgraph-example-monorepo) repository for an example of how to do so.
If you are interested in deploying a graph located inside a monorepo, take a look at [this](https://github.com/langchain-ai/langgraph-example-monorepo) repository for an example of how to do so.
The final repo structure will look something like this:
@@ -35,23 +34,27 @@ After each step, an example file directory is provided to demonstrate how code c
Dependencies can optionally be specified in one of the following files: `pyproject.toml`, `setup.py`, or `requirements.txt`. If none of these files is created, then dependencies can be specified later in the [LangGraph API configuration file](#create-langgraph-api-config).
The dependencies below will be included in the image, you can also use them in your code, as long as with a compatible version range:
```
langgraph>=0.2.0,<0.3.0
langgraph>=0.2.7,<0.3.0
langgraph-checkpoint>=1.0.4
langchain-core>=0.2.27,<0.3.0
langsmith>=0.1.63
orjson>=3.10.1
httpx>=0.27.0
tenacity>=8.3.0
uvicorn>=0.29.0
orjson>=3.9.7
httpx>=0.25.0
tenacity>=8.0.0
uvicorn>=0.26.0
sse-starlette>=2.1.0
uvloop>=0.19.0
httptools>=0.6.1
jsonschema-rs>=0.18.0
uvloop>=0.18.0
httptools>=0.5.0
jsonschema-rs>=0.16.3
croniter>=1.0.1
structlog>=24.4.0
structlog>=23.1.0
redis>=5.0.0,<6.0.0
```
Example `requirements.txt` file:
```
langgraph
langchain_anthropic
@@ -62,6 +65,7 @@ langchain_openai
```
Example file directory:
```bash
my-app/
├── my_agent # all project code lies within here
@@ -73,6 +77,7 @@ my-app/
Environment variables can optionally be specified in a file (e.g. `.env`). See the [Environment Variables reference](../reference/env_var.md) to configure additional variables for a deployment.
Example `.env` file:
```
MY_ENV_VAR_1=foo
MY_ENV_VAR_2=bar
@@ -94,12 +99,11 @@ Implement your graphs! Graphs can be defined in a single file or multiple files.
Example `agent.py` file, which shows how to import from other modules you define (code for the modules is not shown here, please see [this repo](https://github.com/langchain-ai/langgraph-example) to see their implementation):
```python
# my_agent/agent.py
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from langgraph.graph import StateGraph, END, START
from my_agent.utils.nodes import call_model, should_continue, tool_node # import nodes
from my_agent.utils.state import AgentState # import state
@@ -110,7 +114,7 @@ class GraphConfig(TypedDict):
workflow = StateGraph(AgentState, config_schema=GraphConfig)
workflow.add_node("agent", call_model)
workflow.add_node("action", tool_node)
workflow.set_entry_point("agent")
workflow.add_edge(START, "agent")
workflow.add_conditional_edges(
"agent",
should_continue,
@@ -125,9 +129,10 @@ graph = workflow.compile()
```
!!! warning "Assign `CompiledGraph` to Variable"
The build process for LangGraph Cloud requires that the `CompiledGraph` object be assigned to a variable at the top-level of a Python module (alternatively, you can provide [a function that creates a graph](./graph_rebuild.md)).
The build process for LangGraph Cloud requires that the `CompiledGraph` object be assigned to a variable at the top-level of a Python module (alternatively, you can provide [a function that creates a graph](./graph_rebuild.md)).
Example file directory:
```bash
my-app/
├── my_agent # all project code lies within here
@@ -147,6 +152,7 @@ my-app/
Create a [LangGraph API configuration file](../reference/cli.md#configuration-file) called `langgraph.json`. See the [LangGraph CLI reference](../reference/cli.md#configuration-file) for detailed explanations of each key in the JSON object of the configuration file.
Example `langgraph.json` file:
```json
{
"dependencies": ["./my_agent"],
@@ -160,7 +166,7 @@ Example `langgraph.json` file:
Note that the variable name of the `CompiledGraph` appears at the end of the value of each subkey in the top-level `graphs` key (i.e. `:<variable_name>`).
!!! warning "Configuration Location"
The LangGraph API configuration file must be placed in a directory that is at the same level or higher than the Python files that contain compiled graphs and associated dependencies.
The LangGraph API configuration file must be placed in a directory that is at the same level or higher than the Python files that contain compiled graphs and associated dependencies.
Example file directory:
docs/docs/cloud/deployment/setup_pyproject.md
+12 -10
View File
@@ -35,19 +35,21 @@ Dependencies can optionally be specified in one of the following files: `pyproje
The dependencies below will be included in the image, you can also use them in your code, as long as with a compatible version range:
```
langgraph>=0.2.0,<0.3.0
langgraph>=0.2.7,<0.3.0
langgraph-checkpoint>=1.0.4
langchain-core>=0.2.27,<0.3.0
langsmith>=0.1.63
orjson>=3.10.1
httpx>=0.27.0
tenacity>=8.3.0
uvicorn>=0.29.0
orjson>=3.9.7
httpx>=0.25.0
tenacity>=8.0.0
uvicorn>=0.26.0
sse-starlette>=2.1.0
uvloop>=0.19.0
httptools>=0.6.1
jsonschema-rs>=0.18.0
uvloop>=0.18.0
httptools>=0.5.0
jsonschema-rs>=0.16.3
croniter>=1.0.1
structlog>=24.4.0
redis>=5.0.8,<6.0.0
```
Example `pyproject.toml` file:
@@ -109,7 +111,7 @@ Example `agent.py` file, which shows how to import from other modules you define
# my_agent/agent.py
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from langgraph.graph import StateGraph, END, START
from my_agent.utils.nodes import call_model, should_continue, tool_node # import nodes
from my_agent.utils.state import AgentState # import state
@@ -120,7 +122,7 @@ class GraphConfig(TypedDict):
workflow = StateGraph(AgentState, config_schema=GraphConfig)
workflow.add_node("agent", call_model)
workflow.add_node("action", tool_node)
workflow.set_entry_point("agent")
workflow.add_edge(START, "agent")
workflow.add_conditional_edges(
"agent",
should_continue,
docs/docs/cloud/reference/api/api_ref.md
+17
View File
@@ -3,3 +3,20 @@
The LangGraph Cloud API reference is available with each deployment at the `/docs` URL path (e.g. `http://localhost:8124/docs`).
Click <a href="/langgraph/cloud/reference/api/api_ref.html" target="_blank">here</a> to view the API reference.
## Authentication
For deployments to LangGraph Cloud, authentication is required. Pass the `X-Api-Key` header with each request to the LangGraph Cloud API. The value of the header should be set to a valid LangSmith API key for the organization where the API is deployed.
Example `curl` command:
```shell
curl --request POST \
--url http://localhost:8124/assistants/search \
--header 'Content-Type: application/json' \
--header 'X-Api-Key: LANGSMITH_API_KEY' \
--data '{
"metadata": {},
"limit": 10,
"offset": 0
}'
```
docs/docs/how-tos/index.md
+1
View File
@@ -35,6 +35,7 @@ One of LangGraph's main benefits is that it makes human-in-the-loop workflows ea
These guides cover common examples of that.
- [How to add breakpoints](human_in_the_loop/breakpoints.ipynb)
- [How to add dynamic breakpoints](human_in_the_loop/dynamic_breakpoints.ipynb)
- [How to edit graph state](human_in_the_loop/edit-graph-state.ipynb)
- [How to wait for user input](human_in_the_loop/wait-user-input.ipynb)
- [How to view and update past graph state](human_in_the_loop/time-travel.ipynb)
docs/mkdocs.yml
+1
View File
@@ -139,6 +139,7 @@ nav:
- Create custom checkpointer using Redis: how-tos/persistence_redis.ipynb
- Human-in-the-loop:
- Add breakpoints: how-tos/human_in_the_loop/breakpoints.ipynb
- Add dynamic breakpoints: how-tos/human_in_the_loop/dynamic_breakpoints.ipynb
- Wait for user input: how-tos/human_in_the_loop/wait-user-input.ipynb
- View and update past graph state: how-tos/human_in_the_loop/time-travel.ipynb
- Edit graph state: how-tos/human_in_the_loop/edit-graph-state.ipynb
examples/docs/quickstart.ipynb
+113 -30
View File
File diff suppressed because one or more lines are too long
examples/human_in_the_loop/dynamic_breakpoints.ipynb
+417
View File
File diff suppressed because one or more lines are too long
examples/streaming-events-from-within-tools-without-langchain.ipynb
+6 -195
View File
@@ -30,10 +30,7 @@
"id": "47f79af8-58d8-4a48-8d9a-88823d88701f",
"metadata": {},
"outputs": [],
"source": [
"%%capture --no-stderr\n",
"%pip install -U langgraph openai"
]
"source": ["%%capture --no-stderr\n%pip install -U langgraph openai"]
},
{
"cell_type": "code",
@@ -49,18 +46,7 @@
]
}
],
"source": [
"import getpass\n",
"import os\n",
"\n",
"\n",
"def _set_env(var: str):\n",
" if not os.environ.get(var):\n",
" os.environ[var] = getpass.getpass(f\"{var}: \")\n",
"\n",
"\n",
"_set_env(\"OPENAI_API_KEY\")"
]
"source": ["import getpass\nimport os\n\n\ndef _set_env(var: str):\n if not os.environ.get(var):\n os.environ[var] = getpass.getpass(f\"{var}: \")\n\n\n_set_env(\"OPENAI_API_KEY\")"]
},
{
"cell_type": "markdown",
@@ -84,94 +70,7 @@
"id": "d59234f9-173e-469d-a725-c13e0979663e",
"metadata": {},
"outputs": [],
"source": [
"from openai import AsyncOpenAI\n",
"from langchain_core.language_models.chat_models import ChatGenerationChunk\n",
"from langchain_core.messages import AIMessageChunk\n",
"from langchain_core.runnables.config import (\n",
" ensure_config,\n",
" get_callback_manager_for_config,\n",
")\n",
"\n",
"openai_client = AsyncOpenAI()\n",
"# define tool schema for openai tool calling\n",
"\n",
"tool = {\n",
" \"type\": \"function\",\n",
" \"function\": {\n",
" \"name\": \"get_items\",\n",
" \"description\": \"Use this tool to look up which items are in the given place.\",\n",
" \"parameters\": {\n",
" \"type\": \"object\",\n",
" \"properties\": {\"place\": {\"type\": \"string\"}},\n",
" \"required\": [\"place\"],\n",
" },\n",
" },\n",
"}\n",
"\n",
"\n",
"async def call_model(state, config=None):\n",
" config = ensure_config(config | {\"tags\": [\"agent_llm\"]})\n",
" callback_manager = get_callback_manager_for_config(config)\n",
" messages = state[\"messages\"]\n",
"\n",
" llm_run_manager = callback_manager.on_chat_model_start({}, [messages])[0]\n",
" response = await openai_client.chat.completions.create(\n",
" messages=messages, model=\"gpt-3.5-turbo\", tools=[tool], stream=True\n",
" )\n",
"\n",
" response_content = \"\"\n",
" role = None\n",
"\n",
" tool_call_id = None\n",
" tool_call_function_name = None\n",
" tool_call_function_arguments = \"\"\n",
" async for chunk in response:\n",
" delta = chunk.choices[0].delta\n",
" if delta.role is not None:\n",
" role = delta.role\n",
"\n",
" if delta.content:\n",
" response_content += delta.content\n",
" llm_run_manager.on_llm_new_token(delta.content)\n",
"\n",
" if delta.tool_calls:\n",
" # note: for simplicity we're only handling a single tool call here\n",
" if delta.tool_calls[0].function.name is not None:\n",
" tool_call_function_name = delta.tool_calls[0].function.name\n",
" tool_call_id = delta.tool_calls[0].id\n",
"\n",
" # note: we're wrapping the tools calls in ChatGenerationChunk so that the events from .astream_events in the graph can render tool calls correctly\n",
" tool_call_chunk = ChatGenerationChunk(\n",
" message=AIMessageChunk(\n",
" content=\"\",\n",
" additional_kwargs={\"tool_calls\": [delta.tool_calls[0].dict()]},\n",
" )\n",
" )\n",
" llm_run_manager.on_llm_new_token(\"\", chunk=tool_call_chunk)\n",
" tool_call_function_arguments += delta.tool_calls[0].function.arguments\n",
"\n",
" if tool_call_function_name is not None:\n",
" tool_calls = [\n",
" {\n",
" \"id\": tool_call_id,\n",
" \"function\": {\n",
" \"name\": tool_call_function_name,\n",
" \"arguments\": tool_call_function_arguments,\n",
" },\n",
" \"type\": \"function\",\n",
" }\n",
" ]\n",
" else:\n",
" tool_calls = None\n",
"\n",
" response_message = {\n",
" \"role\": role,\n",
" \"content\": response_content,\n",
" \"tool_calls\": tool_calls,\n",
" }\n",
" return {\"messages\": [response_message]}"
]
"source": ["from openai import AsyncOpenAI\nfrom langchain_core.language_models.chat_models import ChatGenerationChunk\nfrom langchain_core.messages import AIMessageChunk\nfrom langchain_core.runnables.config import (\n ensure_config,\n get_callback_manager_for_config,\n)\n\nopenai_client = AsyncOpenAI()\n# define tool schema for openai tool calling\n\ntool = {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"get_items\",\n \"description\": \"Use this tool to look up which items are in the given place.\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\"place\": {\"type\": \"string\"}},\n \"required\": [\"place\"],\n },\n },\n}\n\n\nasync def call_model(state, config=None):\n config = ensure_config(config | {\"tags\": [\"agent_llm\"]})\n callback_manager = get_callback_manager_for_config(config)\n messages = state[\"messages\"]\n\n llm_run_manager = callback_manager.on_chat_model_start({}, [messages])[0]\n response = await openai_client.chat.completions.create(\n messages=messages, model=\"gpt-3.5-turbo\", tools=[tool], stream=True\n )\n\n response_content = \"\"\n role = None\n\n tool_call_id = None\n tool_call_function_name = None\n tool_call_function_arguments = \"\"\n async for chunk in response:\n delta = chunk.choices[0].delta\n if delta.role is not None:\n role = delta.role\n\n if delta.content:\n response_content += delta.content\n llm_run_manager.on_llm_new_token(delta.content)\n\n if delta.tool_calls:\n # note: for simplicity we're only handling a single tool call here\n if delta.tool_calls[0].function.name is not None:\n tool_call_function_name = delta.tool_calls[0].function.name\n tool_call_id = delta.tool_calls[0].id\n\n # note: we're wrapping the tools calls in ChatGenerationChunk so that the events from .astream_events in the graph can render tool calls correctly\n tool_call_chunk = ChatGenerationChunk(\n message=AIMessageChunk(\n content=\"\",\n additional_kwargs={\"tool_calls\": [delta.tool_calls[0].dict()]},\n )\n )\n llm_run_manager.on_llm_new_token(\"\", chunk=tool_call_chunk)\n tool_call_function_arguments += delta.tool_calls[0].function.arguments\n\n if tool_call_function_name is not None:\n tool_calls = [\n {\n \"id\": tool_call_id,\n \"function\": {\n \"name\": tool_call_function_name,\n \"arguments\": tool_call_function_arguments,\n },\n \"type\": \"function\",\n }\n ]\n else:\n tool_calls = None\n\n response_message = {\n \"role\": role,\n \"content\": response_content,\n \"tool_calls\": tool_calls,\n }\n return {\"messages\": [response_message]}"]
},
{
"cell_type": "markdown",
@@ -187,62 +86,7 @@
"id": "b90941d8-afe4-42ec-9262-9c3b87c3b1ec",
"metadata": {},
"outputs": [],
"source": [
"import json\n",
"from langchain_core.callbacks import adispatch_custom_event\n",
"\n",
"\n",
"async def get_items(place: str) -> str:\n",
" \"\"\"Use this tool to look up which items are in the given place.\"\"\"\n",
"\n",
" # this can be replaced with any actual streaming logic that you might have\n",
" def stream(place: str):\n",
" if \"bed\" in place: # For under the bed\n",
" yield from [\"socks\", \"shoes\", \"dust bunnies\"]\n",
" elif \"shelf\" in place: # For 'shelf'\n",
" yield from [\"books\", \"penciles\", \"pictures\"]\n",
" else: # if the agent decides to ask about a different place\n",
" yield \"cat snacks\"\n",
"\n",
" tokens = []\n",
" for token in stream(place):\n",
" await adispatch_custom_event(\n",
" # this will allow you to filter events by name\n",
" \"tool_call_token_stream\",\n",
" {\n",
" \"function_name\": \"get_items\",\n",
" \"arguments\": {\"place\": place},\n",
" \"tool_output_token\": token,\n",
" },\n",
" # this will allow you to filter events by tags\n",
" config={\"tags\": [\"tool_call\"]},\n",
" )\n",
" tokens.append(token)\n",
"\n",
" return \", \".join(tokens)\n",
"\n",
"\n",
"# define mapping to look up functions when running tools\n",
"function_name_to_function = {\"get_items\": get_items}\n",
"\n",
"\n",
"async def call_tools(state):\n",
" messages = state[\"messages\"]\n",
"\n",
" tool_call = messages[-1][\"tool_calls\"][0]\n",
" function_name = tool_call[\"function\"][\"name\"]\n",
" function_arguments = tool_call[\"function\"][\"arguments\"]\n",
" arguments = json.loads(function_arguments)\n",
"\n",
" function_response = await function_name_to_function[function_name](**arguments)\n",
" tool_message = {\n",
" \"tool_call_id\": tool_call[\"id\"],\n",
" \"role\": \"tool\",\n",
" \"name\": function_name,\n",
" \"content\": function_response,\n",
" }\n",
" return {\"messages\": [tool_message]}"
]
"source": ["import json\nfrom langchain_core.callbacks import adispatch_custom_event\n\n\nasync def get_items(place: str) -> str:\n \"\"\"Use this tool to look up which items are in the given place.\"\"\"\n\n # this can be replaced with any actual streaming logic that you might have\n def stream(place: str):\n if \"bed\" in place: # For under the bed\n yield from [\"socks\", \"shoes\", \"dust bunnies\"]\n elif \"shelf\" in place: # For 'shelf'\n yield from [\"books\", \"penciles\", \"pictures\"]\n else: # if the agent decides to ask about a different place\n yield \"cat snacks\"\n\n tokens = []\n for token in stream(place):\n await adispatch_custom_event(\n # this will allow you to filter events by name\n \"tool_call_token_stream\",\n {\n \"function_name\": \"get_items\",\n \"arguments\": {\"place\": place},\n \"tool_output_token\": token,\n },\n # this will allow you to filter events by tags\n config={\"tags\": [\"tool_call\"]},\n )\n tokens.append(token)\n\n return \", \".join(tokens)\n\n\n# define mapping to look up functions when running tools\nfunction_name_to_function = {\"get_items\": get_items}\n\n\nasync def call_tools(state):\n messages = state[\"messages\"]\n\n tool_call = messages[-1][\"tool_calls\"][0]\n function_name = tool_call[\"function\"][\"name\"]\n function_arguments = tool_call[\"function\"][\"arguments\"]\n arguments = json.loads(function_arguments)\n\n function_response = await function_name_to_function[function_name](**arguments)\n tool_message = {\n \"tool_call_id\": tool_call[\"id\"],\n \"role\": \"tool\",\n \"name\": function_name,\n \"content\": function_response,\n }\n return {\"messages\": [tool_message]}"]
},
{
"cell_type": "markdown",
@@ -258,33 +102,7 @@
"id": "228260be-1f9a-4195-80e0-9604f8a5dba6",
"metadata": {},
"outputs": [],
"source": [
"import operator\n",
"from typing import Annotated, TypedDict, Literal\n",
"\n",
"from langgraph.graph import StateGraph, END\n",
"\n",
"\n",
"class State(TypedDict):\n",
" messages: Annotated[list, operator.add]\n",
"\n",
"\n",
"def should_continue(state) -> Literal[\"tools\", END]:\n",
" messages = state[\"messages\"]\n",
" last_message = messages[-1]\n",
" if last_message[\"tool_calls\"]:\n",
" return \"tools\"\n",
" return END\n",
"\n",
"\n",
"workflow = StateGraph(State)\n",
"workflow.set_entry_point(\"model\")\n",
"workflow.add_node(\"model\", call_model) # i.e. our \"agent\"\n",
"workflow.add_node(\"tools\", call_tools)\n",
"workflow.add_conditional_edges(\"model\", should_continue)\n",
"workflow.add_edge(\"tools\", \"model\")\n",
"graph = workflow.compile()"
]
"source": ["import operator\nfrom typing import Annotated, TypedDict, Literal\n\nfrom langgraph.graph import StateGraph, END, START\n\n\nclass State(TypedDict):\n messages: Annotated[list, operator.add]\n\n\ndef should_continue(state) -> Literal[\"tools\", END]:\n messages = state[\"messages\"]\n last_message = messages[-1]\n if last_message[\"tool_calls\"]:\n return \"tools\"\n return END\n\n\nworkflow = StateGraph(State)\nworkflow.add_edge(START, \"model\")\nworkflow.add_node(\"model\", call_model) # i.e. our \"agent\"\nworkflow.add_node(\"tools\", call_tools)\nworkflow.add_conditional_edges(\"model\", should_continue)\nworkflow.add_edge(\"tools\", \"model\")\ngraph = workflow.compile()"]
},
{
"cell_type": "markdown",
@@ -318,14 +136,7 @@
]
}
],
"source": [
"async for event in graph.astream_events(\n",
" {\"messages\": [{\"role\": \"user\", \"content\": \"what's in the bedroom\"}]}, version=\"v2\"\n",
"):\n",
" tags = event.get(\"tags\", [])\n",
" if event[\"event\"] == \"on_custom_event\" and \"tool_call\" in tags:\n",
" print(\"Tool token\", event[\"data\"][\"tool_output_token\"])"
]
"source": ["async for event in graph.astream_events(\n {\"messages\": [{\"role\": \"user\", \"content\": \"what's in the bedroom\"}]}, version=\"v2\"\n):\n tags = event.get(\"tags\", [])\n if event[\"event\"] == \"on_custom_event\" and \"tool_call\" in tags:\n print(\"Tool token\", event[\"data\"][\"tool_output_token\"])"]
}
],
"metadata": {
examples/streaming-tokens-without-langchain.ipynb
+7 -175
View File
@@ -30,10 +30,7 @@
"id": "47f79af8-58d8-4a48-8d9a-88823d88701f",
"metadata": {},
"outputs": [],
"source": [
"%%capture --no-stderr\n",
"%pip install -U langgraph openai"
]
"source": ["%%capture --no-stderr\n%pip install -U langgraph openai"]
},
{
"cell_type": "code",
@@ -49,18 +46,7 @@
]
}
],
"source": [
"import getpass\n",
"import os\n",
"\n",
"\n",
"def _set_env(var: str):\n",
" if not os.environ.get(var):\n",
" os.environ[var] = getpass.getpass(f\"{var}: \")\n",
"\n",
"\n",
"_set_env(\"OPENAI_API_KEY\")"
]
"source": ["import getpass\nimport os\n\n\ndef _set_env(var: str):\n if not os.environ.get(var):\n os.environ[var] = getpass.getpass(f\"{var}: \")\n\n\n_set_env(\"OPENAI_API_KEY\")"]
},
{
"cell_type": "markdown",
@@ -84,94 +70,7 @@
"id": "d59234f9-173e-469d-a725-c13e0979663e",
"metadata": {},
"outputs": [],
"source": [
"from openai import AsyncOpenAI\n",
"from langchain_core.language_models.chat_models import ChatGenerationChunk\n",
"from langchain_core.messages import AIMessageChunk\n",
"from langchain_core.runnables.config import (\n",
" ensure_config,\n",
" get_callback_manager_for_config,\n",
")\n",
"\n",
"openai_client = AsyncOpenAI()\n",
"# define tool schema for openai tool calling\n",
"\n",
"tool = {\n",
" \"type\": \"function\",\n",
" \"function\": {\n",
" \"name\": \"get_items\",\n",
" \"description\": \"Use this tool to look up which items are in the given place.\",\n",
" \"parameters\": {\n",
" \"type\": \"object\",\n",
" \"properties\": {\"place\": {\"type\": \"string\"}},\n",
" \"required\": [\"place\"],\n",
" },\n",
" },\n",
"}\n",
"\n",
"\n",
"async def call_model(state, config=None):\n",
" config = ensure_config(config | {\"tags\": [\"agent_llm\"]})\n",
" callback_manager = get_callback_manager_for_config(config)\n",
" messages = state[\"messages\"]\n",
"\n",
" llm_run_manager = callback_manager.on_chat_model_start({}, [messages])[0]\n",
" response = await openai_client.chat.completions.create(\n",
" messages=messages, model=\"gpt-3.5-turbo\", tools=[tool], stream=True\n",
" )\n",
"\n",
" response_content = \"\"\n",
" role = None\n",
"\n",
" tool_call_id = None\n",
" tool_call_function_name = None\n",
" tool_call_function_arguments = \"\"\n",
" async for chunk in response:\n",
" delta = chunk.choices[0].delta\n",
" if delta.role is not None:\n",
" role = delta.role\n",
"\n",
" if delta.content:\n",
" response_content += delta.content\n",
" llm_run_manager.on_llm_new_token(delta.content)\n",
"\n",
" if delta.tool_calls:\n",
" # note: for simplicity we're only handling a single tool call here\n",
" if delta.tool_calls[0].function.name is not None:\n",
" tool_call_function_name = delta.tool_calls[0].function.name\n",
" tool_call_id = delta.tool_calls[0].id\n",
"\n",
" # note: we're wrapping the tools calls in ChatGenerationChunk so that the events from .astream_events in the graph can render tool calls correctly\n",
" tool_call_chunk = ChatGenerationChunk(\n",
" message=AIMessageChunk(\n",
" content=\"\",\n",
" additional_kwargs={\"tool_calls\": [delta.tool_calls[0].dict()]},\n",
" )\n",
" )\n",
" llm_run_manager.on_llm_new_token(\"\", chunk=tool_call_chunk)\n",
" tool_call_function_arguments += delta.tool_calls[0].function.arguments\n",
"\n",
" if tool_call_function_name is not None:\n",
" tool_calls = [\n",
" {\n",
" \"id\": tool_call_id,\n",
" \"function\": {\n",
" \"name\": tool_call_function_name,\n",
" \"arguments\": tool_call_function_arguments,\n",
" },\n",
" \"type\": \"function\",\n",
" }\n",
" ]\n",
" else:\n",
" tool_calls = None\n",
"\n",
" response_message = {\n",
" \"role\": role,\n",
" \"content\": response_content,\n",
" \"tool_calls\": tool_calls,\n",
" }\n",
" return {\"messages\": [response_message]}"
]
"source": ["from openai import AsyncOpenAI\nfrom langchain_core.language_models.chat_models import ChatGenerationChunk\nfrom langchain_core.messages import AIMessageChunk\nfrom langchain_core.runnables.config import (\n ensure_config,\n get_callback_manager_for_config,\n)\n\nopenai_client = AsyncOpenAI()\n# define tool schema for openai tool calling\n\ntool = {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"get_items\",\n \"description\": \"Use this tool to look up which items are in the given place.\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\"place\": {\"type\": \"string\"}},\n \"required\": [\"place\"],\n },\n },\n}\n\n\nasync def call_model(state, config=None):\n config = ensure_config(config | {\"tags\": [\"agent_llm\"]})\n callback_manager = get_callback_manager_for_config(config)\n messages = state[\"messages\"]\n\n llm_run_manager = callback_manager.on_chat_model_start({}, [messages])[0]\n response = await openai_client.chat.completions.create(\n messages=messages, model=\"gpt-3.5-turbo\", tools=[tool], stream=True\n )\n\n response_content = \"\"\n role = None\n\n tool_call_id = None\n tool_call_function_name = None\n tool_call_function_arguments = \"\"\n async for chunk in response:\n delta = chunk.choices[0].delta\n if delta.role is not None:\n role = delta.role\n\n if delta.content:\n response_content += delta.content\n llm_run_manager.on_llm_new_token(delta.content)\n\n if delta.tool_calls:\n # note: for simplicity we're only handling a single tool call here\n if delta.tool_calls[0].function.name is not None:\n tool_call_function_name = delta.tool_calls[0].function.name\n tool_call_id = delta.tool_calls[0].id\n\n # note: we're wrapping the tools calls in ChatGenerationChunk so that the events from .astream_events in the graph can render tool calls correctly\n tool_call_chunk = ChatGenerationChunk(\n message=AIMessageChunk(\n content=\"\",\n additional_kwargs={\"tool_calls\": [delta.tool_calls[0].dict()]},\n )\n )\n llm_run_manager.on_llm_new_token(\"\", chunk=tool_call_chunk)\n tool_call_function_arguments += delta.tool_calls[0].function.arguments\n\n if tool_call_function_name is not None:\n tool_calls = [\n {\n \"id\": tool_call_id,\n \"function\": {\n \"name\": tool_call_function_name,\n \"arguments\": tool_call_function_arguments,\n },\n \"type\": \"function\",\n }\n ]\n else:\n tool_calls = None\n\n response_message = {\n \"role\": role,\n \"content\": response_content,\n \"tool_calls\": tool_calls,\n }\n return {\"messages\": [response_message]}"]
},
{
"cell_type": "markdown",
@@ -187,41 +86,7 @@
"id": "b756ea32",
"metadata": {},
"outputs": [],
"source": [
"import json\n",
"\n",
"\n",
"async def get_items(place: str) -> str:\n",
" \"\"\"Use this tool to look up which items are in the given place.\"\"\"\n",
" if \"bed\" in place: # For under the bed\n",
" return \"socks, shoes and dust bunnies\"\n",
" if \"shelf\" in place: # For 'shelf'\n",
" return \"books, penciles and pictures\"\n",
" else: # if the agent decides to ask about a different place\n",
" return \"cat snacks\"\n",
"\n",
"\n",
"# define mapping to look up functions when running tools\n",
"function_name_to_function = {\"get_items\": get_items}\n",
"\n",
"\n",
"async def call_tools(state):\n",
" messages = state[\"messages\"]\n",
"\n",
" tool_call = messages[-1][\"tool_calls\"][0]\n",
" function_name = tool_call[\"function\"][\"name\"]\n",
" function_arguments = tool_call[\"function\"][\"arguments\"]\n",
" arguments = json.loads(function_arguments)\n",
"\n",
" function_response = await function_name_to_function[function_name](**arguments)\n",
" tool_message = {\n",
" \"tool_call_id\": tool_call[\"id\"],\n",
" \"role\": \"tool\",\n",
" \"name\": function_name,\n",
" \"content\": function_response,\n",
" }\n",
" return {\"messages\": [tool_message]}"
]
"source": ["import json\n\n\nasync def get_items(place: str) -> str:\n \"\"\"Use this tool to look up which items are in the given place.\"\"\"\n if \"bed\" in place: # For under the bed\n return \"socks, shoes and dust bunnies\"\n if \"shelf\" in place: # For 'shelf'\n return \"books, penciles and pictures\"\n else: # if the agent decides to ask about a different place\n return \"cat snacks\"\n\n\n# define mapping to look up functions when running tools\nfunction_name_to_function = {\"get_items\": get_items}\n\n\nasync def call_tools(state):\n messages = state[\"messages\"]\n\n tool_call = messages[-1][\"tool_calls\"][0]\n function_name = tool_call[\"function\"][\"name\"]\n function_arguments = tool_call[\"function\"][\"arguments\"]\n arguments = json.loads(function_arguments)\n\n function_response = await function_name_to_function[function_name](**arguments)\n tool_message = {\n \"tool_call_id\": tool_call[\"id\"],\n \"role\": \"tool\",\n \"name\": function_name,\n \"content\": function_response,\n }\n return {\"messages\": [tool_message]}"]
},
{
"cell_type": "markdown",
@@ -237,33 +102,7 @@
"id": "228260be-1f9a-4195-80e0-9604f8a5dba6",
"metadata": {},
"outputs": [],
"source": [
"import operator\n",
"from typing import Annotated, TypedDict, Literal\n",
"\n",
"from langgraph.graph import StateGraph, END\n",
"\n",
"\n",
"class State(TypedDict):\n",
" messages: Annotated[list, operator.add]\n",
"\n",
"\n",
"def should_continue(state) -> Literal[\"tools\", END]:\n",
" messages = state[\"messages\"]\n",
" last_message = messages[-1]\n",
" if last_message[\"tool_calls\"]:\n",
" return \"tools\"\n",
" return END\n",
"\n",
"\n",
"workflow = StateGraph(State)\n",
"workflow.set_entry_point(\"model\")\n",
"workflow.add_node(\"model\", call_model) # i.e. our \"agent\"\n",
"workflow.add_node(\"tools\", call_tools)\n",
"workflow.add_conditional_edges(\"model\", should_continue)\n",
"workflow.add_edge(\"tools\", \"model\")\n",
"graph = workflow.compile()"
]
"source": ["import operator\nfrom typing import Annotated, TypedDict, Literal\n\nfrom langgraph.graph import StateGraph, END, START\n\n\nclass State(TypedDict):\n messages: Annotated[list, operator.add]\n\n\ndef should_continue(state) -> Literal[\"tools\", END]:\n messages = state[\"messages\"]\n last_message = messages[-1]\n if last_message[\"tool_calls\"]:\n return \"tools\"\n return END\n\n\nworkflow = StateGraph(State)\nworkflow.add_edge(START, \"model\")\nworkflow.add_node(\"model\", call_model) # i.e. our \"agent\"\nworkflow.add_node(\"tools\", call_tools)\nworkflow.add_conditional_edges(\"model\", should_continue)\nworkflow.add_edge(\"tools\", \"model\")\ngraph = workflow.compile()"]
},
{
"cell_type": "markdown",
@@ -328,14 +167,7 @@
]
}
],
"source": [
"async for event in graph.astream_events(\n",
" {\"messages\": [{\"role\": \"user\", \"content\": \"what's in the bedroom\"}]}, version=\"v2\"\n",
"):\n",
" tags = event.get(\"tags\", [])\n",
" if event[\"event\"] == \"on_chat_model_stream\" and \"agent_llm\" in tags:\n",
" print(\"LLM token\", event[\"data\"][\"chunk\"].dict())"
]
"source": ["async for event in graph.astream_events(\n {\"messages\": [{\"role\": \"user\", \"content\": \"what's in the bedroom\"}]}, version=\"v2\"\n):\n tags = event.get(\"tags\", [])\n if event[\"event\"] == \"on_chat_model_stream\" and \"agent_llm\" in tags:\n print(\"LLM token\", event[\"data\"][\"chunk\"].dict())"]
},
{
"cell_type": "code",
@@ -343,7 +175,7 @@
"id": "adb0f7bc-6e51-478e-bd32-8f72df072d6c",
"metadata": {},
"outputs": [],
"source": []
"source": [""]
}
],
"metadata": {
examples/tutorials/rag-agent-testing-local.ipynb
+1 -1
View File
@@ -355,7 +355,7 @@
"workflow.add_node(\"web_search\", web_search) # web search\n",
"\n",
"# Build graph\n",
"workflow.set_entry_point(\"retrieve\")\n",
"workflow.add_edge(START, retrieve)\n",
"workflow.add_edge(\"retrieve\", \"grade_documents\")\n",
"workflow.add_conditional_edges(\n",
" \"grade_documents\",\n",
examples/web-navigation/web_voyager.ipynb
+2 -2
View File
@@ -457,13 +457,13 @@
"source": [
"from langchain_core.runnables import RunnableLambda\n",
"\n",
"from langgraph.graph import END, StateGraph\n",
"from langgraph.graph import END, START, StateGraph\n",
"\n",
"graph_builder = StateGraph(AgentState)\n",
"\n",
"\n",
"graph_builder.add_node(\"agent\", agent)\n",
"graph_builder.set_entry_point(\"agent\")\n",
"graph_builder.add_edge(START, \"agent\")\n",
"\n",
"graph_builder.add_node(\"update_scratchpad\", update_scratchpad)\n",
"graph_builder.add_edge(\"update_scratchpad\", \"agent\")\n",
libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py
+7 -3
View File
@@ -51,9 +51,13 @@ class JsonPlusSerializer(SerializerProtocol):
if isinstance(obj, Serializable):
return obj.to_json()
elif hasattr(obj, "model_dump") and callable(obj.model_dump):
return self._encode_constructor_args(obj.__class__, kwargs=obj.model_dump())
return self._encode_constructor_args(
obj.__class__, method="model_construct", kwargs=obj.model_dump()
)
elif hasattr(obj, "dict") and callable(obj.dict):
return self._encode_constructor_args(obj.__class__, kwargs=obj.dict())
return self._encode_constructor_args(
obj.__class__, method="construct", kwargs=obj.dict()
)
elif isinstance(obj, pathlib.Path):
return self._encode_constructor_args(pathlib.Path, args=obj.parts)
elif isinstance(obj, re.Pattern):
@@ -146,7 +150,7 @@ class JsonPlusSerializer(SerializerProtocol):
return method(**value["kwargs"])
else:
return method()
except (ImportError, AttributeError):
except (ImportError, AttributeError, TypeError):
return None
return LC_REVIVER(value)
libs/checkpoint/pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "langgraph-checkpoint"
version = "1.0.4"
version = "1.0.6"
description = "Library with base interfaces for LangGraph checkpoint savers."
authors = []
license = "MIT"
libs/checkpoint/tests/test_jsonplus.py
+1 -1
View File
@@ -122,7 +122,7 @@ def test_serde_jsonplus() -> None:
assert dumped == (
"json",
b"""{"path": {"lc": 2, "type": "constructor", "id": ["pathlib", "Path"], "method": null, "args": ["foo", "bar"], "kwargs": {}}, "re": {"lc": 2, "type": "constructor", "id": ["re", "compile"], "method": null, "args": ["foo", 48], "kwargs": {}}, "decimal": {"lc": 2, "type": "constructor", "id": ["decimal", "Decimal"], "method": null, "args": ["1.10101"], "kwargs": {}}, "ip4": {"lc": 2, "type": "constructor", "id": ["ipaddress", "IPv4Address"], "method": null, "args": ["192.168.0.1"], "kwargs": {}}, "deque": {"lc": 2, "type": "constructor", "id": ["collections", "deque"], "method": null, "args": [[1, 2, 3]], "kwargs": {}}, "tzn": {"lc": 2, "type": "constructor", "id": ["zoneinfo", "ZoneInfo"], "method": null, "args": ["America/New_York"], "kwargs": {}}, "date": {"lc": 2, "type": "constructor", "id": ["datetime", "date"], "method": null, "args": [2024, 4, 19], "kwargs": {}}, "time": {"lc": 2, "type": "constructor", "id": ["datetime", "time"], "method": null, "args": [23, 4, 57, 51022, {"lc": 2, "type": "constructor", "id": ["datetime", "timezone"], "method": null, "args": [{"lc": 2, "type": "constructor", "id": ["datetime", "timedelta"], "method": null, "args": [0, 86340, 0], "kwargs": {}}], "kwargs": {}}], "kwargs": {"fold": 0}}, "uid": {"lc": 2, "type": "constructor", "id": ["uuid", "UUID"], "method": null, "args": ["00000000000000000000000000000001"], "kwargs": {}}, "timestamp": {"lc": 2, "type": "constructor", "id": ["datetime", "datetime"], "method": "fromisoformat", "args": ["2024-04-19T23:04:57.051022+23:59"], "kwargs": {}}, "my_slotted_class": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "MyDataclassWSlots"], "method": null, "args": [], "kwargs": {"foo": "bar", "bar": 2}}, "my_dataclass": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "MyDataclass"], "method": null, "args": [], "kwargs": {"foo": "foo", "bar": 1}}, "my_enum": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "MyEnum"], "method": null, "args": ["foo"], "kwargs": {}}, "my_pydantic": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "MyPydantic"], "method": null, "args": [], "kwargs": {"foo": "foo", "bar": 1}}, "my_funny_pydantic": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "MyFunnyPydantic"], "method": null, "args": [], "kwargs": {"foo": "foo", "bar": 1}}, "person": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "Person"], "method": null, "args": [], "kwargs": {"name": "foo"}}, "a_bool": true, "a_none": null, "a_str": "foo", "a_str_nuc": "foo\\u0000", "a_str_uc": "foo \xe2\x9b\xb0\xef\xb8\x8f", "a_str_ucuc": "foo \xe2\x9b\xb0\xef\xb8\x8f\\u0000", "a_str_ucucuc": "foo \\\\u26f0\\\\ufe0f", "text": ["Hello", "Python", "Surrogate", "Example", "String", "With", "Surrogates", "Embedded", "In", "The", "Text", "\xe6\x94\xb6\xe8\x8a\xb1\xf0\x9f\x99\x84\xc2\xb7\xe5\x88\xb0"], "an_int": 1, "a_float": 1.1, "runnable_map": {"lc": 1, "type": "constructor", "id": ["langchain", "schema", "runnable", "RunnableParallel"], "kwargs": {"steps__": {}}, "name": "RunnableParallel<>", "graph": {"nodes": [{"id": 0, "type": "schema", "data": "Parallel<>Input"}, {"id": 1, "type": "schema", "data": "Parallel<>Output"}], "edges": []}}}""",
b"""{"path": {"lc": 2, "type": "constructor", "id": ["pathlib", "Path"], "method": null, "args": ["foo", "bar"], "kwargs": {}}, "re": {"lc": 2, "type": "constructor", "id": ["re", "compile"], "method": null, "args": ["foo", 48], "kwargs": {}}, "decimal": {"lc": 2, "type": "constructor", "id": ["decimal", "Decimal"], "method": null, "args": ["1.10101"], "kwargs": {}}, "ip4": {"lc": 2, "type": "constructor", "id": ["ipaddress", "IPv4Address"], "method": null, "args": ["192.168.0.1"], "kwargs": {}}, "deque": {"lc": 2, "type": "constructor", "id": ["collections", "deque"], "method": null, "args": [[1, 2, 3]], "kwargs": {}}, "tzn": {"lc": 2, "type": "constructor", "id": ["zoneinfo", "ZoneInfo"], "method": null, "args": ["America/New_York"], "kwargs": {}}, "date": {"lc": 2, "type": "constructor", "id": ["datetime", "date"], "method": null, "args": [2024, 4, 19], "kwargs": {}}, "time": {"lc": 2, "type": "constructor", "id": ["datetime", "time"], "method": null, "args": [23, 4, 57, 51022, {"lc": 2, "type": "constructor", "id": ["datetime", "timezone"], "method": null, "args": [{"lc": 2, "type": "constructor", "id": ["datetime", "timedelta"], "method": null, "args": [0, 86340, 0], "kwargs": {}}], "kwargs": {}}], "kwargs": {"fold": 0}}, "uid": {"lc": 2, "type": "constructor", "id": ["uuid", "UUID"], "method": null, "args": ["00000000000000000000000000000001"], "kwargs": {}}, "timestamp": {"lc": 2, "type": "constructor", "id": ["datetime", "datetime"], "method": "fromisoformat", "args": ["2024-04-19T23:04:57.051022+23:59"], "kwargs": {}}, "my_slotted_class": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "MyDataclassWSlots"], "method": null, "args": [], "kwargs": {"foo": "bar", "bar": 2}}, "my_dataclass": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "MyDataclass"], "method": null, "args": [], "kwargs": {"foo": "foo", "bar": 1}}, "my_enum": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "MyEnum"], "method": null, "args": ["foo"], "kwargs": {}}, "my_pydantic": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "MyPydantic"], "method": "model_construct", "args": [], "kwargs": {"foo": "foo", "bar": 1}}, "my_funny_pydantic": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "MyFunnyPydantic"], "method": "construct", "args": [], "kwargs": {"foo": "foo", "bar": 1}}, "person": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "Person"], "method": null, "args": [], "kwargs": {"name": "foo"}}, "a_bool": true, "a_none": null, "a_str": "foo", "a_str_nuc": "foo\\u0000", "a_str_uc": "foo \xe2\x9b\xb0\xef\xb8\x8f", "a_str_ucuc": "foo \xe2\x9b\xb0\xef\xb8\x8f\\u0000", "a_str_ucucuc": "foo \\\\u26f0\\\\ufe0f", "text": ["Hello", "Python", "Surrogate", "Example", "String", "With", "Surrogates", "Embedded", "In", "The", "Text", "\xe6\x94\xb6\xe8\x8a\xb1\xf0\x9f\x99\x84\xc2\xb7\xe5\x88\xb0"], "an_int": 1, "a_float": 1.1, "runnable_map": {"lc": 1, "type": "constructor", "id": ["langchain", "schema", "runnable", "RunnableParallel"], "kwargs": {"steps__": {}}, "name": "RunnableParallel<>", "graph": {"nodes": [{"id": 0, "type": "schema", "data": "Parallel<>Input"}, {"id": 1, "type": "schema", "data": "Parallel<>Output"}], "edges": []}}}""",
)
assert serde.loads_typed(dumped) == {
libs/langgraph/README.md
+2 -2
View File
@@ -59,7 +59,7 @@ from langchain_core.messages import HumanMessage
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, StateGraph, MessagesState
from langgraph.graph import END, START, StateGraph, MessagesState
from langgraph.prebuilt import ToolNode
@@ -107,7 +107,7 @@ workflow.add_node("tools", tool_node)
# Set the entrypoint as `agent`
# This means that this node is the first one called
workflow.set_entry_point("agent")
workflow.add_edge(START, "agent")
# We now add a conditional edge
workflow.add_conditional_edges(
libs/langgraph/langgraph/channels/context.py
+3 -122
View File
@@ -1,124 +1,5 @@
from contextlib import asynccontextmanager, contextmanager
from inspect import signature
from typing import (
Any,
AsyncContextManager,
AsyncGenerator,
ContextManager,
Generator,
Generic,
Optional,
Sequence,
Type,
Union,
)
from langgraph.managed.context import Context as ContextManagedValue
from langchain_core.runnables import RunnableConfig
from typing_extensions import Self
Context = ContextManagedValue.of
from langgraph.channels.base import BaseChannel, Value
from langgraph.errors import EmptyChannelError, InvalidUpdateError
class Context(Generic[Value], BaseChannel[Value, None, None]):
"""Exposes the value of a context manager, for the duration of an invocation.
Context manager is entered before the first step, and exited after the last step.
Optionally, provide an equivalent async context manager, which will be used
instead for async invocations.
```python
import httpx
client = Channels.Context(httpx.Client, httpx.AsyncClient)
```
"""
value: Value
def __init__(
self,
ctx: Union[
None, Type[ContextManager[Value]], Type[AsyncContextManager[Value]]
] = None,
actx: Optional[Type[AsyncContextManager[Value]]] = None,
) -> None:
if ctx is None and actx is None:
raise ValueError("Must provide either sync or async context manager.")
self.ctx = ctx
self.actx = actx
def __eq__(self, value: object) -> bool:
return (
isinstance(value, Context)
and value.ctx == self.ctx
and value.actx == self.actx
)
@property
def ValueType(self) -> Any:
"""The type of the value stored in the channel."""
return None
@property
def UpdateType(self) -> Type[None]:
"""The type of the update received by the channel."""
return None
def checkpoint(self) -> None:
raise EmptyChannelError()
@contextmanager
def from_checkpoint(
self, checkpoint: None, config: RunnableConfig
) -> Generator[Self, None, None]:
if self.ctx is None:
raise ValueError("Cannot enter sync context manager.")
empty = self.__class__(ctx=self.ctx, actx=self.actx)
ctx = (
self.ctx(config)
if signature(self.ctx).parameters.get("config")
else self.ctx()
)
with ctx as value:
empty.value = value
yield empty
@asynccontextmanager
async def afrom_checkpoint(
self, checkpoint: None, config: RunnableConfig
) -> AsyncGenerator[Self, None]:
empty = self.__class__(ctx=self.ctx, actx=self.actx)
if self.actx is not None:
ctx = (
self.actx(config)
if signature(self.actx).parameters.get("config")
else self.actx()
)
else:
ctx = (
self.ctx(config)
if signature(self.ctx).parameters.get("config")
else self.ctx()
)
if hasattr(ctx, "__aenter__"):
async with ctx as value:
empty.value = value
yield empty
else:
with ctx as value:
empty.value = value
yield empty
def update(self, values: Sequence[None]) -> bool:
if values:
raise InvalidUpdateError(
f"At key '{self.key}': Context channel does not accept writes."
)
return False
def get(self) -> Value:
try:
return self.value
except AttributeError:
raise EmptyChannelError()
__all__ = ["Context"]
libs/langgraph/langgraph/constants.py
+2
View File
@@ -12,6 +12,7 @@ CONFIG_KEY_TASK_ID = "__pregel_task_id"
INTERRUPT = "__interrupt__"
ERROR = "__error__"
TASKS = "__pregel_tasks"
RUNTIME_PLACEHOLDER = "__pregel_runtime_placeholder__"
RESERVED = {
INTERRUPT,
ERROR,
@@ -23,6 +24,7 @@ RESERVED = {
CONFIG_KEY_RESUMING,
CONFIG_KEY_TASK_ID,
INPUT,
RUNTIME_PLACEHOLDER,
}
TAG_HIDDEN = "langsmith:hidden"
libs/langgraph/langgraph/graph/state.py
+20 -15
View File
@@ -6,6 +6,7 @@ from functools import partial
from inspect import isclass, isfunction, signature
from typing import (
Any,
Callable,
NamedTuple,
Optional,
Sequence,
@@ -25,7 +26,6 @@ from langchain_core.runnables.utils import (
from langgraph.channels.base import BaseChannel
from langgraph.channels.binop import BinaryOperatorAggregate
from langgraph.channels.context import Context
from langgraph.channels.dynamic_barrier_value import DynamicBarrierValue, WaitForNames
from langgraph.channels.ephemeral_value import EphemeralValue
from langgraph.channels.last_value import LastValue
@@ -382,7 +382,7 @@ class StateGraph(Graph):
raise ValueError(f"Need to add_node `{start}` first")
if end_key == START:
raise ValueError("START cannot be an end node")
if end_key not in self.nodes:
if end_key != END and end_key not in self.nodes:
raise ValueError(f"Need to add_node `{end_key}` first")
self.waiting_edges.add((tuple(start_key), end_key))
@@ -433,16 +433,14 @@ class StateGraph(Graph):
else [
key
for key, val in self.schemas[self.output].items()
if not isinstance(val, Context) and not is_managed_value(val)
if not is_managed_value(val)
]
)
stream_channels = (
"__root__"
if len(self.channels) == 1 and "__root__" in self.channels
else [
key
for key, val in self.channels.items()
if not isinstance(val, Context) and not is_managed_value(val)
key for key, val in self.channels.items() if not is_managed_value(val)
]
)
@@ -510,7 +508,6 @@ class CompiledStateGraph(CompiledGraph):
k: (self.channels[k].UpdateType, None)
for k in self.builder.schemas[self.builder.input]
if isinstance(self.channels[k], BaseChannel)
and not isinstance(self.channels[k], Context)
},
)
@@ -531,7 +528,7 @@ class CompiledStateGraph(CompiledGraph):
output_keys = [
k
for k, v in self.builder.schemas[self.builder.input].items()
if not isinstance(v, Context) and not is_managed_value(v)
if not is_managed_value(v)
]
else:
output_keys = list(self.builder.channels) + [
@@ -658,7 +655,14 @@ class CompiledStateGraph(CompiledGraph):
return ChannelWrite(writes, tags=[TAG_HIDDEN])
# attach branch publisher
self.nodes[start] |= branch.run(branch_writer, _get_state_reader(self.builder))
schema = (
self.builder.nodes[start].input
if start in self.builder.nodes
else self.builder.schema
)
self.nodes[start] |= branch.run(
branch_writer, _get_state_reader(self.builder, schema)
)
# attach branch subscribers
ends = (
@@ -684,16 +688,17 @@ class CompiledStateGraph(CompiledGraph):
)
def _get_state_reader(graph: StateGraph) -> ChannelRead:
state_keys = list(graph.channels)
def _get_state_reader(
builder: StateGraph, schema: Type[Any]
) -> Callable[[RunnableConfig], Any]:
state_keys = list(builder.channels)
select = list(builder.schemas[schema])
return partial(
ChannelRead.do_read,
channel=state_keys[0] if state_keys == ["__root__"] else state_keys,
select=select[0] if select == ["__root__"] else select,
fresh=True,
# coerce state dict to schema class (eg. pydantic model)
mapper=(
None if state_keys == ["__root__"] else partial(_coerce_state, graph.schema)
),
mapper=(None if state_keys == ["__root__"] else partial(_coerce_state, schema)),
)
libs/langgraph/langgraph/managed/base.py
+47 -2
View File
@@ -16,11 +16,16 @@ from typing import (
from langchain_core.runnables import RunnableConfig
from typing_extensions import Self, TypeGuard
from langgraph.constants import RUNTIME_PLACEHOLDER
V = TypeVar("V")
U = TypeVar("U")
class ManagedValue(ABC, Generic[V]):
runtime: bool = False
"""Whether the managed value is always created at runtime, ie. never stored."""
def __init__(self, config: RunnableConfig) -> None:
self.config = config
@@ -74,8 +79,6 @@ class ConfiguredManagedValue(NamedTuple):
ManagedValueSpec = Union[Type[ManagedValue], ConfiguredManagedValue]
ManagedValueMapping = dict[str, ManagedValue]
def is_managed_value(value: Any) -> TypeGuard[ManagedValueSpec]:
return (isclass(value) and issubclass(value, ManagedValue)) or isinstance(
@@ -103,3 +106,45 @@ def is_writable_managed_value(value: Any) -> TypeGuard[Type[WritableManagedValue
ChannelKeyPlaceholder = object()
ChannelTypePlaceholder = object()
class ManagedValueMapping(dict[str, ManagedValue]):
def replace_runtime_values(self, step: int, values: Union[dict[str, Any], Any]):
if not self or not values:
return
if all(not mv.runtime for mv in self.values()):
return
if isinstance(values, dict):
for key, value in values.items():
for chan, mv in self.items():
if mv.runtime and mv(step) is value:
values[key] = {RUNTIME_PLACEHOLDER: chan}
elif hasattr(values, "__dir__") and callable(values.__dir__):
for key in dir(values):
try:
value = getattr(values, key)
for chan, mv in self.items():
if mv.runtime and mv(step) is value:
setattr(values, key, {RUNTIME_PLACEHOLDER: chan})
except AttributeError:
pass
def replace_runtime_placeholders(
self, step: int, values: Union[dict[str, Any], Any]
):
if not self or not values:
return
if all(not mv.runtime for mv in self.values()):
return
if isinstance(values, dict):
for key, value in values.items():
if isinstance(value, dict) and RUNTIME_PLACEHOLDER in value:
values[key] = self[value[RUNTIME_PLACEHOLDER]](step)
elif hasattr(values, "__dir__") and callable(values.__dir__):
for key in dir(values):
try:
value = getattr(values, key)
if isinstance(value, dict) and RUNTIME_PLACEHOLDER in value:
setattr(values, key, self[value[RUNTIME_PLACEHOLDER]](step))
except AttributeError:
pass
libs/langgraph/langgraph/managed/context.py
+87
View File
@@ -0,0 +1,87 @@
from contextlib import asynccontextmanager, contextmanager
from inspect import signature
from typing import (
Any,
AsyncContextManager,
AsyncIterator,
ContextManager,
Iterator,
Optional,
Type,
Union,
)
from langchain_core.runnables import RunnableConfig
from typing_extensions import Self
from langgraph.managed.base import ConfiguredManagedValue, ManagedValue, V
class Context(ManagedValue):
runtime = True
value: V
@staticmethod
def of(
ctx: Union[None, Type[ContextManager[V]], Type[AsyncContextManager[V]]] = None,
actx: Optional[Type[AsyncContextManager[V]]] = None,
) -> ConfiguredManagedValue:
if ctx is None and actx is None:
raise ValueError("Must provide either sync or async context manager.")
return ConfiguredManagedValue(Context, {"ctx": ctx, "actx": actx})
@classmethod
@contextmanager
def enter(cls, config: RunnableConfig, **kwargs: Any) -> Iterator[Self]:
with super().enter(config, **kwargs) as self:
if self.ctx is None:
raise ValueError(
"Synchronous context manager not found. Please initialize Context value with a sync context manager, or invoke your graph asynchronously."
)
ctx = (
self.ctx(config)
if signature(self.ctx).parameters.get("config")
else self.ctx()
)
with ctx as v:
self.value = v
yield self
@classmethod
@asynccontextmanager
async def aenter(cls, config: RunnableConfig, **kwargs: Any) -> AsyncIterator[Self]:
async with super().aenter(config, **kwargs) as self:
if self.actx is not None:
ctx = (
self.actx(config)
if signature(self.actx).parameters.get("config")
else self.actx()
)
else:
ctx = (
self.ctx(config)
if signature(self.ctx).parameters.get("config")
else self.ctx()
)
if hasattr(ctx, "__aenter__"):
async with ctx as v:
self.value = v
yield self
else:
with ctx as v:
self.value = v
yield self
def __init__(
self,
config: RunnableConfig,
*,
ctx: Union[None, Type[ContextManager[V]], Type[AsyncContextManager[V]]] = None,
actx: Optional[Type[AsyncContextManager[V]]] = None,
) -> None:
self.ctx = ctx
self.actx = actx
def __call__(self, step: int) -> V:
return self.value
libs/langgraph/langgraph/managed/shared_value.py
-1
View File
@@ -81,7 +81,6 @@ class SharedValue(WritableManagedValue[Value, Update]):
):
raise ValueError("SharedValue must be a dict")
self.scope = scope
self.config = config
self.value: Value = {}
self.store: BaseStore = config["configurable"].get(CONFIG_KEY_STORE)
if self.store is None:
libs/langgraph/langgraph/prebuilt/chat_agent_executor.py
+8 -3
View File
@@ -130,7 +130,7 @@ def _get_model_preprocessing_runnable(
@deprecated_parameter("messages_modifier", "0.1.9", "state_modifier", removal="0.3.0")
def create_react_agent(
model: LanguageModelLike,
tools: Union[ToolExecutor, Sequence[BaseTool]],
tools: Union[ToolExecutor, Sequence[BaseTool], ToolNode],
*,
state_schema: Optional[StateSchemaType] = None,
messages_modifier: Optional[MessagesModifier] = None,
@@ -144,7 +144,7 @@ def create_react_agent(
Args:
model: The `LangChain` chat model that supports tool calling.
tools: A list of tools or a ToolExecutor instance.
tools: A list of tools, a ToolExecutor, or a ToolNode instance.
state_schema: An optional state schema that defines graph state.
Must have `messages` and `is_last_step` keys.
Defaults to `AgentState` that defines those two keys.
@@ -419,8 +419,13 @@ def create_react_agent(
if isinstance(tools, ToolExecutor):
tool_classes = tools.tools
tool_node = ToolNode(tool_classes)
elif isinstance(tools, ToolNode):
tool_classes = tools.tools_by_name.values()
tool_node = tools
else:
tool_classes = tools
tool_node = ToolNode(tool_classes)
model = model.bind_tools(tool_classes)
# Define the function that determines whether to continue or not
@@ -474,7 +479,7 @@ def create_react_agent(
# Define the two nodes we will cycle between
workflow.add_node("agent", RunnableLambda(call_model, acall_model))
workflow.add_node("tools", ToolNode(tool_classes))
workflow.add_node("tools", tool_node)
# Set the entrypoint as `agent`
# This means that this node is the first one called
libs/langgraph/langgraph/pregel/__init__.py
+63 -16
View File
@@ -38,6 +38,7 @@ from langchain_core.runnables.config import (
ensure_config,
get_async_callback_manager_for_config,
get_callback_manager_for_config,
merge_configs,
patch_config,
)
from langchain_core.runnables.utils import (
@@ -51,7 +52,6 @@ from typing_extensions import Self
from langgraph.channels.base import (
BaseChannel,
)
from langgraph.channels.context import Context
from langgraph.checkpoint.base import (
BaseCheckpointSaver,
CheckpointTuple,
@@ -71,7 +71,12 @@ from langgraph.constants import (
)
from langgraph.errors import GraphInterrupt, GraphRecursionError, InvalidUpdateError
from langgraph.managed.base import ManagedValueSpec
from langgraph.pregel.algo import apply_writes, local_read, prepare_next_tasks
from langgraph.pregel.algo import (
apply_writes,
local_read,
local_write,
prepare_next_tasks,
)
from langgraph.pregel.debug import (
print_step_checkpoint,
print_step_tasks,
@@ -396,11 +401,18 @@ class Pregel(
config_type: Optional[Type[Any]] = None
config: Optional[RunnableConfig] = None
name: str = "LangGraph"
class Config:
arbitrary_types_allowed = True
def with_config(self, config: RunnableConfig | None = None, **kwargs: Any) -> Self:
return self.copy(
update={"config": cast(RunnableConfig, {**(config or {}), **kwargs})}
)
@classmethod
def is_lc_serializable(cls) -> bool:
"""Return whether the graph can be serialized by Langchain."""
@@ -474,6 +486,7 @@ class Pregel(
def get_input_schema(
self, config: Optional[RunnableConfig] = None
) -> Type[BaseModel]:
config = merge_configs(self.config, config)
if isinstance(self.input_channels, str):
return super().get_input_schema(config)
else:
@@ -493,6 +506,7 @@ class Pregel(
def get_output_schema(
self, config: Optional[RunnableConfig] = None
) -> Type[BaseModel]:
config = merge_configs(self.config, config)
if isinstance(self.output_channels, str):
return super().get_output_schema(config)
else:
@@ -511,10 +525,7 @@ class Pregel(
@property
def stream_channels_asis(self) -> Union[str, Sequence[str]]:
return self.stream_channels or [
k
for k in self.channels
if isinstance(self.channels[k], BaseChannel)
and not isinstance(self.channels[k], Context)
k for k in self.channels if isinstance(self.channels[k], BaseChannel)
]
@property
@@ -534,6 +545,7 @@ class Pregel(
if not self.checkpointer:
raise ValueError("No checkpointer set")
config = merge_configs(self.config, config) if self.config else config
saved = self.checkpointer.get_tuple(config)
checkpoint_config = saved.config if saved else config
checkpoint_ns_to_graph: dict[str, Pregel] = _get_checkpoint_ns_to_graph(self)
@@ -553,6 +565,7 @@ class Pregel(
if not self.checkpointer:
raise ValueError("No checkpointer set")
config = merge_configs(self.config, config) if self.config else config
saved = await self.checkpointer.aget_tuple(config)
checkpoint_config = saved.config if saved else config
checkpoint_ns_to_graph: dict[str, Pregel] = _get_checkpoint_ns_to_graph(self)
@@ -594,7 +607,10 @@ class Pregel(
checkpoint_tuples = [
checkpoint_tuple
for checkpoint_tuple in self.checkpointer.list(
config, before=before, limit=limit, filter=filter
merge_configs(self.config, config) if self.config else config,
before=before,
limit=limit,
filter=filter,
)
]
for checkpoint_tuple in checkpoint_tuples:
@@ -632,7 +648,10 @@ class Pregel(
checkpoint_tuples = [
checkpoint_tuple
async for checkpoint_tuple in self.checkpointer.alist(
config, before=before, limit=limit, filter=filter
merge_configs(self.config, config) if self.config else config,
before=before,
limit=limit,
filter=filter,
)
]
@@ -667,6 +686,7 @@ class Pregel(
raise ValueError("No checkpointer set")
# get last checkpoint
config = merge_configs(self.config, config) if self.config else config
saved = self.checkpointer.get_tuple(config)
checkpoint = copy_checkpoint(saved.checkpoint) if saved else empty_checkpoint()
checkpoint_previous_versions = (
@@ -728,7 +748,7 @@ class Pregel(
# update channels
with ChannelsManager(self.channels, checkpoint, config) as (
channels,
_,
managed,
):
# create task to run all writers of the chosen node
writers = self.nodes[as_node].get_writers()
@@ -752,9 +772,22 @@ class Pregel(
run_name=self.name + "UpdateState",
configurable={
# deque.extend is thread-safe
CONFIG_KEY_SEND: task.writes.extend,
CONFIG_KEY_SEND: partial(
local_write,
step + 1,
task.writes.extend,
self.nodes,
channels,
managed,
),
CONFIG_KEY_READ: partial(
local_read, checkpoint, channels, task, config
local_read,
step + 1,
checkpoint,
channels,
managed,
task,
config,
),
},
),
@@ -787,6 +820,7 @@ class Pregel(
raise ValueError("No checkpointer set")
# get last checkpoint
config = merge_configs(self.config, config) if self.config else config
saved = await self.checkpointer.aget_tuple(config)
checkpoint = copy_checkpoint(saved.checkpoint) if saved else empty_checkpoint()
checkpoint_previous_versions = (
@@ -846,7 +880,7 @@ class Pregel(
# update channels, acting as the chosen node
async with AsyncChannelsManager(self.channels, checkpoint, config) as (
channels,
_,
managed,
):
# create task to run all writers of the chosen node
writers = self.nodes[as_node].get_writers()
@@ -870,9 +904,22 @@ class Pregel(
run_name=self.name + "UpdateState",
configurable={
# deque.extend is thread-safe
CONFIG_KEY_SEND: task.writes.extend,
CONFIG_KEY_SEND: partial(
local_write,
step + 1,
task.writes.extend,
self.nodes,
channels,
managed,
),
CONFIG_KEY_READ: partial(
local_read, checkpoint, channels, task, config
local_read,
step + 1,
checkpoint,
channels,
managed,
task,
config,
),
},
),
@@ -1022,7 +1069,7 @@ class Pregel(
{'type': 'task_result', 'timestamp': '2024-06-23T...+00:00', 'step': 2, 'payload': {'id': '...', 'name': 'b', 'result': [('alist', ['there'])]}}
```
"""
config = ensure_config(config)
config = ensure_config(merge_configs(self.config, config))
callback_manager = get_callback_manager_for_config(config)
run_manager = callback_manager.on_chain_start(
dumpd(self),
@@ -1263,7 +1310,7 @@ class Pregel(
{'type': 'task_result', 'timestamp': '2024-06-23T...+00:00', 'step': 2, 'payload': {'id': '...', 'name': 'b', 'result': [('alist', ['there'])]}}
```
"""
config = ensure_config(config)
config = ensure_config(merge_configs(self.config, config))
callback_manager = get_async_callback_manager_for_config(config)
run_manager = await callback_manager.on_chain_start(
dumpd(self),
libs/langgraph/langgraph/pregel/algo.py
+23 -6
View File
@@ -24,7 +24,6 @@ from langchain_core.runnables.config import (
)
from langgraph.channels.base import BaseChannel
from langgraph.channels.context import Context
from langgraph.checkpoint.base import (
BaseCheckpointSaver,
Checkpoint,
@@ -95,28 +94,37 @@ def should_interrupt(
def local_read(
step: int,
checkpoint: Checkpoint,
channels: Mapping[str, BaseChannel],
managed: ManagedValueMapping,
task: WritesProtocol,
config: RunnableConfig,
select: Union[list[str], str],
fresh: bool = False,
) -> Union[dict[str, Any], Any]:
if isinstance(select, str):
managed_keys = []
else:
managed_keys = [k for k in select if k in managed]
select = [k for k in select if k not in managed]
if fresh:
new_checkpoint = create_checkpoint(copy_checkpoint(checkpoint), channels, -1)
context_channels = {k: v for k, v in channels.items() if isinstance(v, Context)}
with ChannelsManager(channels, new_checkpoint, config, skip_context=True) as (
channels,
_,
):
all_channels = {**channels, **context_channels}
apply_writes(new_checkpoint, all_channels, [task], None)
return read_channels(all_channels, select)
apply_writes(new_checkpoint, channels, [task], None)
values = read_channels(channels, select)
else:
return read_channels(channels, select)
values = read_channels(channels, select)
if managed_keys:
values.update({k: managed[k](step) for k in managed_keys})
return values
def local_write(
step: int,
commit: Callable[[Sequence[tuple[str, Any]]], None],
processes: Mapping[str, PregelNode],
channels: Mapping[str, BaseChannel],
@@ -131,6 +139,8 @@ def local_write(
)
if value.node not in processes:
raise InvalidUpdateError(f"Invalid node name {value.node} in packet")
# replace any runtime values with placeholders
managed.replace_runtime_values(step, value.arg)
elif chan not in channels and chan not in managed:
logger.warning(f"Skipping write for channel '{chan}' which has no readers")
commit(writes)
@@ -293,6 +303,7 @@ def prepare_next_tasks(
if for_execution:
proc = processes[packet.node]
if node := proc.get_node():
managed.replace_runtime_placeholders(step, packet.arg)
writes = deque()
tasks.append(
PregelExecutableTask(
@@ -317,6 +328,7 @@ def prepare_next_tasks(
# deque.extend is thread-safe
CONFIG_KEY_SEND: partial(
local_write,
step,
writes.extend,
processes,
channels,
@@ -324,8 +336,10 @@ def prepare_next_tasks(
),
CONFIG_KEY_READ: partial(
local_read,
step,
checkpoint,
channels,
managed,
PregelTaskWrites(packet.node, writes, triggers),
config,
),
@@ -414,6 +428,7 @@ def prepare_next_tasks(
# deque.extend is thread-safe
CONFIG_KEY_SEND: partial(
local_write,
step,
writes.extend,
processes,
channels,
@@ -421,8 +436,10 @@ def prepare_next_tasks(
),
CONFIG_KEY_READ: partial(
local_read,
step,
checkpoint,
channels,
managed,
PregelTaskWrites(name, writes, triggers),
config,
),
libs/langgraph/langgraph/pregel/manager.py
+27 -17
View File
@@ -5,8 +5,6 @@ from typing import AsyncIterator, Iterator, Mapping, Optional, Union
from langchain_core.runnables import RunnableConfig, patch_config
from langgraph.channels.base import BaseChannel
from langgraph.channels.context import Context
from langgraph.channels.last_value import LastValue
from langgraph.checkpoint.base import Checkpoint
from langgraph.constants import CONFIG_KEY_STORE
from langgraph.managed.base import (
@@ -14,6 +12,7 @@ from langgraph.managed.base import (
ManagedValueMapping,
ManagedValueSpec,
)
from langgraph.managed.context import Context
from langgraph.store.base import BaseStore
@@ -31,10 +30,12 @@ def ChannelsManager(
channel_specs: Mapping[str, BaseChannel] = {}
managed_specs: Mapping[str, ManagedValueSpec] = {}
for k, v in specs.items():
if skip_context and isinstance(v, Context):
channel_specs[k] = LastValue(None)
elif isinstance(v, BaseChannel):
if isinstance(v, BaseChannel):
channel_specs[k] = v
elif (
skip_context and isinstance(v, ConfiguredManagedValue) and v.cls is Context
):
managed_specs[k] = Context.of(noop_context)
else:
managed_specs[k] = v
with ExitStack() as stack:
@@ -45,14 +46,16 @@ def ChannelsManager(
)
for k, v in channel_specs.items()
},
{
key: stack.enter_context(
value.cls.enter(config_for_managed, **value.kwargs)
if isinstance(value, ConfiguredManagedValue)
else value.enter(config_for_managed)
)
for key, value in managed_specs.items()
},
ManagedValueMapping(
{
key: stack.enter_context(
value.cls.enter(config_for_managed, **value.kwargs)
if isinstance(value, ConfiguredManagedValue)
else value.enter(config_for_managed)
)
for key, value in managed_specs.items()
}
),
)
@@ -70,10 +73,12 @@ async def AsyncChannelsManager(
channel_specs: Mapping[str, BaseChannel] = {}
managed_specs: Mapping[str, ManagedValueSpec] = {}
for k, v in specs.items():
if skip_context and isinstance(v, Context):
channel_specs[k] = LastValue(None)
elif isinstance(v, BaseChannel):
if isinstance(v, BaseChannel):
channel_specs[k] = v
elif (
skip_context and isinstance(v, ConfiguredManagedValue) and v.cls is Context
):
managed_specs[k] = Context.of(noop_context)
else:
managed_specs[k] = v
async with AsyncExitStack() as stack:
@@ -102,5 +107,10 @@ async def AsyncChannelsManager(
for k, v in channel_specs.items()
},
# managed: build mapping from spec to result
{tasks[task]: task.result() for task in done},
ManagedValueMapping({tasks[task]: task.result() for task in done}),
)
@contextmanager
def noop_context() -> Iterator[None]:
yield None
libs/langgraph/langgraph/pregel/read.py
+5 -5
View File
@@ -67,19 +67,19 @@ class ChannelRead(RunnableCallable):
def _read(self, _: Any, config: RunnableConfig) -> Any:
return self.do_read(
config, channel=self.channel, fresh=self.fresh, mapper=self.mapper
config, select=self.channel, fresh=self.fresh, mapper=self.mapper
)
async def _aread(self, _: Any, config: RunnableConfig) -> Any:
return self.do_read(
config, channel=self.channel, fresh=self.fresh, mapper=self.mapper
config, select=self.channel, fresh=self.fresh, mapper=self.mapper
)
@staticmethod
def do_read(
config: RunnableConfig,
*,
channel: Union[str, list[str]],
select: Union[str, list[str]],
fresh: bool = False,
mapper: Optional[Callable[[Any], Any]] = None,
) -> Any:
@@ -91,9 +91,9 @@ class ChannelRead(RunnableCallable):
"Make sure to call in the context of a Pregel process"
)
if mapper:
return mapper(read(channel, fresh))
return mapper(read(select, fresh))
else:
return read(channel, fresh)
return read(select, fresh)
DEFAULT_BOUND: RunnablePassthrough = RunnablePassthrough()
libs/langgraph/pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "langgraph"
version = "0.2.12"
version = "0.2.14"
description = "Building stateful, multi-actor applications with LLMs"
authors = []
license = "MIT"
libs/langgraph/tests/__snapshots__/test_pregel.ambr
+70
View File
@@ -2279,6 +2279,76 @@
'''
# ---
# name: test_conditional_state_graph_with_list_edge_inputs
'''
{
"nodes": [
{
"id": "__start__",
"type": "schema",
"data": "__start__"
},
{
"id": "A",
"type": "runnable",
"data": {
"id": [
"langgraph",
"utils",
"RunnableCallable"
],
"name": "A"
}
},
{
"id": "B",
"type": "runnable",
"data": {
"id": [
"langgraph",
"utils",
"RunnableCallable"
],
"name": "B"
}
},
{
"id": "__end__",
"type": "schema",
"data": "__end__"
}
],
"edges": [
{
"source": "A",
"target": "__end__"
},
{
"source": "B",
"target": "__end__"
},
{
"source": "__start__",
"target": "A"
},
{
"source": "__start__",
"target": "B"
}
]
}
'''
# ---
# name: test_conditional_state_graph_with_list_edge_inputs.1
'''
graph TD;
A --> __end__;
B --> __end__;
__start__ --> A;
__start__ --> B;
'''
# ---
# name: test_conditional_state_graph[postgres]
'{"title": "LangGraphInput", "type": "object", "properties": {"input": {"title": "Input", "type": "string"}, "agent_outcome": {"title": "Agent Outcome", "anyOf": [{"$ref": "#/definitions/AgentAction"}, {"$ref": "#/definitions/AgentFinish"}]}, "intermediate_steps": {"title": "Intermediate Steps", "type": "array", "items": {"type": "array", "minItems": 2, "maxItems": 2, "items": [{"$ref": "#/definitions/AgentAction"}, {"type": "string"}]}}}, "definitions": {"AgentAction": {"title": "AgentAction", "description": "Represents a request to execute an action by an agent.\\n\\nThe action consists of the name of the tool to execute and the input to pass\\nto the tool. The log is used to pass along extra information about the action.", "type": "object", "properties": {"tool": {"title": "Tool", "type": "string"}, "tool_input": {"title": "Tool Input", "anyOf": [{"type": "string"}, {"type": "object"}]}, "log": {"title": "Log", "type": "string"}, "type": {"title": "Type", "default": "AgentAction", "enum": ["AgentAction"], "type": "string"}}, "required": ["tool", "tool_input", "log"]}, "AgentFinish": {"title": "AgentFinish", "description": "Final return value of an ActionAgent.\\n\\nAgents return an AgentFinish when they have reached a stopping condition.", "type": "object", "properties": {"return_values": {"title": "Return Values", "type": "object"}, "log": {"title": "Log", "type": "string"}, "type": {"title": "Type", "default": "AgentFinish", "enum": ["AgentFinish"], "type": "string"}}, "required": ["return_values", "log"]}}}'
# ---
libs/langgraph/tests/test_channels.py
+1 -83
View File
@@ -1,14 +1,9 @@
import operator
from contextlib import asynccontextmanager, contextmanager
from typing import AsyncGenerator, Generator, Sequence, Union
from typing import Sequence, Union
import httpx
import pytest
from langchain_core.runnables import RunnableConfig
from pytest_mock import MockerFixture
from langgraph.channels.binop import BinaryOperatorAggregate
from langgraph.channels.context import Context
from langgraph.channels.last_value import LastValue
from langgraph.channels.topic import Topic
from langgraph.errors import EmptyChannelError, InvalidUpdateError
@@ -257,80 +252,3 @@ async def test_binop_async() -> None:
checkpoint, {}
) as channel:
assert channel.get() == 10
def test_ctx_manager(mocker: MockerFixture) -> None:
setup = mocker.Mock()
cleanup = mocker.Mock()
@contextmanager
def an_int() -> Generator[int, None, None]:
setup()
try:
yield 5
finally:
cleanup()
with Context(an_int, None).from_checkpoint(None, {}) as channel:
assert setup.call_count == 1
assert cleanup.call_count == 0
assert channel.ValueType is None
assert channel.UpdateType is None
assert channel.get() == 5
with pytest.raises(InvalidUpdateError):
channel.update([5]) # type: ignore
assert setup.call_count == 1
assert cleanup.call_count == 1
def test_ctx_manager_ctx(mocker: MockerFixture) -> None:
with Context(httpx.Client).from_checkpoint(None, {}) as channel:
assert channel.ValueType is None
assert channel.UpdateType is None
assert isinstance(channel.get(), httpx.Client)
with pytest.raises(InvalidUpdateError):
channel.update([5]) # type: ignore
with pytest.raises(EmptyChannelError):
channel.checkpoint()
async def test_ctx_manager_async(mocker: MockerFixture) -> None:
setup = mocker.Mock()
cleanup = mocker.Mock()
@contextmanager
def an_int_sync(config: RunnableConfig) -> Generator[int, None, None]:
try:
yield 5
finally:
pass
@asynccontextmanager
async def an_int() -> AsyncGenerator[int, None]:
setup()
try:
yield 5
finally:
cleanup()
async with Context(an_int_sync, an_int).afrom_checkpoint(None, {}) as channel:
assert setup.call_count == 1
assert cleanup.call_count == 0
assert channel.ValueType is None
assert channel.UpdateType is None
assert channel.get() == 5
with pytest.raises(InvalidUpdateError):
channel.update([5]) # type: ignore
assert setup.call_count == 1
assert cleanup.call_count == 1
libs/langgraph/tests/test_pregel.py
+33 -4
View File
@@ -3595,6 +3595,24 @@ def test_conditional_state_graph(
]
def test_conditional_state_graph_with_list_edge_inputs(snapshot: SnapshotAssertion):
class State(TypedDict):
foo: Annotated[list[str], operator.add]
graph_builder = StateGraph(State)
graph_builder.add_node("A", lambda x: {"foo": ["A"]})
graph_builder.add_node("B", lambda x: {"foo": ["B"]})
graph_builder.add_edge(START, "A")
graph_builder.add_edge(START, "B")
graph_builder.add_edge(["A", "B"], END)
app = graph_builder.compile()
assert app.invoke({"foo": []}) == {"foo": ["A", "B"]}
assert json.dumps(app.get_graph().to_json(), indent=2) == snapshot
assert app.get_graph().draw_mermaid(with_styles=False) == snapshot
def test_state_graph_w_config_inherited_state_keys(snapshot: SnapshotAssertion) -> None:
from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.language_models.fake import FakeStreamingListLLM
@@ -4047,7 +4065,7 @@ def test_prebuilt_tool_chat(snapshot: SnapshotAssertion) -> None:
["memory", "sqlite", "postgres", "postgres_pipe"],
)
def test_state_graph_packets(
request: pytest.FixtureRequest, checkpointer_name: str
request: pytest.FixtureRequest, checkpointer_name: str, mocker: MockerFixture
) -> None:
from langchain_core.language_models.fake_chat_models import (
FakeMessagesListChatModel,
@@ -4067,6 +4085,7 @@ def test_state_graph_packets(
class AgentState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
session: Annotated[httpx.Client, Context(httpx.Client)]
@tool()
def search_api(query: str) -> str:
@@ -4110,6 +4129,7 @@ def test_state_graph_packets(
)
def agent(data: AgentState) -> AgentState:
assert isinstance(data["session"], httpx.Client)
return {
"messages": model.invoke(data["messages"]),
"something_extra": "hi there",
@@ -4117,16 +4137,26 @@ def test_state_graph_packets(
# Define decision-making logic
def should_continue(data: AgentState) -> str:
assert isinstance(data["session"], httpx.Client)
assert (
data["something_extra"] == "hi there"
), "nodes can pass extra data to their cond edges, which isn't saved in state"
# Logic to decide whether to continue in the loop or exit
if tool_calls := data["messages"][-1].tool_calls:
return [Send("tools", tool_call) for tool_call in tool_calls]
return [
Send("tools", {"call": tool_call, "my_session": data["session"]})
for tool_call in tool_calls
]
else:
return END
def tools_node(tool_call: ToolCall, config: RunnableConfig) -> AgentState:
class ToolInput(TypedDict):
call: ToolCall
my_session: httpx.Client
def tools_node(input: ToolInput, config: RunnableConfig) -> AgentState:
assert isinstance(input["my_session"], httpx.Client)
tool_call = input["call"]
time.sleep(tool_call["args"].get("idx", 0) / 10)
output = tools_by_name[tool_call["name"]].invoke(tool_call["args"], config)
return {
@@ -7497,7 +7527,6 @@ def test_in_one_fan_out_state_graph_waiting_edge_custom_state_class_pydantic1(
return {"answer": ",".join(data.docs)}
def decider(data: State) -> str:
print("decider", data)
assert isinstance(data, State)
return "retriever_two"
libs/langgraph/tests/test_pregel_async.py
+18 -3
View File
@@ -21,6 +21,9 @@ from uuid import UUID
import httpx
import pytest
from langchain_core.messages import (
ToolCall,
)
from langchain_core.runnables import (
RunnableConfig,
RunnableLambda,
@@ -3882,6 +3885,12 @@ async def test_prebuilt_tool_chat() -> None:
]
# defined outside to allow deserializer to see it
class ToolInput(BaseModel, arbitrary_types_allowed=True):
call: ToolCall
my_session: httpx.AsyncClient
async def test_state_graph_packets() -> None:
from langchain_core.language_models.fake_chat_models import (
FakeMessagesListChatModel,
@@ -3890,13 +3899,13 @@ async def test_state_graph_packets() -> None:
AIMessage,
BaseMessage,
HumanMessage,
ToolCall,
ToolMessage,
)
from langchain_core.tools import tool
class AgentState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
session: Annotated[httpx.AsyncClient, Context(httpx.AsyncClient)]
@tool()
def search_api(query: str) -> str:
@@ -3941,13 +3950,19 @@ async def test_state_graph_packets() -> None:
# Define decision-making logic
def should_continue(data: AgentState) -> str:
assert isinstance(data["session"], httpx.AsyncClient)
# Logic to decide whether to continue in the loop or exit
if tool_calls := data["messages"][-1].tool_calls:
return [Send("tools", tool_call) for tool_call in tool_calls]
return [
Send("tools", ToolInput(call=tool_call, my_session=data["session"]))
for tool_call in tool_calls
]
else:
return END
async def tools_node(tool_call: ToolCall, config: RunnableConfig) -> AgentState:
async def tools_node(input: ToolInput, config: RunnableConfig) -> AgentState:
assert isinstance(input.my_session, httpx.AsyncClient)
tool_call = input.call
await asyncio.sleep(tool_call["args"].get("idx", 0) / 10)
output = await tools_by_name[tool_call["name"]].ainvoke(
tool_call["args"], config
libs/sdk-js/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@langchain/langgraph-sdk",
"version": "0.0.6",
"version": "0.0.7",
"description": "Client library for interacting with the LangGraph API",
"type": "module",
"packageManager": "yarn@1.22.19",
libs/sdk-js/src/client.mts
+4
View File
@@ -24,6 +24,7 @@ import {
interface ClientConfig {
apiUrl?: string;
apiKey?: string;
callerOptions?: AsyncCallerParams;
timeoutMs?: number;
defaultHeaders?: Record<string, string | null | undefined>;
@@ -48,6 +49,9 @@ class BaseClient {
this.timeoutMs = config?.timeoutMs || 12_000;
this.apiUrl = config?.apiUrl || "http://localhost:8123";
this.defaultHeaders = config?.defaultHeaders || {};
if (config?.apiKey != null) {
this.defaultHeaders["X-Api-Key"] = config.apiKey;
}
}
protected prepareFetchOptions(
libs/sdk-js/yarn.lock
+3 -3
View File
@@ -1219,9 +1219,9 @@ micromark@^2.11.3, micromark@~2.11.0, micromark@~2.11.3:
parse-entities "^2.0.0"
micromatch@^4.0.4:
version "4.0.7"
resolved "https://registry.yarnpkg.com/micromatch/-/micromatch-4.0.7.tgz#33e8190d9fe474a9895525f5618eee136d46c2e5"
integrity sha512-LPP/3KorzCwBxfeUuZmaR6bG2kdeHSbe0P2tY3FLRU4vYrjYz5hI4QZwV0njUx3jeuKe67YukQ1LSPZBKDqO/Q==
version "4.0.8"
resolved "https://registry.yarnpkg.com/micromatch/-/micromatch-4.0.8.tgz#d66fa18f3a47076789320b9b1af32bd86d9fa202"
integrity sha512-PXwfBhYu0hBCPw8Dn0E+WDYb7af3dSLVWKi3HGv84IdF4TyFoC0ysxFd0Goxw7nSv4T/PzEJQxsYsEiFCKo2BA==
dependencies:
braces "^3.0.3"
picomatch "^2.3.1"
libs/sdk-py/langgraph_sdk/client.py
+10
View File
@@ -29,6 +29,7 @@ from langgraph_sdk.schema import (
GraphSchema,
Metadata,
MultitaskStrategy,
OnCompletionBehavior,
OnConflictBehavior,
Run,
RunCreate,
@@ -985,6 +986,7 @@ class RunsClient:
feedback_keys: Optional[list[str]] = None,
on_disconnect: Optional[DisconnectMode] = None,
webhook: Optional[str] = None,
on_completion: Optional[OnCompletionBehavior] = None,
) -> AsyncIterator[StreamPart]:
...
@@ -1004,6 +1006,7 @@ class RunsClient:
on_disconnect: Optional[DisconnectMode] = None,
webhook: Optional[str] = None,
multitask_strategy: Optional[MultitaskStrategy] = None,
on_completion: Optional[OnCompletionBehavior] = None,
) -> AsyncIterator[StreamPart]:
"""Create a run and stream the results.
@@ -1070,6 +1073,7 @@ class RunsClient:
"checkpoint_id": checkpoint_id,
"multitask_strategy": multitask_strategy,
"on_disconnect": on_disconnect,
"on_completion": on_completion,
}
endpoint = (
f"/threads/{thread_id}/runs/stream"
@@ -1092,6 +1096,7 @@ class RunsClient:
interrupt_before: Optional[list[str]] = None,
interrupt_after: Optional[list[str]] = None,
webhook: Optional[str] = None,
on_completion: Optional[OnCompletionBehavior] = None,
) -> Run:
...
@@ -1125,6 +1130,7 @@ class RunsClient:
interrupt_after: Optional[list[str]] = None,
webhook: Optional[str] = None,
multitask_strategy: Optional[MultitaskStrategy] = None,
on_completion: Optional[OnCompletionBehavior] = None,
) -> Run:
"""Create a background run.
@@ -1221,6 +1227,7 @@ class RunsClient:
"webhook": webhook,
"checkpoint_id": checkpoint_id,
"multitask_strategy": multitask_strategy,
"on_completion": on_completion,
}
payload = {k: v for k, v in payload.items() if v is not None}
if thread_id:
@@ -1268,6 +1275,7 @@ class RunsClient:
interrupt_after: Optional[list[str]] = None,
webhook: Optional[str] = None,
on_disconnect: Optional[DisconnectMode] = None,
on_completion: Optional[OnCompletionBehavior] = None,
) -> Union[list[dict], dict[str, Any]]:
...
@@ -1285,6 +1293,7 @@ class RunsClient:
webhook: Optional[str] = None,
on_disconnect: Optional[DisconnectMode] = None,
multitask_strategy: Optional[MultitaskStrategy] = None,
on_completion: Optional[OnCompletionBehavior] = None,
) -> Union[list[dict], dict[str, Any]]:
"""Create a run, wait until it finishes and return the final state.
@@ -1364,6 +1373,7 @@ class RunsClient:
"checkpoint_id": checkpoint_id,
"multitask_strategy": multitask_strategy,
"on_disconnect": on_disconnect,
"on_completion": on_completion,
}
endpoint = (
f"/threads/{thread_id}/runs/wait" if thread_id is not None else "/runs/wait"
libs/sdk-py/langgraph_sdk/schema.py
+2
View File
@@ -15,6 +15,8 @@ MultitaskStrategy = Literal["reject", "interrupt", "rollback", "enqueue"]
OnConflictBehavior = Literal["raise", "do_nothing"]
OnCompletionBehavior = Literal["delete", "keep"]
All = Literal["*"]