nat.plugins.langchain.agent.rewoo_agent.agent#

Attributes#

logger

Classes#

`ReWOOEvidence`
`ReWOOPlanStep`
`ReWOOGraphState`	State schema for the ReWOO Agent Graph
`ReWOOAgentGraph`	Configurable ReWOO Agent.

Module Contents#

logger#

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

Bases: pydantic.BaseModel

placeholder: str#

tool: str#

tool_input: Any#

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

Bases: pydantic.BaseModel

plan: str#

evidence: ReWOOEvidence#

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

Bases: pydantic.BaseModel

State schema for the ReWOO Agent Graph

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.

messages: list[langchain_core.messages.base.BaseMessage] = None#

task: langchain_core.messages.human.HumanMessage = None#

plan: langchain_core.messages.ai.AIMessage = None#

steps: langchain_core.messages.ai.AIMessage = None#

evidence_map: dict[str, ReWOOPlanStep] = None#

execution_levels: list[list[str]] = None#

current_level: int = None#

intermediate_results: dict[str, langchain_core.messages.tool.ToolMessage] = None#

result: langchain_core.messages.ai.AIMessage = None#

class ReWOOAgentGraph( llm: langchain_core.language_models.BaseChatModel, planner_prompt: langchain_core.prompts.chat.ChatPromptTemplate, solver_prompt: langchain_core.prompts.chat.ChatPromptTemplate, tools: list[langchain_core.tools.BaseTool], use_tool_schema: bool = True, callbacks: list[langchain_core.callbacks.base.AsyncCallbackHandler] | None = None, detailed_logs: bool = False, log_response_max_chars: int = 1000, tool_call_max_retries: int = 3, raise_tool_call_error: bool = True, )#

Bases: nat.plugins.langchain.agent.base.BaseAgent

Configurable ReWOO Agent.

Args:: detailed_logs: Toggles logging of inputs, outputs, and intermediate steps.

planner_prompt#

solver_prompt#

tools_dict#

tool_call_max_retries = 3#

raise_tool_call_error = True#

_get_tool(tool_name: str)#

static _get_current_level_status(state: ReWOOGraphState) → tuple[int, bool]#

Get the current execution level and whether it’s complete.

Args:: state: The ReWOO graph state.
Returns:: tuple of (current_level, is_complete). Level -1 means all execution is complete.

static _parse_planner_output(planner_output: str) → list[ReWOOPlanStep]#

static _parse_planner_dependencies( steps: list[ReWOOPlanStep], ) → tuple[dict[str, ReWOOPlanStep], list[list[str]]]#

Parse planner steps to identify dependencies and create execution levels for parallel processing. This creates a dependency map and identifies which evidence placeholders can be executed in parallel.

Args:: steps: list of plan steps from the planner.
Returns:: A mapping from evidence placeholders to step info and execution levels for parallel processing.

static _replace_placeholder( placeholder: str, tool_input: str | dict, tool_output: str | dict, ) → str | dict#

static _parse_tool_input(tool_input: str | dict)#

async planner_node(state: ReWOOGraphState)#

async executor_node(state: ReWOOGraphState)#

Execute tools in parallel for the current dependency level.

This replaces the sequential execution with parallel execution of tools that have no dependencies between them.

async _execute_single_tool( placeholder: str, step_info: ReWOOPlanStep, intermediate_results: dict[str, langchain_core.messages.tool.ToolMessage], ) → langchain_core.messages.tool.ToolMessage#

Execute a single tool with proper placeholder replacement.

Args:: placeholder (str): The evidence placeholder (e.g., “#E1”). step_info (ReWOOPlanStep): Step information containing tool and tool_input. intermediate_results (dict[str, ToolMessage]): Current intermediate results for placeholder replacement.
Returns:: ToolMessage with the tool execution result.

async solver_node(state: ReWOOGraphState)#

async conditional_edge(state: ReWOOGraphState)#

async _build_graph( state_schema: type, ) → langgraph.graph.state.CompiledStateGraph#

async build_graph()#

static validate_planner_prompt(planner_prompt: str) → bool#

static validate_solver_prompt(solver_prompt: str) → bool#