agents.rag.multi_agent_rag.state¶

Enhanced RAG State Schema for Multi-Agent RAG Systems.

This module provides comprehensive state management for complex RAG workflows, supporting document processing, grading, multi-step retrieval, and conditional routing.

Classes¶

`DocumentGradingResult`	Result of document grading process.
`MultiAgentRAGState`	Comprehensive state schema for multi-agent RAG systems.
`QueryStatus`	Status of query processing.
`RAGOperationType`	Types of RAG operations that can be performed.
`RAGStep`	Represents a single step in the RAG workflow.

Module Contents¶

class agents.rag.multi_agent_rag.state.DocumentGradingResult(/, **data)¶

Bases: pydantic.BaseModel

Result of document grading process.

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.

Parameters:: data (Any)

class agents.rag.multi_agent_rag.state.MultiAgentRAGState(/, **data)¶

Bases: haive.core.schema.state_schema.StateSchema

Comprehensive state schema for multi-agent RAG systems.

Supports complex RAG workflows with document grading, multi-step retrieval, conditional routing, and state tracking across multiple agents.

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.

Parameters:: data (Any)

add_workflow_step(operation_type, agent_name, input_data=None, output_data=None)¶

Add a new workflow step.

Parameters:

operation_type (RAGOperationType)
agent_name (str)
input_data (dict[str, Any] | None)
output_data (dict[str, Any] | None)

Return type:

str

get_latest_step(operation_type=None)¶

Get the most recent workflow step, optionally filtered by operation type.

Parameters:: operation_type (RAGOperationType | None)
Return type:: RAGStep | None

get_relevant_documents(min_score=0.5)¶

Get documents that passed relevance threshold.

Parameters:: min_score (float)
Return type:: list[langchain_core.documents.Document]

should_refine_query()¶

Determine if query should be refined based on state.

Return type:: bool

update_quality_metrics()¶

Update quality metrics based on current state.

Return type:: None

class agents.rag.multi_agent_rag.state.QueryStatus¶

Bases: str, enum.Enum

Status of query processing.

Initialize self. See help(type(self)) for accurate signature.

class agents.rag.multi_agent_rag.state.RAGOperationType¶

Bases: str, enum.Enum

Types of RAG operations that can be performed.

Initialize self. See help(type(self)) for accurate signature.

class agents.rag.multi_agent_rag.state.RAGStep(/, **data)¶

Bases: pydantic.BaseModel

Represents a single step in the RAG workflow.

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.

Parameters:: data (Any)