agents.rag.multi_agent_rag.multi_rag

Multi-Agent RAG System Implementation.

from typing import Any This module provides complete multi-agent RAG workflows using the multi-agent framework with conditional routing, sequential processing, and parallel execution patterns.

Classes

AdaptiveRAGMultiAgent	Advanced RAG system that adapts its strategy based on query complexity and results.
BaseRAGMultiAgent	Base multi-agent RAG system with retrieve -> grade -> generate workflow.
ConditionalRAGMultiAgent	Conditional multi-agent RAG system with smart routing based on document quality.
IterativeRAGMultiAgent	Multi-agent RAG system with iterative document processing.
ParallelRAGMultiAgent	Parallel multi-agent RAG system for consensus-based processing.

Functions

create_conditional_rag_system([documents, custom_grader])	Create a conditional RAG system with smart routing.
create_iterative_rag_system([documents, custom_grader])	Create an iterative RAG system with document-by-document processing.
create_sequential_rag_system([documents, use_grading, ...])	Create a sequential RAG system with configurable components.
document_quality_check(state)	Check document quality and route accordingly.
should_grade_documents(state)	Conditional routing function to determine if documents need grading.
should_refine_query(state)	Conditional routing function to determine if query needs refinement.
test_agent_compatibility(agent1, agent2)	Test compatibility between two agents using the compatibility module.
validate_multi_agent_compatibility(agents)	Validate compatibility across multiple agents in a workflow.

Module Contents

class agents.rag.multi_agent_rag.multi_rag.AdaptiveRAGMultiAgent(simple_rag=None, complex_rag=None, consensus_rag=None, **kwargs)

Bases: haive.agents.multi.base.ConditionalAgent

Advanced RAG system that adapts its strategy based on query complexity and results.

This system demonstrates sophisticated conditional routing with multiple decision points and fallback strategies.

Init .

Parameters:

• simple_rag (BaseRAGMultiAgent | None) – [TODO: Add description]

• complex_rag (IterativeRAGMultiAgent | None) – [TODO: Add description]

• consensus_rag (ParallelRAGMultiAgent | None) – [TODO: Add description]

class agents.rag.multi_agent_rag.multi_rag.BaseRAGMultiAgent(retrieval_agent=None, grading_agent=None, answer_agent=None, **kwargs)

Bases: haive.agents.multi.base.SequentialAgent

Base multi-agent RAG system with retrieve -> grade -> generate workflow.

This is the simple sequential RAG agent as mentioned in the user prompt.

Init .

Parameters:

• retrieval_agent (haive.agents.rag.multi_agent_rag.agents.SimpleRAGAgent | None) – [TODO: Add description]

• grading_agent (haive.agents.rag.multi_agent_rag.agents.DocumentGradingAgent | None) – [TODO: Add description]

• answer_agent (haive.agents.rag.multi_agent_rag.agents.SimpleRAGAnswerAgent | None) – [TODO: Add description]

class agents.rag.multi_agent_rag.multi_rag.ConditionalRAGMultiAgent(retrieval_agent=None, grading_agent=None, answer_agent=None, query_refiner=None, **kwargs)

Bases: haive.agents.multi.base.ConditionalAgent

Conditional multi-agent RAG system with smart routing based on document quality.

This system uses conditional routing to decide whether to grade documents, refine queries, or generate answers based on the current state.

Init .

Parameters:

• retrieval_agent (haive.agents.rag.multi_agent_rag.agents.SimpleRAGAgent | None) – [TODO: Add description]

• grading_agent (haive.agents.rag.multi_agent_rag.agents.DocumentGradingAgent | None) – [TODO: Add description]

• answer_agent (haive.agents.rag.multi_agent_rag.agents.SimpleRAGAnswerAgent | None) – [TODO: Add description]

• query_refiner (Any | None) – [TODO: Add description]

class agents.rag.multi_agent_rag.multi_rag.IterativeRAGMultiAgent(retrieval_agent=None, iterative_grader=None, answer_agent=None, custom_grader_callable=None, **kwargs)

