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

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

Browse Source
This commit is contained in:
Vadym Barda
2024-08-23 14:27:40 -04:00
committed by GitHub
parent 1333d8b478 ed7b2c9e8a
commit 4c4d7054ab
43 changed files with 5110 additions and 2049 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/chatbot-simulation-evaluation/agent-simulation-evaluation.ipynb
+3 -2
View File
@@ -235,7 +235,7 @@
" # Call the chat bot\n",
" chat_bot_response = my_chat_bot(messages)\n",
" # Respond with an AI Message\n",
" return {\"messages\":[AIMessage(content=chat_bot_response[\"content\"])]}"
" return {\"messages\": [AIMessage(content=chat_bot_response[\"content\"])]}"
]
},
{
@@ -270,7 +270,7 @@
" # Call the simulated user\n",
" response = simulated_user.invoke({\"messages\": new_messages})\n",
" # This response is an AI message - we need to flip this to be a human message\n",
" return {\"messages\":[HumanMessage(content=response.content)]}"
" return {\"messages\": [HumanMessage(content=response.content)]}"
]
},
{
@@ -331,6 +331,7 @@
"class State(TypedDict):\n",
" messages: Annotated[list, add_messages]\n",
"\n",
"\n",
"graph_builder = StateGraph(State)\n",
"graph_builder.add_node(\"user\", simulated_user_node)\n",
"graph_builder.add_node(\"chat_bot\", chat_bot_node)\n",
examples/chatbots/information-gather-prompting.ipynb
+14 -7
View File
@@ -79,7 +79,7 @@
"\n",
"\n",
"def info_chain(state):\n",
" messages = get_messages_info(state['messages'])\n",
" messages = get_messages_info(state[\"messages\"])\n",
" response = llm_with_tool.invoke(messages)\n",
" return {\"messages\": [response]}"
]
@@ -126,7 +126,7 @@
"\n",
"\n",
"def prompt_gen_chain(state):\n",
" messages = get_prompt_messages(state['messages'])\n",
" messages = get_prompt_messages(state[\"messages\"])\n",
" response = llm.invoke(messages)\n",
" return {\"messages\": [response]}"
]
@@ -158,7 +158,7 @@
"\n",
"\n",
"def get_state(state) -> Literal[\"add_tool_message\", \"info\", \"__end__\"]:\n",
" messages = state['messages']\n",
" messages = state[\"messages\"]\n",
" if isinstance(messages[-1], AIMessage) and messages[-1].tool_calls:\n",
" return \"add_tool_message\"\n",
" elif not isinstance(messages[-1], HumanMessage):\n",
@@ -190,9 +190,11 @@
"from typing import Annotated\n",
"from typing_extensions import TypedDict\n",
"\n",
"\n",
"class State(TypedDict):\n",
" messages: Annotated[list, add_messages]\n",
"\n",
"\n",
"memory = MemorySaver()\n",
"workflow = StateGraph(State)\n",
"workflow.add_node(\"info\", info_chain)\n",
@@ -201,9 +203,14 @@
"\n",
"@workflow.add_node\n",
"def add_tool_message(state: State):\n",
" return {\"messages\": [ToolMessage(\n",
" content=\"Prompt generated!\", tool_call_id=state['messages'][-1].tool_calls[0][\"id\"]\n",
" )]}\n",
" return {\n",
" \"messages\": [\n",
" ToolMessage(\n",
" content=\"Prompt generated!\",\n",
" tool_call_id=state[\"messages\"][-1].tool_calls[0][\"id\"],\n",
" )\n",
" ]\n",
" }\n",
"\n",
"\n",
"workflow.add_conditional_edges(\"info\", get_state)\n",
@@ -364,7 +371,7 @@
" for output in graph.stream(\n",
" {\"messages\": [HumanMessage(content=user)]}, config=config, stream_mode=\"updates\"\n",
" ):\n",
" last_message = next(iter(output.values()))['messages'][-1]\n",
" last_message = next(iter(output.values()))[\"messages\"][-1]\n",
" last_message.pretty_print()\n",
"\n",
" if output and \"prompt\" in output:\n",
examples/create-react-agent-hitl.ipynb
+1 -1
View File
@@ -239,7 +239,7 @@
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.11.1"
"version": "3.12.2"
}
},
"nbformat": 4,
examples/customer-support/customer-support.ipynb
+15 -9
View File
@@ -225,7 +225,14 @@
"\n",
"Define the (`fetch_user_flight_information`) tool to let the agent see the current user's flight information. Then define tools to search for flights and manage the passenger's bookings stored in the SQL database.\n",
"\n",
"We use `ensure_config` to pass in the `passenger_id` in via configurable parameters. The LLM never has to provide these explicitly, they are provided for a given invocation of the graph so that each user cannot access other passengers' booking information."
"We the can [access the RunnableConfig](https://python.langchain.com/v0.2/docs/how_to/tool_configure/#inferring-by-parameter-type) for a given run to check the `passenger_id` of the user accessing this application. The LLM never has to provide these explicitly, they are provided for a given invocation of the graph so that each user cannot access other passengers' booking information.\n",
"\n",
"<div class=\"admonition warning\">\n",
" <p class=\"admonition-title\">Compatibility</p>\n",
" <p>\n",
" This tutorial expects `langchain-core>=0.2.16` to use the injected RunnableConfig. Prior to that, you'd use `ensure_config` to collect the config from context.\n",
" </p>\n",
"</div> \n"
]
},
{
@@ -240,18 +247,17 @@
"from typing import Optional\n",
"\n",
"import pytz\n",
"from langchain_core.runnables import ensure_config\n",
"from langchain_core.runnables import RunnableConfig\n",
"\n",
"\n",
"@tool\n",
"def fetch_user_flight_information() -> list[dict]:\n",
"def fetch_user_flight_information(config: RunnableConfig) -> list[dict]:\n",
" \"\"\"Fetch all tickets for the user along with corresponding flight information and seat assignments.\n",
"\n",
" Returns:\n",
" A list of dictionaries where each dictionary contains the ticket details,\n",
" associated flight details, and the seat assignments for each ticket belonging to the user.\n",
" \"\"\"\n",
" config = ensure_config() # Fetch from the context\n",
" configuration = config.get(\"configurable\", {})\n",
" passenger_id = configuration.get(\"passenger_id\", None)\n",
" if not passenger_id:\n",
@@ -328,9 +334,10 @@
"\n",
"\n",
"@tool\n",
"def update_ticket_to_new_flight(ticket_no: str, new_flight_id: int) -> str:\n",
"def update_ticket_to_new_flight(\n",
" ticket_no: str, new_flight_id: int, *, config: RunnableConfig\n",
") -> str:\n",
" \"\"\"Update the user's ticket to a new valid flight.\"\"\"\n",
" config = ensure_config()\n",
" configuration = config.get(\"configurable\", {})\n",
" passenger_id = configuration.get(\"passenger_id\", None)\n",
" if not passenger_id:\n",
@@ -396,9 +403,8 @@
"\n",
"\n",
"@tool\n",
"def cancel_ticket(ticket_no: str) -> str:\n",
"def cancel_ticket(ticket_no: str, *, config: RunnableConfig) -> str:\n",
" \"\"\"Cancel the user's ticket and remove it from the database.\"\"\"\n",
" config = ensure_config()\n",
" configuration = config.get(\"configurable\", {})\n",
" passenger_id = configuration.get(\"passenger_id\", None)\n",
" if not passenger_id:\n",
@@ -4407,7 +4413,7 @@
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.11.2"
"version": "3.12.2"
}
},
"nbformat": 4,
examples/human_in_the_loop/review-tool-calls.ipynb
+35 -30
View File
@@ -141,15 +141,18 @@
" print(\"----\")\n",
" return \"Sunny!\"\n",
"\n",
"model = ChatAnthropic(model_name=\"claude-3-5-sonnet-20240620\").bind_tools([weather_search])\n",
"\n",
"model = ChatAnthropic(model_name=\"claude-3-5-sonnet-20240620\").bind_tools(\n",
" [weather_search]\n",
")\n",
"\n",
"\n",
"class State(MessagesState):\n",
" \"\"\"Simple state.\"\"\"\n",
"\n",
"\n",
"def call_llm(state):\n",
" return {\n",
" \"messages\": [model.invoke(state['messages'])]\n",
" }\n",
" return {\"messages\": [model.invoke(state[\"messages\"])]}\n",
"\n",
"\n",
"def human_review_node(state):\n",
@@ -159,28 +162,30 @@
"def run_tool(state):\n",
" new_messages = []\n",
" tools = {\"weather_search\": weather_search}\n",
" tool_calls = state['messages'][-1].tool_calls\n",
" tool_calls = state[\"messages\"][-1].tool_calls\n",
" for tool_call in tool_calls:\n",
" tool = tools[tool_call['name']]\n",
" result = tool.invoke(tool_call['args'])\n",
" new_messages.append({\n",
" \"role\": \"tool\",\n",
" \"name\": tool_call['name'],\n",
" \"content\": result,\n",
" \"tool_call_id\": tool_call['id']\n",
" })\n",
" tool = tools[tool_call[\"name\"]]\n",
" result = tool.invoke(tool_call[\"args\"])\n",
" new_messages.append(\n",
" {\n",
" \"role\": \"tool\",\n",
" \"name\": tool_call[\"name\"],\n",
" \"content\": result,\n",
" \"tool_call_id\": tool_call[\"id\"],\n",
" }\n",
" )\n",
" return {\"messages\": new_messages}\n",
"\n",
"\n",
"def route_after_llm(state) -> Literal[END, \"human_review_node\"]:\n",
" if len(state['messages'][-1].tool_calls) == 0:\n",
" if len(state[\"messages\"][-1].tool_calls) == 0:\n",
" return END\n",
" else:\n",
" return \"human_review_node\"\n",
"\n",
"\n",
"def route_after_human(state) -> Literal[\"run_tool\", \"call_llm\"]:\n",
" if isinstance(state['messages'][-1], AIMessage):\n",
" if isinstance(state[\"messages\"][-1], AIMessage):\n",
" return \"run_tool\"\n",
" else:\n",
" return \"call_llm\"\n",
@@ -460,35 +465,35 @@
"print(\"Current State:\")\n",
"print(state.values)\n",
"print(\"\\nCurrent Tool Call ID:\")\n",
"current_content = state.values['messages'][-1].content\n",
"current_id = state.values['messages'][-1].id\n",
"tool_call_id = state.values['messages'][-1].tool_calls[0]['id']\n",
"current_content = state.values[\"messages\"][-1].content\n",
"current_id = state.values[\"messages\"][-1].id\n",
"tool_call_id = state.values[\"messages\"][-1].tool_calls[0][\"id\"]\n",
"print(tool_call_id)\n",
"\n",
"# We now need to construct a replacement tool call.\n",
"# We will change the argument to be `San Francisco, USA`\n",
"# Note that we could change any number of arguments or tool names - it just has to be a valid one\n",
"new_message = {\n",
" \"role\": \"assistant\", \n",
" \"role\": \"assistant\",\n",
" \"content\": current_content,\n",
" \"tool_calls\": [\n",
" {\n",
" \"id\": tool_call_id,\n",
" \"name\": \"weather_search\",\n",
" \"args\": {\"city\": \"San Francisco, USA\"}\n",
" \"args\": {\"city\": \"San Francisco, USA\"},\n",
" }\n",
" ],\n",
" # This is important - this needs to be the same as the message you replacing!\n",
" # Otherwise, it will show up as a separate message\n",
" \"id\": current_id\n",
" \"id\": current_id,\n",
"}\n",
"graph.update_state(\n",
" # This is the config which represents this thread\n",
" thread, \n",
" thread,\n",
" # This is the updated value we want to push\n",
" {\"messages\": [new_message]}, \n",
" {\"messages\": [new_message]},\n",
" # We push this update acting as our human_review_node\n",
" as_node=\"human_review_node\"\n",
" as_node=\"human_review_node\",\n",
")\n",
"\n",
"# Let's now continue executing from here\n",
@@ -595,26 +600,26 @@
"print(\"Current State:\")\n",
"print(state.values)\n",
"print(\"\\nCurrent Tool Call ID:\")\n",
"tool_call_id = state.values['messages'][-1].tool_calls[0]['id']\n",
"tool_call_id = state.values[\"messages\"][-1].tool_calls[0][\"id\"]\n",
"print(tool_call_id)\n",
"\n",
"# We now need to construct a replacement tool call.\n",
"# We will change the argument to be `San Francisco, USA`\n",
"# Note that we could change any number of arguments or tool names - it just has to be a valid one\n",
"new_message = {\n",
" \"role\": \"tool\", \n",
" \"role\": \"tool\",\n",
" # This is our natural language feedback\n",
" \"content\": \"User requested changes: pass in the country as well\",\n",
" \"name\": \"weather_search\",\n",
" \"tool_call_id\": tool_call_id\n",
" \"tool_call_id\": tool_call_id,\n",
"}\n",
"graph.update_state(\n",
" # This is the config which represents this thread\n",
" thread, \n",
" thread,\n",
" # This is the updated value we want to push\n",
" {\"messages\": [new_message]}, \n",
" {\"messages\": [new_message]},\n",
" # We push this update acting as our human_review_node\n",
" as_node=\"human_review_node\"\n",
" as_node=\"human_review_node\",\n",
")\n",
"\n",
"# Let's now continue executing from here\n",
examples/input_output_schema.ipynb
+4
View File
@@ -33,15 +33,19 @@
"from langgraph.graph import StateGraph, START, END\n",
"from typing import TypedDict\n",
"\n",
"\n",
"class InputState(TypedDict):\n",
" question: str\n",
"\n",
"\n",
"class OutputState(TypedDict):\n",
" answer: str\n",
"\n",
"\n",
"def answer_node(state: InputState):\n",
" return {\"answer\": \"bye\"}\n",
"\n",
"\n",
"graph = StateGraph(input=InputState, output=OutputState)\n",
"graph.add_node(answer_node)\n",
"graph.add_edge(START, \"answer_node\")\n",
examples/llm-compiler/LLMCompiler.ipynb
+7 -3
View File
@@ -526,7 +526,7 @@
" \"tasks\": tasks,\n",
" }\n",
" )\n",
" return {\"messages\":[scheduled_tasks]}"
" return {\"messages\": [scheduled_tasks]}"
]
},
{
@@ -653,7 +653,7 @@
" )\n",
" ]\n",
" else:\n",
" return {\"messages\":response + [AIMessage(content=decision.action.response)]}\n",
" return {\"messages\": response + [AIMessage(content=decision.action.response)]}\n",
"\n",
"\n",
"def select_recent_messages(state) -> dict:\n",
@@ -726,9 +726,11 @@
"from langgraph.graph.message import add_messages\n",
"from typing import Annotated\n",
"\n",
"\n",
"class State(TypedDict):\n",
" messages: Annotated[list, add_messages]\n",
"\n",
"\n",
"graph_builder = StateGraph(State)\n",
"\n",
"# 1. Define vertices\n",
@@ -794,7 +796,9 @@
}
],
"source": [
"for step in chain.stream({\"messages\":[HumanMessage(content=\"What's the GDP of New York?\")]}):\n",
"for step in chain.stream(\n",
" {\"messages\": [HumanMessage(content=\"What's the GDP of New York?\")]}\n",
"):\n",
" print(step)\n",
" print(\"---\")"
]
examples/many-tools.ipynb
+3 -3
View File
@@ -328,9 +328,9 @@
" \"set more_information_needed False and populate a blank string for the query.\"\n",
" )\n",
" input_messages = [system] + state[\"messages\"]\n",
" response = llm.bind_tools(\n",
" [QueryForTools], tool_choice=True\n",
" ).invoke(input_messages)\n",
" response = llm.bind_tools([QueryForTools], tool_choice=True).invoke(\n",
" input_messages\n",
" )\n",
" query = response.tool_calls[0][\"args\"][\"query\"]\n",
" tool_documents = vector_store.similarity_search(query)\n",
" if hack_remove_tool_condition:\n",
examples/pass-run-time-values-to-tools.ipynb
+1
View File
@@ -329,6 +329,7 @@
"\n",
"tools = [get_context, cite_context_sources]\n",
"\n",
"\n",
"# Define the function that calls the model\n",
"def call_model(state, config):\n",
" messages = state[\"messages\"]\n",
examples/pass_private_state.ipynb
+2 -2
View File
@@ -72,12 +72,12 @@
"# Node to retrieve documents\n",
"def retrieve_documents(state: QueryOutputState) -> DocumentOutputState:\n",
" # Replace this with real logic\n",
" return {\"docs\": [state['query']] * 2}\n",
" return {\"docs\": [state[\"query\"]] * 2}\n",
"\n",
"\n",
"# Node to generate answer\n",
"def generate(state: GenerateInputState) -> OverallState:\n",
" return {\"answer\": \"\\n\\n\".join(state['docs'] + [state['question']])}\n",
" return {\"answer\": \"\\n\\n\".join(state[\"docs\"] + [state[\"question\"]])}\n",
"\n",
"\n",
"graph = StateGraph(OverallState)\n",
examples/persistence.ipynb
+1 -1
View File
@@ -587,7 +587,7 @@
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.11.9"
"version": "3.12.2"
}
},
"nbformat": 4,
examples/persistence_mongodb.ipynb
+10 -4
View File
@@ -630,7 +630,7 @@
" upsert=True,\n",
" )\n",
" )\n",
" await self.db[\"checkpoint_writes\"].bulk_write(operations)\n"
" await self.db[\"checkpoint_writes\"].bulk_write(operations)"
]
},
{
@@ -685,7 +685,9 @@
"metadata": {},
"outputs": [],
"source": [
"with MongoDBSaver.from_conn_info(host=\"localhost\", port=27017, db_name=\"checkpoints\") as checkpointer:\n",
"with MongoDBSaver.from_conn_info(\n",
" host=\"localhost\", port=27017, db_name=\"checkpoints\"\n",
") as checkpointer:\n",
" graph = create_react_agent(model, tools=tools, checkpointer=checkpointer)\n",
" config = {\"configurable\": {\"thread_id\": \"1\"}}\n",
" res = graph.invoke({\"messages\": [(\"human\", \"what's the weather in sf\")]}, config)\n",
@@ -796,10 +798,14 @@
"metadata": {},
"outputs": [],
"source": [
"async with AsyncMongoDBSaver.from_conn_info(host=\"localhost\", port=27017, db_name=\"checkpoints\") as checkpointer:\n",
"async with AsyncMongoDBSaver.from_conn_info(\n",
" host=\"localhost\", port=27017, db_name=\"checkpoints\"\n",
") as checkpointer:\n",
" graph = create_react_agent(model, tools=tools, checkpointer=checkpointer)\n",
" config = {\"configurable\": {\"thread_id\": \"2\"}}\n",
" res = await graph.ainvoke({\"messages\": [(\"human\", \"what's the weather in nyc\")]}, config)\n",
" res = await graph.ainvoke(\n",
" {\"messages\": [(\"human\", \"what's the weather in nyc\")]}, config\n",
" )\n",
"\n",
" latest_checkpoint = await checkpointer.aget(config)\n",
" latest_checkpoint_tuple = await checkpointer.aget_tuple(config)\n",
examples/persistence_postgres.ipynb
+6 -7
View File
@@ -122,7 +122,7 @@
"metadata": {},
"outputs": [],
"source": [
"DB_URI = \"postgresql://postgres:postgres@localhost:5441/postgres?sslmode=disable\""
"DB_URI = \"postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable\""
]
},
{
@@ -134,10 +134,9 @@
"source": [
"from psycopg.rows import dict_row\n",
"\n",
"connection_kwargs ={\n",
"connection_kwargs = {\n",
" \"autocommit\": True,\n",
" \"prepare_threshold\": 0,\n",
" \"row_factory\": dict_row,\n",
"}"
]
},
@@ -166,7 +165,7 @@
" # Example configuration\n",
" conninfo=DB_URI,\n",
" max_size=20,\n",
" kwargs=connection_kwargs\n",
" kwargs=connection_kwargs,\n",
")\n",
"\n",
"with pool.connection() as conn:\n",
@@ -394,7 +393,7 @@
" # Example configuration\n",
" conninfo=DB_URI,\n",
" max_size=20,\n",
" kwargs=connection_kwargs\n",
" kwargs=connection_kwargs,\n",
") as pool, pool.connection() as conn:\n",
" checkpointer = AsyncPostgresSaver(conn)\n",
"\n",
@@ -551,9 +550,9 @@
],
"metadata": {
"kernelspec": {
"display_name": "langgraph-postgres",
"display_name": "langgraph",
"language": "python",
"name": "langgraph-postgres"
"name": "langgraph"
},
"language_info": {
"codemirror_mode": {
examples/persistence_redis.ipynb
+9 -3
View File
@@ -530,7 +530,9 @@
"\n",
" @classmethod\n",
" @asynccontextmanager\n",
" async def from_conn_info(cls, *, host: str, port: int, db: int) -> AsyncIterator[\"AsyncRedisSaver\"]:\n",
" async def from_conn_info(\n",
" cls, *, host: str, port: int, db: int\n",
" ) -> AsyncIterator[\"AsyncRedisSaver\"]:\n",
" conn = None\n",
" try:\n",
" conn = AsyncRedis(host=host, port=port, db=db)\n",
@@ -887,10 +889,14 @@
"metadata": {},
"outputs": [],
"source": [
"async with AsyncRedisSaver.from_conn_info(host=\"localhost\", port=6379, db=0) as checkpointer:\n",
"async with AsyncRedisSaver.from_conn_info(\n",
" host=\"localhost\", port=6379, db=0\n",
") as checkpointer:\n",
" graph = create_react_agent(model, tools=tools, checkpointer=checkpointer)\n",
" config = {\"configurable\": {\"thread_id\": \"2\"}}\n",
" res = await graph.ainvoke({\"messages\": [(\"human\", \"what's the weather in nyc\")]}, config)\n",
" res = await graph.ainvoke(\n",
" {\"messages\": [(\"human\", \"what's the weather in nyc\")]}, config\n",
" )\n",
"\n",
" latest_checkpoint = await checkpointer.aget(config)\n",
" latest_checkpoint_tuple = await checkpointer.aget_tuple(config)\n",
examples/reflection/reflection.ipynb
+1 -1
View File
@@ -269,7 +269,7 @@
"class State(TypedDict):\n",
" messages: Annotated[list, add_messages]\n",
"\n",
" \n",
"\n",
"async def generation_node(state: Sequence[BaseMessage]):\n",
" return await generate.ainvoke({\"messages\": state})\n",
"\n",
examples/reflexion/reflexion.ipynb
+1
View File
@@ -392,6 +392,7 @@
"class State(TypedDict):\n",
" messages: Annotated[list, add_messages]\n",
"\n",
"\n",
"MAX_ITERATIONS = 5\n",
"builder = StateGraph(State)\n",
"builder.add_node(\"draft\", first_responder.respond)\n",
examples/streaming-content.ipynb
+3 -1
View File
@@ -68,7 +68,9 @@
" # It's completely optional, but useful if you have many functions with similar names\n",
" gen = RunnableGenerator(my_generator).with_config(\n",
" tags=[\"should_stream\"],\n",
" callbacks=config.get(\"callbacks\", []) # <-- Propagate callbacks (Python <= 3.10)\n",
" callbacks=config.get(\n",
" \"callbacks\", []\n",
" ), # <-- Propagate callbacks (Python <= 3.10)\n",
" )\n",
" async for message in gen.astream(state):\n",
" messages.append(message)\n",
examples/tutorials/rag-agent-testing-local.ipynb
+10 -9
View File
@@ -169,9 +169,7 @@
"from langchain_core.output_parsers import JsonOutputParser\n",
"\n",
"# JSON\n",
"llm = ChatOllama(model=\"llama3.1\", \n",
" format=\"json\", \n",
" temperature=0)\n",
"llm = ChatOllama(model=\"llama3.1\", format=\"json\", temperature=0)\n",
"\n",
"\n",
"prompt = PromptTemplate(\n",
@@ -210,6 +208,7 @@
"from IPython.display import Image, display\n",
"from langgraph.graph import START, END, StateGraph\n",
"\n",
"\n",
"class GraphState(TypedDict):\n",
" \"\"\"\n",
" Represents the state of our graph.\n",
@@ -381,21 +380,22 @@
"metadata": {},
"outputs": [],
"source": [
"import uuid \n",
"import uuid\n",
"\n",
"\n",
"def predict_custom_agent_answer(example: dict):\n",
" \n",
" config = {\"configurable\": {\"thread_id\": str(uuid.uuid4())}}\n",
" \n",
"\n",
" state_dict = custom_graph.invoke(\n",
" {\"question\": example[\"input\"], \"steps\": []}, config\n",
" )\n",
" \n",
"\n",
" return {\"response\": state_dict[\"generation\"], \"steps\": state_dict[\"steps\"]}\n",
"\n",
"\n",
"example = {\"input\": \"What are the types of agent memory?\"}\n",
"#response = predict_custom_agent_answer(example)\n",
"#response"
"# response = predict_custom_agent_answer(example)\n",
"# response"
]
},
{
@@ -544,6 +544,7 @@
" \"generate_answer\",\n",
"]\n",
"\n",
"\n",
"def check_trajectory_custom(root_run: Run, example: Example) -> dict:\n",
" \"\"\"\n",
" Check if all expected tools are called in exact order and without any additional tool calls.\n",
examples/tutorials/tool-calling-agent-local.ipynb
+4
View File
@@ -134,6 +134,7 @@
" for d in web_results\n",
" ]\n",
"\n",
"\n",
"# Tool list\n",
"tools = [retrieve_documents, web_search]"
]
@@ -152,9 +153,11 @@
"from langgraph.graph.message import AnyMessage, add_messages\n",
"from typing_extensions import TypedDict\n",
"\n",
"\n",
"class State(TypedDict):\n",
" messages: Annotated[list[AnyMessage], add_messages]\n",
"\n",
"\n",
"class Assistant:\n",
" def __init__(self, runnable: Runnable):\n",
" \"\"\"\n",
@@ -291,6 +294,7 @@
"source": [
"import uuid\n",
"\n",
"\n",
"def predict_react_agent_answer(example: dict):\n",
" \"\"\"Use this for answer evaluation\"\"\"\n",
"\n",
libs/checkpoint-postgres/README.md
-2
View File
@@ -44,7 +44,6 @@ with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
}
},
"pending_sends": [],
"current_tasks": {}
}
# store checkpoint
@@ -87,7 +86,6 @@ async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
}
},
"pending_sends": [],
"current_tasks": {}
}
# store checkpoint
libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py
+33 -29
View File
@@ -66,7 +66,7 @@ class PostgresSaver(BasePostgresSaver):
the first time checkpointer is used.
"""
with self.lock:
with self.conn.cursor(binary=True) as cur:
with self.conn.cursor(binary=True, row_factory=dict_row) as cur:
try:
version = cur.execute(
"SELECT v FROM checkpoint_migrations ORDER BY v DESC LIMIT 1"
@@ -127,31 +127,33 @@ class PostgresSaver(BasePostgresSaver):
if limit:
query += f" LIMIT {limit}"
# if we change this to use .stream() we need to make sure to close the cursor
for value in self.conn.execute(query, args, binary=True):
yield CheckpointTuple(
{
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": value["checkpoint_id"],
with self._cursor() as cur:
cur.execute(query, args, binary=True)
for value in cur:
yield CheckpointTuple(
{
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": value["checkpoint_id"],
}
},
{
**self._load_checkpoint(value["checkpoint"]),
"channel_values": self._load_blobs(value["channel_values"]),
},
self._load_metadata(value["metadata"]),
{
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": value["parent_checkpoint_id"],
}
}
},
{
**self._load_checkpoint(value["checkpoint"]),
"channel_values": self._load_blobs(value["channel_values"]),
},
self._load_metadata(value["metadata"]),
{
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": value["parent_checkpoint_id"],
}
}
if value["parent_checkpoint_id"]
else None,
self._load_writes(value["pending_writes"]),
)
if value["parent_checkpoint_id"]
else None,
self._load_writes(value["pending_writes"]),
)
def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
"""Get a checkpoint tuple from the database.
@@ -199,7 +201,7 @@ class PostgresSaver(BasePostgresSaver):
where = "WHERE thread_id = %s AND checkpoint_ns = %s ORDER BY checkpoint_id DESC LIMIT 1"
with self._cursor() as cur:
cur = self.conn.execute(
cur.execute(
self.SELECT_SQL + where,
args,
binary=True,
@@ -336,7 +338,7 @@ class PostgresSaver(BasePostgresSaver):
# in multiple threads/coroutines, but only one cursor can be
# used at a time
try:
with self.conn.cursor(binary=True) as cur:
with self.conn.cursor(binary=True, row_factory=dict_row) as cur:
yield cur
finally:
if pipeline:
@@ -344,8 +346,10 @@ class PostgresSaver(BasePostgresSaver):
elif pipeline:
# a connection not in pipeline mode can only be used by one
# thread/coroutine at a time, so we acquire a lock
with self.lock, self.conn.pipeline(), self.conn.cursor(binary=True) as cur:
with self.lock, self.conn.pipeline(), self.conn.cursor(
binary=True, row_factory=dict_row
) as cur:
yield cur
else:
with self.lock, self.conn.cursor(binary=True) as cur:
with self.lock, self.conn.cursor(binary=True, row_factory=dict_row) as cur:
yield cur
libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py
+35 -31
View File
@@ -64,7 +64,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
the first time checkpointer is used.
"""
async with self.lock:
async with self.conn.cursor(binary=True) as cur:
async with self.conn.cursor(binary=True, row_factory=dict_row) as cur:
try:
results = await cur.execute(
"SELECT v FROM checkpoint_migrations ORDER BY v DESC LIMIT 1"
@@ -110,33 +110,35 @@ class AsyncPostgresSaver(BasePostgresSaver):
if limit:
query += f" LIMIT {limit}"
# if we change this to use .stream() we need to make sure to close the cursor
async for value in await self.conn.execute(query, args, binary=True):
yield CheckpointTuple(
{
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": value["checkpoint_id"],
async with self._cursor() as cur:
await cur.execute(query, args, binary=True)
async for value in cur:
yield CheckpointTuple(
{
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": value["checkpoint_id"],
}
},
{
**self._load_checkpoint(value["checkpoint"]),
"channel_values": await asyncio.to_thread(
self._load_blobs, value["channel_values"]
),
},
self._load_metadata(value["metadata"]),
{
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": value["parent_checkpoint_id"],
}
}
},
{
**self._load_checkpoint(value["checkpoint"]),
"channel_values": await asyncio.to_thread(
self._load_blobs, value["channel_values"]
),
},
self._load_metadata(value["metadata"]),
{
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": value["parent_checkpoint_id"],
}
}
if value["parent_checkpoint_id"]
else None,
await asyncio.to_thread(self._load_writes, value["pending_writes"]),
)
if value["parent_checkpoint_id"]
else None,
await asyncio.to_thread(self._load_writes, value["pending_writes"]),
)
async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
"""Get a checkpoint tuple from the database asynchronously.
@@ -163,7 +165,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
where = "WHERE thread_id = %s AND checkpoint_ns = %s ORDER BY checkpoint_id DESC LIMIT 1"
async with self._cursor() as cur:
cur = await self.conn.execute(
await cur.execute(
self.SELECT_SQL + where,
args,
binary=True,
@@ -293,7 +295,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
# in multiple threads/coroutines, but only one cursor can be
# used at a time
try:
async with self.conn.cursor(binary=True) as cur:
async with self.conn.cursor(binary=True, row_factory=dict_row) as cur:
yield cur
finally:
if pipeline:
@@ -302,9 +304,11 @@ class AsyncPostgresSaver(BasePostgresSaver):
# a connection not in pipeline mode can only be used by one
# thread/coroutine at a time, so we acquire a lock
async with self.lock, self.conn.pipeline(), self.conn.cursor(
binary=True
binary=True, row_factory=dict_row
) as cur:
yield cur
else:
async with self.lock, self.conn.cursor(binary=True) as cur:
async with self.lock, self.conn.cursor(
binary=True, row_factory=dict_row
) as cur:
yield cur
libs/checkpoint-sqlite/README.md
+1 -3
View File
@@ -35,7 +35,6 @@ with SqliteSaver.from_conn_string(":memory:") as checkpointer:
}
},
"pending_sends": [],
"current_tasks": {}
}
# store checkpoint
@@ -78,7 +77,6 @@ async with AsyncSqliteSaver.from_conn_string(":memory:") as checkpointer:
}
},
"pending_sends": [],
"current_tasks": {}
}
# store checkpoint
@@ -89,4 +87,4 @@ async with AsyncSqliteSaver.from_conn_string(":memory:") as checkpointer:
# list checkpoints
[c async for c in checkpointer.alist(read_config)]
```
```
libs/checkpoint/README.md
-1
View File
@@ -74,7 +74,6 @@ checkpoint = {
}
},
"pending_sends": [],
"current_tasks": {}
}
# store checkpoint
libs/checkpoint/langgraph/checkpoint/base/__init__.py
+1 -6
View File
@@ -97,9 +97,6 @@ class Checkpoint(TypedDict):
pending_sends: List[SendProtocol]
"""List of packets sent to nodes but not yet processed.
Cleared by the next checkpoint."""
current_tasks: Dict[str, TaskInfo]
"""Map from task ID to task info."""
# TODO remove this
def empty_checkpoint() -> Checkpoint:
@@ -111,7 +108,6 @@ def empty_checkpoint() -> Checkpoint:
channel_versions={},
versions_seen={},
pending_sends=[],
current_tasks={},
)
@@ -124,7 +120,6 @@ def copy_checkpoint(checkpoint: Checkpoint) -> Checkpoint:
channel_versions=checkpoint["channel_versions"].copy(),
versions_seen={k: v.copy() for k, v in checkpoint["versions_seen"].items()},
pending_sends=checkpoint.get("pending_sends", []).copy(),
current_tasks=checkpoint.get("current_tasks", {}).copy(),
)
@@ -156,7 +151,6 @@ def create_checkpoint(
channel_versions=checkpoint["channel_versions"],
versions_seen=checkpoint["versions_seen"],
pending_sends=checkpoint.get("pending_sends", []),
current_tasks={},
)
@@ -451,3 +445,4 @@ saving regular writes.
Each Checkpointer implementation should use this mapping in put_writes.
"""
WRITES_IDX_MAP = {ERROR: -1}
# TODO To store scheduled status of tasks, add a special channel here
libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py
+4 -1
View File
@@ -7,6 +7,7 @@ import re
from collections import deque
from datetime import date, datetime, time, timedelta, timezone
from enum import Enum
from inspect import isclass
from ipaddress import (
IPv4Address,
IPv4Interface,
@@ -111,7 +112,7 @@ class JsonPlusSerializer(SerializerProtocol):
obj.__class__, method="fromhex", args=[obj.hex()]
)
elif isinstance(obj, BaseException):
return self._encode_constructor_args(obj.__class__, args=obj.args)
return repr(obj)
else:
raise TypeError(
f"Object of type {obj.__class__.__name__} is not JSON serializable"
@@ -135,6 +136,8 @@ class JsonPlusSerializer(SerializerProtocol):
method = getattr(cls, value["method"])
else:
method = cls
if isclass(method) and issubclass(method, BaseException):
return None
if value["args"] and value["kwargs"]:
return method(*value["args"], **value["kwargs"])
elif value["args"]:
libs/checkpoint/pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "langgraph-checkpoint"
version = "1.0.3"
version = "1.0.4"
description = "Library with base interfaces for LangGraph checkpoint savers."
authors = []
license = "MIT"
libs/cli/pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "langgraph-cli"
version = "0.1.50"
version = "0.1.51"
description = "CLI for interacting with LangGraph API"
authors = []
license = "MIT"
libs/langgraph/langgraph/constants.py
+2 -2
View File
@@ -108,5 +108,5 @@ class Send:
@dataclass
class Interrupt:
when: Literal["before", "during", "after"]
value: Any = None
value: Any
when: Literal["during"] = "during"
libs/langgraph/langgraph/errors.py
+1 -1
View File
@@ -40,7 +40,7 @@ class NodeInterrupt(GraphInterrupt):
"""Raised by a node to interrupt execution."""
def __init__(self, value: Any) -> None:
super().__init__([Interrupt("during", value)])
super().__init__([Interrupt(value)])
class EmptyInputError(Exception):
libs/langgraph/langgraph/pregel/executor.py
+12 -11
View File
@@ -120,14 +120,14 @@ class AsyncBackgroundExecutor(AsyncContextManager):
def done(self, task: asyncio.Task) -> None:
try:
task.result()
except GraphInterrupt:
# This exception is an interruption signal, not an error
# so we don't want to re-raise it on exit
self.tasks.pop(task)
except BaseException:
pass
else:
if exc := task.exception():
# This exception is an interruption signal, not an error
# so we don't want to re-raise it on exit
if isinstance(exc, GraphInterrupt):
self.tasks.pop(task)
else:
self.tasks.pop(task)
except asyncio.CancelledError:
self.tasks.pop(task)
async def __aenter__(self) -> Submit:
@@ -146,12 +146,13 @@ class AsyncBackgroundExecutor(AsyncContextManager):
# wait for all tasks to finish
if self.tasks:
await asyncio.wait(self.tasks)
# re-raise the first exception that occurred in a task
# if there's already an exception being raised, don't raise another one
if exc_type is None:
# if there's already an exception being raised, don't raise another one
# re-raise the first exception that occurred in a task
for task in self.tasks:
try:
task.result()
if exc := task.exception():
raise exc
except asyncio.CancelledError:
pass
libs/langgraph/langgraph/pregel/loop.py
-4
View File
@@ -155,10 +155,6 @@ class PregelLoop:
self.stream_keys = stream_keys
self.is_nested = CONFIG_KEY_READ in self.config.get("configurable", {})
def mark_tasks_scheduled(self, tasks: Sequence[PregelExecutableTask]) -> None:
"""Mark tasks as scheduled, to be used by queue-based executors."""
raise NotImplementedError
def put_writes(self, task_id: str, writes: Sequence[tuple[str, Any]]) -> None:
"""Put writes for a task, to be read by the next tick."""
if not writes:
libs/langgraph/pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "langgraph"
version = "0.2.11"
version = "0.2.12"
description = "Building stateful, multi-actor applications with LLMs"
authors = []
license = "MIT"
libs/langgraph/tests/__snapshots__/test_pregel.ambr
+2900
View File
File diff suppressed because one or more lines are too long
libs/langgraph/tests/any_str.py
-18
View File
@@ -23,24 +23,6 @@ class AnyVersion:
return hash(str(self))
class ExceptionLike:
def __init__(self, exc: Exception) -> None:
self.exc = exc
def __eq__(self, value: object) -> bool:
return (
isinstance(value, Exception)
and self.exc.__class__ == value.__class__
and str(self.exc) == str(value)
)
def __hash__(self) -> int:
return hash((self.exc.__class__, str(self.exc)))
def __repr__(self) -> str:
return str(self.exc)
class UnsortedSequence:
def __init__(self, *values: Any) -> None:
self.seq = values
libs/langgraph/tests/memory_assert.py
-11
View File
@@ -24,8 +24,6 @@ class NoopSerializer(SerializerProtocol):
class MemorySaverAssertImmutable(MemorySaver):
serde = NoopSerializer()
storage_for_copies: defaultdict[str, dict[str, dict[str, Checkpoint]]]
def __init__(
@@ -74,15 +72,6 @@ class MemorySaverAssertCheckpointMetadata(MemorySaver):
should produce a side effect that can be asserted.
"""
serde = NoopSerializer()
def __init__(
self,
*,
serde: Optional[SerializerProtocol] = None,
) -> None:
super().__init__(serde=serde)
def put(
self,
config: RunnableConfig,
libs/langgraph/tests/test_pregel.py
+1015 -902
View File
File diff suppressed because it is too large Load Diff
libs/langgraph/tests/test_pregel_async.py
+909 -928
View File
File diff suppressed because it is too large Load Diff
libs/sdk-js/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@langchain/langgraph-sdk",
"version": "0.0.5",
"version": "0.0.6",
"description": "Client library for interacting with the LangGraph API",
"type": "module",
"packageManager": "yarn@1.22.19",
libs/sdk-js/src/client.mts
+10 -1
View File
@@ -571,6 +571,7 @@ export class RunsClient extends BaseClient {
);
let parser: EventSourceParser;
let onEndEvent: () => void;
const textDecoder = new TextDecoder();
const stream: ReadableStream<{ event: string; data: any }> = (
@@ -594,9 +595,17 @@ export class RunsClient extends BaseClient {
});
}
});
onEndEvent = () => {
ctrl.enqueue({ event: "end", data: undefined });
};
},
async transform(chunk) {
parser.feed(textDecoder.decode(chunk));
const payload = textDecoder.decode(chunk);
parser.feed(payload);
// eventsource-parser will ignore events
// that are not terminated by a newline
if (payload.trim() === "event: end") onEndEvent();
},
}),
);
libs/sdk-js/src/schema.ts
+12 -7
View File
@@ -2,6 +2,16 @@ import type { JSONSchema7 } from "json-schema";
type Optional<T> = T | null | undefined;
type RunStatus =
| "pending"
| "running"
| "error"
| "success"
| "timeout"
| "interrupted";
type ThreadStatus = "idle" | "busy" | "interrupted";
export interface Config {
/**
* Tags for this call and any sub-calls (eg. a Chain calling an LLM).
@@ -80,6 +90,7 @@ export interface Thread {
created_at: string;
updated_at: string;
metadata: Metadata;
status: ThreadStatus;
}
export interface Cron {
@@ -112,12 +123,6 @@ export interface Run {
assistant_id: string;
created_at: string;
updated_at: string;
status:
| "pending"
| "running"
| "error"
| "success"
| "timeout"
| "interrupted";
status: RunStatus;
metadata: Metadata;
}
libs/sdk-py/langgraph_sdk/client.py
+39 -4
View File
@@ -25,6 +25,7 @@ from langgraph_sdk.schema import (
Assistant,
Config,
Cron,
DisconnectMode,
GraphSchema,
Metadata,
MultitaskStrategy,
@@ -963,6 +964,8 @@ class RunsClient:
interrupt_before: Optional[list[str]] = None,
interrupt_after: Optional[list[str]] = None,
feedback_keys: Optional[list[str]] = None,
on_disconnect: Optional[DisconnectMode] = None,
webhook: Optional[str] = None,
multitask_strategy: Optional[MultitaskStrategy] = None,
) -> AsyncIterator[StreamPart]:
...
@@ -980,6 +983,8 @@ class RunsClient:
interrupt_before: Optional[list[str]] = None,
interrupt_after: Optional[list[str]] = None,
feedback_keys: Optional[list[str]] = None,
on_disconnect: Optional[DisconnectMode] = None,
webhook: Optional[str] = None,
) -> AsyncIterator[StreamPart]:
...
@@ -996,6 +1001,7 @@ class RunsClient:
interrupt_before: Optional[list[str]] = None,
interrupt_after: Optional[list[str]] = None,
feedback_keys: Optional[list[str]] = None,
on_disconnect: Optional[DisconnectMode] = None,
webhook: Optional[str] = None,
multitask_strategy: Optional[MultitaskStrategy] = None,
) -> AsyncIterator[StreamPart]:
@@ -1019,6 +1025,8 @@ class RunsClient:
webhook: Webhook to call after LangGraph API call is done.
multitask_strategy: Multitask strategy to use.
Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.
on_disconnect: The disconnect mode to use.
Must be one of 'cancel' or 'continue'.
Returns:
AsyncIterator[StreamPart]: Asynchronous iterator of stream results.
@@ -1061,6 +1069,7 @@ class RunsClient:
"webhook": webhook,
"checkpoint_id": checkpoint_id,
"multitask_strategy": multitask_strategy,
"on_disconnect": on_disconnect,
}
endpoint = (
f"/threads/{thread_id}/runs/stream"
@@ -1129,9 +1138,7 @@ class RunsClient:
config: The configuration for the assistant.
checkpoint_id: The checkpoint to start streaming from.
interrupt_before: Nodes to interrupt immediately before they get executed.
interrupt_after: Nodes to Nodes to interrupt immediately after they get executed.
webhook: Webhook to call after LangGraph API call is done.
multitask_strategy: Multitask strategy to use.
Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.
@@ -1242,6 +1249,8 @@ class RunsClient:
checkpoint_id: Optional[str] = None,
interrupt_before: Optional[list[str]] = None,
interrupt_after: Optional[list[str]] = None,
webhook: Optional[str] = None,
on_disconnect: Optional[DisconnectMode] = None,
multitask_strategy: Optional[MultitaskStrategy] = None,
) -> Union[list[dict], dict[str, Any]]:
...
@@ -1257,6 +1266,8 @@ class RunsClient:
config: Optional[Config] = None,
interrupt_before: Optional[list[str]] = None,
interrupt_after: Optional[list[str]] = None,
webhook: Optional[str] = None,
on_disconnect: Optional[DisconnectMode] = None,
) -> Union[list[dict], dict[str, Any]]:
...
@@ -1272,6 +1283,7 @@ class RunsClient:
interrupt_before: Optional[list[str]] = None,
interrupt_after: Optional[list[str]] = None,
webhook: Optional[str] = None,
on_disconnect: Optional[DisconnectMode] = None,
multitask_strategy: Optional[MultitaskStrategy] = None,
) -> Union[list[dict], dict[str, Any]]:
"""Create a run, wait until it finishes and return the final state.
@@ -1286,12 +1298,12 @@ class RunsClient:
config: The configuration for the assistant.
checkpoint_id: The checkpoint to start streaming from.
interrupt_before: Nodes to interrupt immediately before they get executed.
interrupt_after: Nodes to Nodes to interrupt immediately after they get executed.
webhook: Webhook to call after LangGraph API call is done.
multitask_strategy: Multitask strategy to use.
Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.
on_disconnect: The disconnect mode to use.
Must be one of 'cancel' or 'continue'.
Returns:
Union[list[dict], dict[str, Any]]: The output of the run.
@@ -1351,6 +1363,7 @@ class RunsClient:
"webhook": webhook,
"checkpoint_id": checkpoint_id,
"multitask_strategy": multitask_strategy,
"on_disconnect": on_disconnect,
}
endpoint = (
f"/threads/{thread_id}/runs/wait" if thread_id is not None else "/runs/wait"
@@ -1451,6 +1464,28 @@ class RunsClient:
""" # noqa: E501
return await self.http.get(f"/threads/{thread_id}/runs/{run_id}/join")
def join_stream(self, thread_id: str, run_id: str) -> AsyncIterator[StreamPart]:
"""Stream output from a run in real-time, until the run is done.
Output is not buffered, so any output produced before this call will
not be received here.
Args:
thread_id: The thread ID to join.
run_id: The run ID to join.
Returns:
None
Example Usage:
await client.runs.join(
thread_id="thread_id_to_join",
run_id="run_id_to_join"
)
""" # noqa: E501
return self.http.stream(f"/threads/{thread_id}/runs/{run_id}/stream", "GET")
async def delete(self, thread_id: str, run_id: str) -> None:
"""Delete a run.
libs/sdk-py/langgraph_sdk/schema.py
+2
View File
@@ -9,6 +9,8 @@ ThreadStatus = Literal["idle", "busy", "interrupted"]
StreamMode = Literal["values", "messages", "updates", "events", "debug"]
DisconnectMode = Literal["cancel", "continue"]
MultitaskStrategy = Literal["reject", "interrupt", "rollback", "enqueue"]
OnConflictBehavior = Literal["raise", "do_nothing"]