Skip to main content

Agent Graphs

LangGraph-based agent state machines and workflows for building sophisticated RAG agents with multi-turn capabilities, safety policies, and cost tracking.

Core Classes​

AgentState​

Description: TypedDict defining the state structure for RAG agents

Fields:

  • messages (List[BaseMessage]): Chat messages in the conversation
  • query (str): Current user query
  • retrieved_docs (List[Dict]): Documents retrieved from vector store
  • reranked_docs (List[Dict]): Documents after reranking
  • plan (str, optional): Agent's execution plan
  • action (str, optional): Current action being performed
  • answer (str, optional): Final answer to user
  • error (str, optional): Error message if any
  • metadata (Dict): Additional metadata
  • step_count (int): Current step number
  • max_steps (int): Maximum allowed steps
  • cost_tracker (Dict): Cost tracking information
  • latency_tracker (Dict): Latency tracking information
  • awaiting_clarification (bool): Whether agent is waiting for clarification
  • clarification_slots (List[SlotDefinition]): Required clarification slots
  • filled_slots (Dict): Filled clarification slots
  • clarification_rounds (int): Number of clarification rounds

RAGAgentGraph​

Description: Main LangGraph-based RAG agent implementation

Parameters:

  • config (AgentConfig): Agent configuration
  • llm (ChatOpenAI): Language model instance
  • tools (List[BaseTool]): Available tools
  • safety_policy (SafetyPolicy, optional): Safety policy
  • tool_policy (ToolPolicy, optional): Tool usage policy
  • middleware (List, optional): Middleware components

Returns: Configured LangGraph StateGraph

Example:

from recoagent.agents import RAGAgentGraph, AgentConfig
from langchain_openai import ChatOpenAI

# Configure agent
config = AgentConfig(
    max_steps=5,
    enable_clarification=True,
    safety_checks=True
)

# Initialize LLM
llm = ChatOpenAI(model="gpt-4", temperature=0.1)

# Create agent graph
agent_graph = RAGAgentGraph(
    config=config,
    llm=llm,
    tools=[retrieval_tool, web_search_tool],
    safety_policy=safety_policy
)

# Run agent
result = agent_graph.invoke({"query": "What is machine learning?"})

AgentConfig​

Description: Configuration class for RAG agents

Parameters:

  • max_steps (int): Maximum execution steps
  • enable_clarification (bool): Enable multi-turn clarification
  • safety_checks (bool): Enable safety policy checks
  • cost_threshold (float): Maximum cost per query
  • latency_threshold (float): Maximum latency in seconds
  • retrieval_config (Dict): Retrieval configuration
  • reranking_config (Dict): Reranking configuration

Usage Examples​

Basic Agent Setup​

from recoagent.agents import RAGAgentGraph, AgentConfig
from recoagent.agents.tools import RetrievalTool
from langchain_openai import ChatOpenAI

# Create configuration
config = AgentConfig(
    max_steps=3,
    enable_clarification=False,
    safety_checks=True
)

# Initialize components
llm = ChatOpenAI(model="gpt-3.5-turbo")
retrieval_tool = RetrievalTool(vector_store=vector_store)

# Create agent
agent = RAGAgentGraph(
    config=config,
    llm=llm,
    tools=[retrieval_tool]
)

# Run query
result = agent.invoke({
    "query": "Explain quantum computing",
    "messages": []
})

Multi-Turn Clarification Agent​

from recoagent.agents import RAGAgentGraph, AgentConfig
from recoagent.agents.multi_turn_agent import SlotDefinition

# Configure for clarification
config = AgentConfig(
    max_steps=5,
    enable_clarification=True,
    safety_checks=True
)

# Define clarification slots
slots = [
    SlotDefinition(
        name="domain",
        description="What domain are you interested in?",
        required=True,
        options=["technology", "science", "business"]
    ),
    SlotDefinition(
        name="complexity",
        description="What complexity level?",
        required=True,
        options=["beginner", "intermediate", "advanced"]
    )
]

# Create agent with clarification
agent = RAGAgentGraph(
    config=config,
    llm=llm,
    tools=tools,
    clarification_slots=slots
)

# Run with clarification
result = agent.invoke({
    "query": "I need help with something",
    "messages": []
})

Advanced Agent with Middleware​

from recoagent.agents import RAGAgentGraph, AgentConfig
from recoagent.agents.middleware import GuardrailsMiddleware, CostTrackingMiddleware
from recoagent.agents.policies import SafetyPolicy

# Create middleware
guardrails = GuardrailsMiddleware()
cost_tracker = CostTrackingMiddleware(max_cost=0.10)

# Create safety policy
safety_policy = SafetyPolicy(
    blocked_topics=["harmful", "illegal"],
    max_response_length=1000
)

# Configure agent
config = AgentConfig(
    max_steps=5,
    safety_checks=True,
    cost_threshold=0.10
)

# Create agent with middleware
agent = RAGAgentGraph(
    config=config,
    llm=llm,
    tools=tools,
    safety_policy=safety_policy,
    middleware=[guardrails, cost_tracker]
)

API Reference​

AgentState Methods​

The AgentState is a TypedDict and doesn't have methods, but provides type safety for state management.

RAGAgentGraph Methods​

invoke(state: Dict) -> Dict​

Execute the agent with given state

Parameters:

  • state (Dict): Initial agent state

Returns: Final agent state with answer

stream(state: Dict) -> Iterator[Dict]​

Stream agent execution steps

Parameters:

  • state (Dict): Initial agent state

Returns: Iterator of intermediate states

AgentConfig Methods​

validate() -> bool​

Validate configuration parameters

Returns: True if valid, raises exception if invalid

See Also​