Bases: haive.agents.multi.base.SequentialAgent

Multi-agent RAG system with iterative document processing.

This system demonstrates iterating over retrieved documents and processing each one individually, as mentioned in the user prompt.

Init .

Parameters:

• retrieval_agent (haive.agents.rag.multi_agent_rag.agents.SimpleRAGAgent | None) – [TODO: Add description]

• iterative_grader (haive.agents.rag.multi_agent_rag.agents.IterativeDocumentGradingAgent | None) – [TODO: Add description]

• answer_agent (haive.agents.rag.multi_agent_rag.agents.SimpleRAGAnswerAgent | None) – [TODO: Add description]

• custom_grader_callable (collections.abc.Callable | None) – [TODO: Add description]

class agents.rag.multi_agent_rag.multi_rag.ParallelRAGMultiAgent(rag_agents=None, **kwargs)

Bases: haive.agents.multi.base.ParallelAgent

Parallel multi-agent RAG system for consensus-based processing.

This system runs multiple RAG agents in parallel and aggregates their results.

Init .

Parameters:

rag_agents (list[BaseRAGMultiAgent] | None) – [TODO: Add description]

agents.rag.multi_agent_rag.multi_rag.create_conditional_rag_system(documents=None, custom_grader=None)

Create a conditional RAG system with smart routing.

Parameters:

• documents (list[langchain_core.documents.Document] | None)

• custom_grader (collections.abc.Callable | None)

Return type:

ConditionalRAGMultiAgent

agents.rag.multi_agent_rag.multi_rag.create_iterative_rag_system(documents=None, custom_grader=None)

Create an iterative RAG system with document-by-document processing.

Parameters:

• documents (list[langchain_core.documents.Document] | None)

• custom_grader (collections.abc.Callable | None)

Return type:

IterativeRAGMultiAgent

agents.rag.multi_agent_rag.multi_rag.create_sequential_rag_system(documents=None, use_grading=True, use_citations=False)

Create a sequential RAG system with configurable components.

Parameters:

• documents (list[langchain_core.documents.Document] | None)

• use_grading (bool)

• use_citations (bool)

Return type:

haive.agents.multi.base.SequentialAgent

agents.rag.multi_agent_rag.multi_rag.document_quality_check(state)

Check document quality and route accordingly.

Returns:

Documents are good enough for generation - “insufficient”: Need more/better documents - “regrade”: Need different grading approach

Return type:

• “sufficient”

Parameters:

state (haive.agents.rag.multi_agent_rag.state.MultiAgentRAGState)

agents.rag.multi_agent_rag.multi_rag.should_grade_documents(state)

Conditional routing function to determine if documents need grading.

Returns:

If documents need to be graded - “generate”: If documents are already good enough - “refine”: If query needs refinement

Return type:

• “grade”

Parameters:

state (haive.agents.rag.multi_agent_rag.state.MultiAgentRAGState)

agents.rag.multi_agent_rag.multi_rag.should_refine_query(state)

Conditional routing function to determine if query needs refinement.

Returns:

Try retrieval again with refined query - “generate”: Generate answer with available documents - “END”: Stop processing (failure case)

Return type:

• “retrieve”

Parameters:

state (haive.agents.rag.multi_agent_rag.state.MultiAgentRAGState)

agents.rag.multi_agent_rag.multi_rag.test_agent_compatibility(agent1, agent2)

Test compatibility between two agents using the compatibility module.

This demonstrates using the compatibility module to test if agents can work together as mentioned in the user prompt.

Parameters:

• agent1 (Any)

• agent2 (Any)

Return type:

dict[str, Any]

agents.rag.multi_agent_rag.multi_rag.validate_multi_agent_compatibility(agents)

Validate compatibility across multiple agents in a workflow.

Returns a comprehensive compatibility report for the agent chain.

Parameters:

agents (list[Any])

Return type:

dict[str, Any]