agents.rag.query_planning.agent¶

Query Planning Agentic RAG Agent.

from typing import Any Implementation of query planning RAG with structured decomposition and execution. Provides intelligent query analysis, planning, and multi-stage retrieval strategies.

Classes¶

`QueryComplexity`	Query complexity levels.
`QueryPlan`	Complete query execution plan.
`QueryPlanningRAGAgent`	Query Planning RAG agent with structured decomposition and execution.
`QueryPlanningResult`	Complete result from query planning execution.
`QueryType`	Types of queries for planning.
`SubQuery`	Individual sub-query in a decomposed plan.
`SubQueryResult`	Result from executing a sub-query.

Functions¶

`create_query_planning_rag_agent`(documents[, ...])	Create a Query Planning RAG agent.
`get_query_planning_rag_io_schema`()	Get I/O schema for Query Planning RAG agents.

Module Contents¶

class agents.rag.query_planning.agent.QueryComplexity¶

Bases: str, enum.Enum

Query complexity levels.

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

class agents.rag.query_planning.agent.QueryPlan(/, **data)¶

Bases: pydantic.BaseModel

Complete query execution plan.

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.query_planning.agent.QueryPlanningRAGAgent¶

Bases: haive.agents.base.agent.Agent

Query Planning RAG agent with structured decomposition and execution.

This agent uses conditional edges to execute sub-queries in a planned order.

build_graph()¶

Build the query planning graph with conditional edges.

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

create_query_plan(state)¶

Create a query execution plan.

Parameters:: state (dict[str, Any])
Return type:: dict[str, Any]

execute_sub_query(state)¶

Execute the current sub-query.

Parameters:: state (dict[str, Any])
Return type:: dict[str, Any]

classmethod from_documents(documents, llm_config=None, planning_depth=3, **kwargs)¶

Create Query Planning RAG agent from documents.

Parameters:

documents (list[langchain_core.documents.Document]) – Documents to index for retrieval
llm_config (haive.core.models.llm.base.LLMConfig | None) – LLM configuration
planning_depth (int) – Maximum depth of query decomposition
**kwargs – Additional arguments

Returns:

QueryPlanningRAGAgent instance

setup_agent()¶

Initialize engines.

Return type:: None

should_continue_execution(state)¶

Determine if more sub-queries should be executed.

Parameters:: state (dict[str, Any])
Return type:: str

synthesize_results(state)¶

Synthesize sub-query results into final answer.

Parameters:: state (dict[str, Any])
Return type:: dict[str, Any]

class agents.rag.query_planning.agent.QueryPlanningResult(/, **data)¶

Bases: pydantic.BaseModel

Complete result from query planning execution.

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.query_planning.agent.QueryType¶

Bases: str, enum.Enum

Types of queries for planning.

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

class agents.rag.query_planning.agent.SubQuery(/, **data)¶

Bases: pydantic.BaseModel

Individual sub-query in a decomposed plan.

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.query_planning.agent.SubQueryResult(/, **data)¶

Bases: pydantic.BaseModel

Result from executing a sub-query.

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)

agents.rag.query_planning.agent.create_query_planning_rag_agent(documents, llm_config=None, planning_mode='comprehensive', **kwargs)¶

Create a Query Planning RAG agent.

Parameters:

documents (list[langchain_core.documents.Document]) – Documents for retrieval
llm_config (haive.core.models.llm.base.LLMConfig | None) – LLM configuration
planning_mode (str) – Planning mode (“simple”, “moderate”, “comprehensive”)
**kwargs – Additional arguments

Returns:

Configured Query Planning RAG agent

Return type:

QueryPlanningRAGAgent

agents.rag.query_planning.agent.get_query_planning_rag_io_schema()¶

Get I/O schema for Query Planning RAG agents.

Return type:: dict[str, list[str]]