nat.plugins.langchain.langgraph_workflow#

Attributes#

`GraphDefType`
`logger`

Classes#

`LanggraphWrapperInput`	Input model for the LangGraph wrapper.
`LanggraphWrapperOutput`	Output model for the LangGraph wrapper.
`LanggraphWrapperConfig`	Configuration model for the LangGraph wrapper.
`LanggraphWrapperFunction`	Function for the LangGraph wrapper.

Functions#

register(config, b)

Module Contents#

GraphDefType#

logger#

class LanggraphWrapperInput(/, **data: Any)#

Bases: pydantic.BaseModel

Input model for the LangGraph wrapper.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

messages: list[langchain_core.messages.MessageLikeRepresentation] | langchain_core.prompt_values.PromptValue#

class LanggraphWrapperOutput(/, **data: Any)#

Bases: pydantic.BaseModel

Output model for the LangGraph wrapper.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

messages: list[langchain_core.messages.BaseMessage]#

class LanggraphWrapperConfig#

Bases: nat.data_models.function.FunctionBaseConfig

Configuration model for the LangGraph wrapper.

model_config#

description: str = ''#

dependencies: list[pydantic.DirectoryPath] = None#

graph: str#

env: pydantic.FilePath | dict[str, str] | None = None#

class LanggraphWrapperFunction( *, config: LanggraphWrapperConfig, description: str | None = None, graph: langgraph.graph.state.CompiledStateGraph, )#

Bases: nat.builder.function.Function[LanggraphWrapperInput, LanggraphWrapperOutput, LanggraphWrapperOutput]

Function for the LangGraph wrapper.

Initialize the LangGraph wrapper function.

Args:: config: The configuration for the LangGraph wrapper. description: The description of the LangGraph wrapper. graph: The graph to wrap.

_graph#

async _ainvoke(value: LanggraphWrapperInput) → LanggraphWrapperOutput#

async _astream( value: LanggraphWrapperInput, ) → collections.abc.AsyncGenerator[LanggraphWrapperOutput, None]#

static _parse_stream_output(output: dict) → LanggraphWrapperOutput#: Unwrap node-keyed dicts that LangGraph astream() yields.

static convert_to_str(value: LanggraphWrapperOutput) → str#: Convert the output to a string.

static convert_chat_request( value: nat.data_models.api_server.ChatRequest, ) → LanggraphWrapperInput#: Convert a ChatRequest to LanggraphWrapperInput.

static convert_str(value: str) → LanggraphWrapperInput#: Convert a plain text string to LanggraphWrapperInput.

static convert_to_chat_response( value: LanggraphWrapperOutput, ) → nat.data_models.api_server.ChatResponse#: Convert LanggraphWrapperOutput to ChatResponse.

static convert_to_chat_response_chunk( value: LanggraphWrapperOutput, ) → nat.data_models.api_server.ChatResponseChunk#: Convert LanggraphWrapperOutput to ChatResponseChunk.

async register( config: LanggraphWrapperConfig, b: nat.builder.builder.Builder, )#