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(/, **data: Any)#

Bases: nat.data_models.function.FunctionBaseConfig

Configuration 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].

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, types.NoneType, 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#

_convert_input(value: Any) → LanggraphWrapperInput#

async _ainvoke(value: LanggraphWrapperInput) → LanggraphWrapperOutput#

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

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

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