RAG Agent

RAG Agent

This example demonstrates how to build a RAG Agent - an autonomous agent that can search a knowledge base and use other tools to answer complex questions. Unlike a simple RAG pipeline, a RAG agent can decide when to search, what to search for, and can combine multiple searches with other tools.

Why RAG Agents?

graph TD
    A[Question] --> B[Agent]
    B --> C{Need to Search?}
    C -->|Yes| D[search_documents]
    D --> E[Results]
    E --> B
    C -->|No| F{Use Calculator?}
    F -->|Yes| G[calculate]
    G --> B
    F -->|No| H[Final Answer]

Traditional RAG pipelines always retrieve documents, even for simple questions. A RAG agent is smarter:

• Decides IF retrieval is needed
• Can reformulate queries for better search results
• Can perform multiple searches and combine results
• Can use other tools (calculator, web search, etc.) alongside retrieval

Building a RAG Agent

Important Tool Constraints:

• No Optional Parameters: All tool parameters must be required. OpenAI and other LLM providers require all parameters in their JSON schemas.

• Complete Docstring Required: Every parameter must be documented in the Args: section. The Tool extracts descriptions to build the JSON schema sent to the LLM. Missing descriptions raise a ValueError.

# Define the search tool
@synalinks.saving.register_synalinks_serializable()
async def search_knowledge_base(query: str):
    """Search the knowledge base for documents relevant to the query.

    Use this tool to find information about company policies, procedures,
    products, or any documented knowledge.

    Args:
        query (str): The search query describing what information you need.
    """
    results = await knowledge_base.hybrid_search(query, k=3)
    return {"documents": results}

# Create the agent
inputs = synalinks.Input(data_model=synalinks.ChatMessages)
outputs = await synalinks.FunctionCallingAgent(
    tools=[synalinks.Tool(search_knowledge_base)],
    language_model=language_model,
    autonomous=True,
    max_iterations=5,
)(inputs)

Key Takeaways

• Autonomous Decision Making: The agent decides when to search and what queries to use.
• Multi-Tool Support: Combine document search with other tools like calculators or APIs.
• Iterative Reasoning: The agent can search multiple times and refine its understanding.
• Conversational: Maintains context across multiple turns.

Program Visualization

API References

• FunctionCallingAgent
• KnowledgeBase
• ChatMessages (Base DataModels)
• EmbeddingModel
• Program

Document

Bases: DataModel

A document stored in the knowledge base.

Source code in examples/14_rag_agent.py

class Document(synalinks.DataModel):
    """A document stored in the knowledge base."""

    id: str = synalinks.Field(
        description="Unique document identifier",
    )
    title: str = synalinks.Field(
        description="Document title",
    )
    content: str = synalinks.Field(
        description="The main text content of the document",
    )
    category: str = synalinks.Field(
        description="Document category",
    )

calculate(expression) async

Perform mathematical calculations.

Use this for any math operations like addition, multiplication, percentages, or complex expressions.

Parameters:

Name	Type	Description	Default
expression	str	A mathematical expression to evaluate, e.g., '100 * 0.15' or '(50 + 30) / 2'.	required

Source code in examples/14_rag_agent.py

@synalinks.saving.register_synalinks_serializable()
async def calculate(expression: str):
    """Perform mathematical calculations.

    Use this for any math operations like addition, multiplication,
    percentages, or complex expressions.

    Args:
        expression (str): A mathematical expression to evaluate,
            e.g., '100 * 0.15' or '(50 + 30) / 2'.
    """
    # Validate expression
    allowed_chars = "0123456789+-*/().% "
    if not all(c in allowed_chars for c in expression):
        return {
            "status": "error",
            "message": "Invalid characters in expression",
            "result": None,
        }

    try:
        # Handle percentage
        expr = expression.replace("%", "/100")
        result = eval(expr, {"__builtins__": {}}, {})
        return {
            "status": "success",
            "expression": expression,
            "result": round(float(result), 2),
        }
    except Exception as e:
        return {
            "status": "error",
            "message": str(e),
            "result": None,
        }

get_current_date() async

Get the current date and time.

Use this when you need to know today's date or the current time. This function takes no arguments.

Source code in examples/14_rag_agent.py

@synalinks.saving.register_synalinks_serializable()
async def get_current_date():
    """Get the current date and time.

    Use this when you need to know today's date or the current time.
    This function takes no arguments.
    """
    from datetime import datetime

    now = datetime.now()
    return {
        "date": now.strftime("%Y-%m-%d"),
        "time": now.strftime("%H:%M:%S"),
        "day_of_week": now.strftime("%A"),
    }

search_knowledge_base(query) async

Search the knowledge base for documents relevant to the query.

Use this tool to find information about company policies, procedures, products, or any documented knowledge. Returns the most relevant documents.

Parameters:

Name	Type	Description	Default
query	str	The search query describing what information you need. Be specific and use relevant keywords.	required

Source code in examples/14_rag_agent.py

@synalinks.saving.register_synalinks_serializable()
async def search_knowledge_base(query: str):
    """Search the knowledge base for documents relevant to the query.

    Use this tool to find information about company policies, procedures,
    products, or any documented knowledge. Returns the most relevant documents.

    Args:
        query (str): The search query describing what information you need.
            Be specific and use relevant keywords.
    """
    global _knowledge_base

    if _knowledge_base is None:
        return {"error": "Knowledge base not initialized"}

    results = await _knowledge_base.hybrid_search(query, k=30)
    return {"documents": results}