agents.rag.factories.compatible_rag_factory¶

Compatible RAG Workflow Factory.

Generic factory for building composable RAG workflows based on I/O schema compatibility. Uses the enhanced multi-agent base with automatic compatibility checking, agent replacement, and workflow optimization. Allows replacing agents by compatible I/O schemas.

Key Features:

Automatic I/O schema compatibility analysis
Agent replacement based on schema compatibility
Workflow optimization for better field flow
Component-based RAG workflow building
Integration with search tools from haive.tools
Enhanced system prompts and grading

Classes¶

`CompatibleRAGFactory`	Factory for building RAG workflows with I/O schema compatibility.
`RAGComponent`	Available RAG components for composition.
`WorkflowPattern`	Common workflow patterns from the architecture guide.

Functions¶

`create_plug_and_play_component`(component_type, documents)	Create any RAG component as a standalone agent.
`get_component_compatibility_info`(component_type)	Get I/O schema information for a component type.

Module Contents¶

class agents.rag.factories.compatible_rag_factory.CompatibleRAGFactory(documents, llm_config=None, enable_search_tools=False, default_embedding_model=None)¶

Factory for building RAG workflows with I/O schema compatibility.

Uses the enhanced multi-agent base with automatic compatibility checking, agent replacement, and workflow optimization.

Initialize factory with common dependencies.

Parameters:

documents (list[langchain_core.documents.Document]) – Document collection for retrieval
llm_config (haive.core.models.llm.base.LLMConfig | None) – LLM configuration for all components
enable_search_tools (bool) – Whether to integrate external search tools
default_embedding_model (str | None) – Default embedding model for vector stores

analyze_workflow_compatibility(workflow)¶

Analyze I/O compatibility of existing workflow.

Parameters:: workflow (SequentialAgent | ConditionalAgent) – Workflow to analyze
Returns:: Compatibility analysis report
Return type:: dict[str, Any]

build_graph()¶

Build graph with compatibility-aware callable sequence.

Return type:: haive.core.graph.state_graph.base_graph2.BaseGraph

classmethod create_agentic_search_workflow(documents, llm_config=None, **kwargs)¶

Create agentic workflow with search tool integration.

Parameters:

documents (list[langchain_core.documents.Document])
llm_config (haive.core.models.llm.base.LLMConfig | None)

Return type:

ConditionalAgent

classmethod create_decomposed_graded_workflow(documents, llm_config=None, enable_hallucination_grading=True, **kwargs)¶

Create Query Decomposition → Retrieval → Grading → Hallucination Check → Answer workflow.

Parameters:

documents (list[langchain_core.documents.Document])
llm_config (haive.core.models.llm.base.LLMConfig | None)
enable_hallucination_grading (bool)

Return type:

SequentialAgent

create_from_schema_compatibility(component_sequence, auto_optimize=True, **kwargs)¶

Create workflow by chaining components with compatible I/O schemas.

This is the core generic functionality - it analyzes I/O schemas and builds a compatible chain using the enhanced multi-agent base.

Parameters:

component_sequence (list[RAGComponent]) – Ordered list of components to chain
auto_optimize (bool) – Whether to auto-optimize agent order
**kwargs – Additional configuration

Returns:

SequentialAgent with compatible component chain

Return type:

SequentialAgent

classmethod create_full_pipeline_workflow(documents, llm_config=None, **kwargs)¶

Create comprehensive pipeline with all major components.

Pipeline: Decomposition → HyDE → Retrieval → Grading → Hallucination Check → Answer

Parameters:

documents (list[langchain_core.documents.Document])
llm_config (haive.core.models.llm.base.LLMConfig | None)

Return type:

SequentialAgent

classmethod create_graded_hyde_workflow(documents, llm_config=None, enable_search_tools=False, **kwargs)¶

Create HyDE → Comprehensive Grading → Answer workflow.

Parameters:

documents (list[langchain_core.documents.Document])
llm_config (haive.core.models.llm.base.LLMConfig | None)
enable_search_tools (bool)

Return type:

SequentialAgent

classmethod create_modular_rag_workflow(documents, components, llm_config=None, **kwargs)¶

Create custom workflow from list of components.

This is the most generic method - just pass the components you want!

Parameters:

documents (list[langchain_core.documents.Document])
components (list[RAGComponent])
llm_config (haive.core.models.llm.base.LLMConfig | None)

Return type:

SequentialAgent

create_workflow(pattern, components=None, routing_conditions=None, **kwargs)¶

Create a workflow based on pattern and components.

Parameters:

pattern (WorkflowPattern) – Workflow pattern to use
components (list[RAGComponent] | None) – Optional component overrides
routing_conditions (dict[str, collections.abc.Callable] | None) – Conditional routing logic
**kwargs – Additional configuration

Returns:

Configured multi-agent workflow

Return type:

SequentialAgent | ConditionalAgent | ParallelAgent

replace_agent_in_workflow(workflow, target_agent_name, replacement_component, **kwargs)¶

Replace an agent in existing workflow based on I/O compatibility.

Uses the enhanced multi-agent base compatibility checking.

Parameters:

workflow (SequentialAgent | ConditionalAgent) – Existing workflow to modify
target_agent_name (str) – Name of agent to replace
replacement_component (RAGComponent) – New component to use
**kwargs – Configuration for new component

Returns:

True if replacement successful

Return type:

bool

suggest_workflow_optimizations(workflow)¶

Suggest optimizations for workflow based on I/O compatibility.

Parameters:: workflow (SequentialAgent | ConditionalAgent) – Workflow to optimize
Returns:: Optimization suggestions
Return type:: dict[str, Any]

class agents.rag.factories.compatible_rag_factory.RAGComponent¶

Bases: str, enum.Enum

Available RAG components for composition.

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

class agents.rag.factories.compatible_rag_factory.WorkflowPattern¶

Bases: str, enum.Enum

Common workflow patterns from the architecture guide.

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

agents.rag.factories.compatible_rag_factory.create_plug_and_play_component(component_type, documents, llm_config=None, **kwargs)¶

Create any RAG component as a standalone agent.

This function allows creating any component independently for plug-and-play usage.

Parameters:

component_type (RAGComponent) – Type of component to create
documents (list[langchain_core.documents.Document]) – Documents for components that need them
llm_config (haive.core.models.llm.base.LLMConfig | None) – LLM configuration
**kwargs – Component-specific arguments

Returns:

Standalone agent that can be plugged into any workflow

Return type:

haive.agents.base.agent.Agent

Examples

# Create standalone components decomposer = create_plug_and_play_component(

RAGComponent.ADAPTIVE_DECOMPOSITION, docs

) hallucination_grader = create_plug_and_play_component(

RAGComponent.ADVANCED_HALLUCINATION_GRADING, docs

)

# Use with any workflow workflow = SequentialAgent(agents=[decomposer, retriever, hallucination_grader])

agents.rag.factories.compatible_rag_factory.get_component_compatibility_info(component_type)¶

Get I/O schema information for a component type.

Parameters:: component_type (RAGComponent) – Component to get info for
Returns:: Dict with ‘inputs’ and ‘outputs’ lists
Return type:: dict[str, list[str]]