diff --git a/libs/langgraph/langgraph/prebuilt/chat_agent_executor.py b/libs/langgraph/langgraph/prebuilt/chat_agent_executor.py index fc812ccbc..4c8699360 100644 --- a/libs/langgraph/langgraph/prebuilt/chat_agent_executor.py +++ b/libs/langgraph/langgraph/prebuilt/chat_agent_executor.py @@ -212,6 +212,7 @@ def create_react_agent( Args: model: The `LangChain` chat model that supports tool calling. tools: A list of tools, a ToolExecutor, or a ToolNode instance. + If an empty list is provided, the agent will consist of a single LLM node without tool calling. 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. @@ -540,19 +541,10 @@ def create_react_agent( # get the tool functions wrapped in a tool class from the ToolNode tool_classes = list(tool_node.tools_by_name.values()) - if _should_bind_tools(model, tool_classes): - model = cast(BaseChatModel, model).bind_tools(tool_classes) + tool_calling_enabled = len(tool_classes) > 0 - # Define the function that determines whether to continue or not - def should_continue(state: AgentState) -> Literal["tools", "__end__"]: - messages = state["messages"] - last_message = messages[-1] - # If there is no function call, then we finish - if not isinstance(last_message, AIMessage) or not last_message.tool_calls: - return "__end__" - # Otherwise if there is, we continue - else: - return "tools" + if _should_bind_tools(model, tool_classes) and tool_calling_enabled: + model = cast(BaseChatModel, model).bind_tools(tool_classes) # we're passing store here for validation preprocessor = _get_model_preprocessing_runnable( @@ -635,6 +627,30 @@ def create_react_agent( # We return a list, because this will get added to the existing list return {"messages": [response]} + if not tool_calling_enabled: + # Define a new graph + workflow = StateGraph(state_schema or AgentState) + workflow.add_node("agent", RunnableCallable(call_model, acall_model)) + workflow.set_entry_point("agent") + return workflow.compile( + checkpointer=checkpointer, + store=store, + interrupt_before=interrupt_before, + interrupt_after=interrupt_after, + debug=debug, + ) + + # Define the function that determines whether to continue or not + def should_continue(state: AgentState) -> Literal["tools", "__end__"]: + messages = state["messages"] + last_message = messages[-1] + # If there is no function call, then we finish + if not isinstance(last_message, AIMessage) or not last_message.tool_calls: + return "__end__" + # Otherwise if there is, we continue + else: + return "tools" + # Define a new graph workflow = StateGraph(state_schema or AgentState) diff --git a/libs/langgraph/langgraph/pregel/algo.py b/libs/langgraph/langgraph/pregel/algo.py index 1410e432f..3c3008ce4 100644 --- a/libs/langgraph/langgraph/pregel/algo.py +++ b/libs/langgraph/langgraph/pregel/algo.py @@ -1,3 +1,4 @@ +import sys from collections import defaultdict, deque from functools import partial from hashlib import sha1 @@ -66,6 +67,7 @@ from langgraph.types import All, LoopProtocol, PregelExecutableTask, PregelTask from langgraph.utils.config import merge_configs, patch_config GetNextVersion = Callable[[Optional[V], BaseChannel], V] +SUPPORTS_EXC_NOTES = sys.version_info >= (3, 11) class WritesProtocol(Protocol): @@ -634,6 +636,12 @@ def prepare_single_task( ) except StopIteration: return + except Exception as exc: + if SUPPORTS_EXC_NOTES: + exc.add_note( + f"Before task with name '{name}' and path '{task_path[:3]}'" + ) + raise # create task id checkpoint_ns = f"{parent_ns}{NS_SEP}{name}" if parent_ns else name diff --git a/libs/langgraph/langgraph/pregel/retry.py b/libs/langgraph/langgraph/pregel/retry.py index 6e52a7c41..2d0f2b6da 100644 --- a/libs/langgraph/langgraph/pregel/retry.py +++ b/libs/langgraph/langgraph/pregel/retry.py @@ -1,6 +1,7 @@ import asyncio import logging import random +import sys import time from dataclasses import replace from functools import partial @@ -18,6 +19,7 @@ from langgraph.types import Command, PregelExecutableTask, RetryPolicy from langgraph.utils.config import patch_configurable logger = logging.getLogger(__name__) +SUPPORTS_EXC_NOTES = sys.version_info >= (3, 11) def run_with_retry( @@ -60,6 +62,8 @@ def run_with_retry( # if interrupted, end raise except Exception as exc: + if SUPPORTS_EXC_NOTES: + exc.add_note(f"During task with name '{task.name}' and id '{task.id}'") if retry_policy is None: raise # increment attempts @@ -152,6 +156,8 @@ async def arun_with_retry( # if interrupted, end raise except Exception as exc: + if SUPPORTS_EXC_NOTES: + exc.add_note(f"During task with name '{task.name}' and id '{task.id}'") if retry_policy is None: raise # increment attempts diff --git a/libs/langgraph/tests/test_prebuilt.py b/libs/langgraph/tests/test_prebuilt.py index a6655a451..0997668b2 100644 --- a/libs/langgraph/tests/test_prebuilt.py +++ b/libs/langgraph/tests/test_prebuilt.py @@ -102,6 +102,9 @@ class FakeToolCallingModel(BaseChatModel): tools: Sequence[Union[Dict[str, Any], Type[BaseModel], Callable, BaseTool]], **kwargs: Any, ) -> Runnable[LanguageModelInput, BaseMessage]: + if len(tools) == 0: + raise ValueError("Must provide at least one tool") + tool_dicts = [] for tool in tools: if not isinstance(tool, BaseTool): diff --git a/libs/sdk-js/package.json b/libs/sdk-js/package.json index bc37a9d4e..ded99a00b 100644 --- a/libs/sdk-js/package.json +++ b/libs/sdk-js/package.json @@ -1,6 +1,6 @@ { "name": "@langchain/langgraph-sdk", - "version": "0.0.26", + "version": "0.0.29", "description": "Client library for interacting with the LangGraph API", "type": "module", "packageManager": "yarn@1.22.19", diff --git a/libs/sdk-js/src/client.ts b/libs/sdk-js/src/client.ts index 4829a831f..010864419 100644 --- a/libs/sdk-js/src/client.ts +++ b/libs/sdk-js/src/client.ts @@ -33,6 +33,7 @@ import { OnConflictBehavior, } from "./types.js"; import { mergeSignals } from "./utils/signals.js"; +import { getEnvironmentVariable } from "./utils/env.js"; /** * Get the API key from the environment. @@ -53,7 +54,7 @@ export function getApiKey(apiKey?: string): string | undefined { const prefixes = ["LANGGRAPH", "LANGSMITH", "LANGCHAIN"]; for (const prefix of prefixes) { - const envKey = process.env[`${prefix}_API_KEY`]; + const envKey = getEnvironmentVariable(`${prefix}_API_KEY`); if (envKey) { // Remove surrounding quotes return envKey.trim().replace(/^["']|["']$/g, ""); diff --git a/libs/sdk-js/src/index.ts b/libs/sdk-js/src/index.ts index f86406100..764abbc67 100644 --- a/libs/sdk-js/src/index.ts +++ b/libs/sdk-js/src/index.ts @@ -15,6 +15,7 @@ export type { ThreadStatus, Cron, Checkpoint, + Interrupt, } from "./schema.js"; export type { OnConflictBehavior, Command } from "./types.js"; diff --git a/libs/sdk-js/src/schema.ts b/libs/sdk-js/src/schema.ts index 1b9dae1fe..dd79e07ba 100644 --- a/libs/sdk-js/src/schema.ts +++ b/libs/sdk-js/src/schema.ts @@ -137,6 +137,16 @@ export interface AssistantGraph { }>; } +/** + * An interrupt thrown inside a thread. + */ +export interface Interrupt { + value: unknown; + when: "during"; + resumable: boolean; + ns?: string[]; +} + export interface Thread { /** The ID of the thread. */ thread_id: string; @@ -155,6 +165,9 @@ export interface Thread { /** The current state of the thread. */ values: ValuesType; + + /** Interrupts which were thrown in this thread */ + interrupts: Record>; } export interface Cron { @@ -210,12 +223,7 @@ export interface ThreadTask { name: string; result?: unknown; error: Optional; - interrupts: Array<{ - value: unknown; - when: "during"; - resumable: boolean; - ns?: string[]; - }>; + interrupts: Array; checkpoint: Optional; state: Optional; } diff --git a/libs/sdk-js/src/utils/async_caller.ts b/libs/sdk-js/src/utils/async_caller.ts index d58823ba4..d5de9131a 100644 --- a/libs/sdk-js/src/utils/async_caller.ts +++ b/libs/sdk-js/src/utils/async_caller.ts @@ -4,6 +4,7 @@ import PQueueMod from "p-queue"; const STATUS_NO_RETRY = [ 400, // Bad Request 401, // Unauthorized + 402, // Payment required 403, // Forbidden 404, // Not Found 405, // Method Not Allowed diff --git a/libs/sdk-js/src/utils/env.ts b/libs/sdk-js/src/utils/env.ts new file mode 100644 index 000000000..738c14fd5 --- /dev/null +++ b/libs/sdk-js/src/utils/env.ts @@ -0,0 +1,11 @@ +export function getEnvironmentVariable(name: string): string | undefined { + // Certain setups (Deno, frontend) will throw an error if you try to access environment variables + try { + return typeof process !== "undefined" + ? // eslint-disable-next-line no-process-env + process.env?.[name] + : undefined; + } catch (e) { + return undefined; + } +} diff --git a/libs/sdk-py/langgraph_sdk/client.py b/libs/sdk-py/langgraph_sdk/client.py index 6a3bb6c9c..a018bb3b6 100644 --- a/libs/sdk-py/langgraph_sdk/client.py +++ b/libs/sdk-py/langgraph_sdk/client.py @@ -59,6 +59,7 @@ from langgraph_sdk.schema import ( ThreadStatus, ThreadUpdateStateResponse, ) +from langgraph_sdk.sse import EventSource logger = logging.getLogger(__name__) @@ -292,7 +293,7 @@ class HttpClient: else: logger.error(f"Error from langgraph-api: {body}", exc_info=e) raise e - async for event in sse.aiter_sse(): + async for event in EventSource(sse.response).aiter_sse(): yield StreamPart( event.event, orjson.loads(event.data) if event.data else None ) @@ -2426,7 +2427,7 @@ class SyncHttpClient: else: logger.error(f"Error from langgraph-api: {body}", exc_info=e) raise e - for event in sse.iter_sse(): + for event in EventSource(sse.response).iter_sse(): yield StreamPart( event.event, orjson.loads(event.data) if event.data else None ) diff --git a/libs/sdk-py/langgraph_sdk/sse.py b/libs/sdk-py/langgraph_sdk/sse.py new file mode 100644 index 000000000..b87788387 --- /dev/null +++ b/libs/sdk-py/langgraph_sdk/sse.py @@ -0,0 +1,106 @@ +"""Adapted from httpx_sse to split lines on \n, \r, \r\n per the SSE spec.""" + +import io +from typing import AsyncIterator, Iterator + +import httpx +import httpx_sse +import httpx_sse._decoders + + +class BytesLineDecoder: + """ + Handles incrementally reading lines from text. + + Has the same behaviour as the stdllib bytes splitlines, + but handling the input iteratively. + """ + + def __init__(self) -> None: + self.buffer = io.BytesIO() + self.trailing_cr: bool = False + + def decode(self, text: bytes) -> list[bytes]: + # See https://docs.python.org/3/glossary.html#term-universal-newlines + NEWLINE_CHARS = b"\n\r" + + # We always push a trailing `\r` into the next decode iteration. + if self.trailing_cr: + text = b"\r" + text + self.trailing_cr = False + if text.endswith(b"\r"): + self.trailing_cr = True + text = text[:-1] + + if not text: + # NOTE: the edge case input of empty text doesn't occur in practice, + # because other httpx internals filter out this value + return [] # pragma: no cover + + trailing_newline = text[-1] in NEWLINE_CHARS + lines = text.splitlines() + + if len(lines) == 1 and not trailing_newline: + # No new lines, buffer the input and continue. + self.buffer.append(lines[0]) + return [] + + if self.buffer: + # Include any existing buffer in the first portion of the + # splitlines result. + lines = [self.buffer.getvalue() + lines[0]] + lines[1:] + self.buffer.truncate(0) + + if not trailing_newline: + # If the last segment of splitlines is not newline terminated, + # then drop it from our output and start a new buffer. + self.buffer.write(lines.pop()) + + return lines + + def flush(self) -> list[bytes]: + if not self.buffer and not self.trailing_cr: + return [] + + lines = [self.buffer.getvalue()] if self.buffer else [] + self.buffer.truncate(0) + self.trailing_cr = False + return lines + + +async def aiter_lines_raw(response: httpx.Response) -> AsyncIterator[bytes]: + decoder = BytesLineDecoder() + async for chunk in response.aiter_bytes(): + for line in decoder.decode(chunk): + yield line + for line in decoder.flush(): + yield line + + +def iter_lines_raw(response: httpx.Response) -> Iterator[bytes]: + decoder = BytesLineDecoder() + for chunk in response.iter_bytes(): + for line in decoder.decode(chunk): + yield line + for line in decoder.flush(): + yield line + + +class EventSource(httpx_sse.EventSource): + async def aiter_sse(self) -> AsyncIterator[httpx_sse.ServerSentEvent]: + self._check_content_type() + decoder = httpx_sse._decoders.SSEDecoder() + async for line in aiter_lines_raw(self._response): + line = line.rstrip(b"\n") + sse = decoder.decode(line.decode()) + if sse is not None: + yield sse + + def iter_sse(self) -> Iterator[httpx_sse.ServerSentEvent]: + self._check_content_type() + decoder = httpx_sse._decoders.SSEDecoder() + for line in iter_lines_raw(self._response): + line = line.rstrip(b"\n") + sse = decoder.decode(line.decode()) + if sse is not None: + yield sse diff --git a/libs/sdk-py/pyproject.toml b/libs/sdk-py/pyproject.toml index 393750ba6..c9e0401da 100644 --- a/libs/sdk-py/pyproject.toml +++ b/libs/sdk-py/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "langgraph-sdk" -version = "0.1.36" +version = "0.1.37" description = "SDK for interacting with LangGraph API" authors = [] license = "MIT"