Function Calling: RAG with Actions

Function calling transforms RAG from a passive query system into an active agent capable of interacting with the real world. Your assistant no longer just answers questions: it acts.

Why Function Calling + RAG?

Traditional RAG:
User --> Query --> Search --> Response
         (read-only, passive)

RAG + Function Calling:
User --> Query --> Search --> Analysis --> Actions --> Response
         (read + write, proactive)

Concrete comparison

Architecture

+-------------------------------------------------------------+
|                    RAG + Actions System                      |
+-------------------------------------------------------------+
|                                                              |
|  +----------+    +----------+    +----------------------+   |
|  |  User    |--->|   LLM    |--->|  Function Router     |   |
|  |  Query   |    | (GPT-4)  |    |                      |   |
|  +----------+    +----------+    +----------------------+   |
|                       |                    |                 |
|                       v                    v                 |
|              +----------------------------------------+     |
|              |           Available Tools              |     |
|              +------------+-----------+---------------+     |
|              |   RAG      |  Actions  |   External    |     |
|              |  Search    |  (CRUD)   |   APIs        |     |
|              +------------+-----------+---------------+     |
|              | - search   | - create  | - send_email  |     |
|              | - filter   | - update  | - call_api    |     |
|              | - retrieve | - delete  | - webhook     |     |
|              +------------+-----------+---------------+     |
|                                                              |
+-------------------------------------------------------------+

Complete Implementation

1. Function Definitions (OpenAI Format)

DEVELOPERpython
from typing import Any
import json
from openai import OpenAI
from pydantic import BaseModel, Field

# Tool definitions for OpenAI
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "search_knowledge_base",
            "description": "Search the knowledge base to find relevant information. Use for any question about products, policies, or documentation.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Natural language search query"
                    },
                    "category": {
                        "type": "string",
                        "enum": ["products", "policies", "faq", "technical"],
                        "description": "Category to filter results"
                    },
                    "top_k": {
                        "type": "integer",
                        "default": 5,
                        "description": "Number of results to return"
                    }
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "create_support_ticket",
            "description": "Create a support ticket when the user has an unresolved issue or needs human assistance.",
            "parameters": {
                "type": "object",
                "properties": {
                    "title": {
                        "type": "string",
                        "description": "Brief description of the issue"
                    },
                    "description": {
                        "type": "string",
                        "description": "Detailed explanation of the problem"
                    },
                    "priority": {
                        "type": "string",
                        "enum": ["low", "medium", "high", "urgent"],
                        "description": "Ticket priority level"
                    },
                    "category": {
                        "type": "string",
                        "enum": ["billing", "technical", "product", "other"]
                    }
                },
                "required": ["title", "description"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "update_user_preferences",
            "description": "Update user account preferences or settings.",
            "parameters": {
                "type": "object",
                "properties": {
                    "preference_key": {
                        "type": "string",
                        "description": "The preference to update"
                    },
                    "new_value": {
                        "type": "string",
                        "description": "The new value for the preference"
                    }
                },
                "required": ["preference_key", "new_value"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "schedule_callback",
            "description": "Schedule a phone callback from customer support.",
            "parameters": {
                "type": "object",
                "properties": {
                    "phone_number": {
                        "type": "string",
                        "description": "Phone number to call"
                    },
                    "preferred_time": {
                        "type": "string",
                        "description": "Preferred callback time (ISO 8601)"
                    },
                    "reason": {
                        "type": "string",
                        "description": "Reason for the callback"
                    }
                },
                "required": ["phone_number", "reason"]
            }
        }
    }
]

2. Function Implementations

DEVELOPERpython
from datetime import datetime
from qdrant_client import QdrantClient
from sentence_transformers import SentenceTransformer

class FunctionImplementations:
    """Concrete implementations of available functions."""

    def __init__(self):
        self.qdrant = QdrantClient(host="localhost", port=6333)
        self.encoder = SentenceTransformer("all-MiniLM-L6-v2")
        self.collection_name = "knowledge_base"

    def search_knowledge_base(
        self,
        query: str,
        category: str = None,
        top_k: int = 5
    ) -> dict:
        """
        Search the vector knowledge base.
        Returns relevant documents with sources.
        """
        # Encode query
        query_vector = self.encoder.encode(query).tolist()

        # Build filter if category specified
        search_filter = None
        if category:
            search_filter = {
                "must": [
                    {"key": "category", "match": {"value": category}}
                ]
            }

        # Vector search
        results = self.qdrant.search(
            collection_name=self.collection_name,
            query_vector=query_vector,
            query_filter=search_filter,
            limit=top_k
        )

        # Format results
        documents = []
        for hit in results:
            documents.append({
                "content": hit.payload.get("content", ""),
                "source": hit.payload.get("source", "Unknown"),
                "category": hit.payload.get("category", ""),
                "relevance_score": round(hit.score, 3)
            })

        return {
            "success": True,
            "query": query,
            "results_count": len(documents),
            "documents": documents
        }

    def create_support_ticket(
        self,
        title: str,
        description: str,
        priority: str = "medium",
        category: str = "other"
    ) -> dict:
        """
        Create a support ticket in the system.
        Returns ticket ID and confirmation.
        """
        # Generate ticket ID
        ticket_id = f"TKT-{datetime.now().strftime('%Y%m%d%H%M%S')}"

        # In production: save to database
        ticket_data = {
            "id": ticket_id,
            "title": title,
            "description": description,
            "priority": priority,
            "category": category,
            "status": "open",
            "created_at": datetime.now().isoformat()
        }

        # Simulate database save
        # db.tickets.insert_one(ticket_data)

        return {
            "success": True,
            "ticket_id": ticket_id,
            "message": f"Ticket {ticket_id} created successfully",
            "priority": priority,
            "estimated_response": self._get_response_time(priority)
        }

    def update_user_preferences(
        self,
        preference_key: str,
        new_value: str,
        user_id: str = None
    ) -> dict:
        """Update user preferences."""
        # Validate preference key
        valid_preferences = [
            "language", "timezone", "notifications",
            "email_frequency", "theme"
        ]

        if preference_key not in valid_preferences:
            return {
                "success": False,
                "error": f"Invalid preference. Valid options: {valid_preferences}"
            }

        # In production: update database
        return {
            "success": True,
            "preference": preference_key,
            "old_value": "previous_value",  # Would fetch from DB
            "new_value": new_value,
            "message": f"Preference '{preference_key}' updated to '{new_value}'"
        }

    def schedule_callback(
        self,
        phone_number: str,
        reason: str,
        preferred_time: str = None
    ) -> dict:
        """Schedule a support callback."""
        # Validate phone number format
        import re
        if not re.match(r'^\+?[\d\s-]{10,}$', phone_number):
            return {
                "success": False,
                "error": "Invalid phone number format"
            }

        callback_id = f"CB-{datetime.now().strftime('%Y%m%d%H%M%S')}"

        return {
            "success": True,
            "callback_id": callback_id,
            "phone_number": phone_number,
            "scheduled_time": preferred_time or "Within 2 business hours",
            "message": "Callback scheduled. Our team will contact you soon."
        }

    def _get_response_time(self, priority: str) -> str:
        """Get estimated response time based on priority."""
        response_times = {
            "urgent": "1 hour",
            "high": "4 hours",
            "medium": "24 hours",
            "low": "48 hours"
        }
        return response_times.get(priority, "24 hours")

3. RAG with Actions Engine

DEVELOPERpython
class RAGWithActions:
    """
    Main engine combining RAG search with actionable functions.
    Manages the conversation loop with automatic tool execution.
    """

    def __init__(self, api_key: str):
        self.client = OpenAI(api_key=api_key)
        self.functions = FunctionImplementations()
        self.tools = TOOLS

        # Map function names to implementations
        self.function_map = {
            "search_knowledge_base": self.functions.search_knowledge_base,
            "create_support_ticket": self.functions.create_support_ticket,
            "update_user_preferences": self.functions.update_user_preferences,
            "schedule_callback": self.functions.schedule_callback
        }

        self.system_prompt = """You are a helpful customer support assistant with access to a knowledge base and action capabilities.

Your capabilities:
1. Search the knowledge base for product info, policies, and FAQs
2. Create support tickets for unresolved issues
3. Update user preferences and settings
4. Schedule phone callbacks

Guidelines:
- Always search the knowledge base first before creating tickets
- Only create tickets if you cannot resolve the issue
- Ask for confirmation before taking irreversible actions
- Be concise but thorough in your responses
- Cite sources when providing information from the knowledge base"""

    def process_message(
        self,
        user_message: str,
        conversation_history: list = None
    ) -> dict:
        """
        Process a user message, executing any required tool calls.
        Returns the final response and action log.
        """
        messages = conversation_history or []
        messages.append({"role": "user", "content": user_message})

        # Add system prompt if first message
        if len(messages) == 1:
            messages.insert(0, {"role": "system", "content": self.system_prompt})

        action_log = []
        max_iterations = 5  # Prevent infinite loops
        iteration = 0

        while iteration < max_iterations:
            iteration += 1

            # Call LLM
            response = self.client.chat.completions.create(
                model="gpt-4o",
                messages=messages,
                tools=self.tools,
                tool_choice="auto"
            )

            assistant_message = response.choices[0].message

            # Check if tool calls are needed
            if not assistant_message.tool_calls:
                # No more tools to call, return final response
                return {
                    "response": assistant_message.content,
                    "actions_taken": action_log,
                    "conversation": messages + [
                        {"role": "assistant", "content": assistant_message.content}
                    ]
                }

            # Process each tool call
            messages.append(assistant_message)

            for tool_call in assistant_message.tool_calls:
                function_name = tool_call.function.name
                function_args = json.loads(tool_call.function.arguments)

                # Execute function
                if function_name in self.function_map:
                    result = self.function_map[function_name](**function_args)
                else:
                    result = {"error": f"Unknown function: {function_name}"}

                # Log action
                action_log.append({
                    "function": function_name,
                    "arguments": function_args,
                    "result": result
                })

                # Add result to messages
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps(result)
                })

        # Max iterations reached
        return {
            "response": "I apologize, but I encountered an issue processing your request. Please try again.",
            "actions_taken": action_log,
            "error": "Max iterations reached"
        }

4. Usage Example

DEVELOPERpython
# Initialize
rag_actions = RAGWithActions(api_key="sk-...")

# Example 1: Simple search
result = rag_actions.process_message(
    "What is your return policy for electronics?"
)
print(result["response"])
# Output: "Based on our policy documentation, electronics can be returned
# within 30 days of purchase with original packaging..."
print(result["actions_taken"])
# Output: [{"function": "search_knowledge_base", "arguments": {...}, "result": {...}}]

# Example 2: Search + Action
result = rag_actions.process_message(
    "I ordered a laptop 3 days ago and it's defective. I need help."
)
print(result["response"])
# Output: "I've searched our return policy and you're eligible for a return.
# I've also created support ticket TKT-20260127143022 with high priority.
# Our team will contact you within 4 hours to arrange the return."
print(result["actions_taken"])
# Output: [
#   {"function": "search_knowledge_base", ...},
#   {"function": "create_support_ticket", "result": {"ticket_id": "TKT-..."}}
# ]

# Example 3: Multi-turn conversation
conversation = []
result = rag_actions.process_message(
    "I want to change my notification settings",
    conversation_history=conversation
)
# Assistant: "What would you like to change your notification settings to?"
conversation = result["conversation"]

result = rag_actions.process_message(
    "Set it to weekly emails only",
    conversation_history=conversation
)
# Assistant: "Done! I've updated your email_frequency preference to 'weekly'."

Advanced Patterns

Pattern 1: Confirmation Before Action

DEVELOPERpython
TOOLS_WITH_CONFIRMATION = [
    {
        "type": "function",
        "function": {
            "name": "request_action_confirmation",
            "description": "Request user confirmation before executing a sensitive action.",
            "parameters": {
                "type": "object",
                "properties": {
                    "action_type": {"type": "string"},
                    "action_description": {"type": "string"},
                    "consequences": {"type": "string"}
                },
                "required": ["action_type", "action_description"]
            }
        }
    }
]

class RAGWithConfirmation(RAGWithActions):
    """RAG system that asks for confirmation before sensitive actions."""

    SENSITIVE_ACTIONS = ["create_support_ticket", "schedule_callback", "delete_account"]

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.pending_actions = {}

    def process_message(self, user_message: str, conversation_history: list = None):
        # Check for confirmation responses
        if self._is_confirmation(user_message):
            return self._execute_pending_action(user_message, conversation_history)

        # Normal processing with confirmation injection
        return super().process_message(user_message, conversation_history)

    def _is_confirmation(self, message: str) -> bool:
        confirmations = ["yes", "confirm", "ok", "proceed", "do it"]
        return message.lower().strip() in confirmations

    def _execute_pending_action(self, confirmation: str, history: list):
        if not self.pending_actions:
            return {"response": "No pending action to confirm.", "actions_taken": []}

        action = self.pending_actions.pop("current")
        result = self.function_map[action["function"]](**action["arguments"])

        return {
            "response": f"Action completed: {result.get('message', 'Success')}",
            "actions_taken": [{"function": action["function"], "result": result}]
        }

Pattern 2: Conditional Workflows

DEVELOPERpython
class WorkflowEngine:
    """
    Execute conditional workflows based on RAG results and actions.
    """

    def __init__(self, rag_system: RAGWithActions):
        self.rag = rag_system
        self.workflows = {}

    def register_workflow(self, trigger: str, workflow: callable):
        """Register a workflow triggered by specific conditions."""
        self.workflows[trigger] = workflow

    def process_with_workflows(self, user_message: str) -> dict:
        """Process message and trigger applicable workflows."""
        # Initial RAG processing
        result = self.rag.process_message(user_message)

        # Check for workflow triggers
        for action in result.get("actions_taken", []):
            function_name = action.get("function")
            action_result = action.get("result", {})

            # Trigger workflow if conditions met
            if function_name == "search_knowledge_base":
                if action_result.get("results_count", 0) == 0:
                    # No results found - trigger escalation
                    self._trigger_workflow("no_results", result, user_message)

            elif function_name == "create_support_ticket":
                if action_result.get("priority") == "urgent":
                    # Urgent ticket - trigger notification
                    self._trigger_workflow("urgent_ticket", result, action_result)

        return result

    def _trigger_workflow(self, trigger: str, context: dict, data: any):
        """Execute a registered workflow."""
        if trigger in self.workflows:
            self.workflows[trigger](context, data)

# Example workflows
def escalation_workflow(context: dict, query: str):
    """Escalate when no knowledge base results found."""
    print(f"ESCALATION: No results for query '{query}'")
    # Send alert to support team
    # Log for knowledge base improvement

def urgent_notification_workflow(context: dict, ticket: dict):
    """Send immediate notification for urgent tickets."""
    print(f"URGENT: Ticket {ticket['ticket_id']} needs immediate attention")
    # Send SMS/Slack/PagerDuty alert

# Usage
workflow_engine = WorkflowEngine(rag_actions)
workflow_engine.register_workflow("no_results", escalation_workflow)
workflow_engine.register_workflow("urgent_ticket", urgent_notification_workflow)

result = workflow_engine.process_with_workflows("My production system is down!")

Pattern 3: Parallel Tool Execution

DEVELOPERpython
import asyncio
from concurrent.futures import ThreadPoolExecutor

class AsyncRAGWithActions:
    """
    RAG system with parallel tool execution for performance.
    """

    def __init__(self, api_key: str, max_workers: int = 4):
        self.client = OpenAI(api_key=api_key)
        self.functions = FunctionImplementations()
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
        self.tools = TOOLS
        self.function_map = {
            "search_knowledge_base": self.functions.search_knowledge_base,
            "create_support_ticket": self.functions.create_support_ticket,
            "update_user_preferences": self.functions.update_user_preferences,
            "schedule_callback": self.functions.schedule_callback
        }

    async def process_message_async(self, user_message: str) -> dict:
        """Process with parallel tool execution."""
        messages = [
            {"role": "system", "content": "You are a helpful assistant..."},
            {"role": "user", "content": user_message}
        ]

        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=self.tools,
            tool_choice="auto"
        )

        assistant_message = response.choices[0].message

        if not assistant_message.tool_calls:
            return {"response": assistant_message.content, "actions_taken": []}

        # Execute all tool calls in parallel
        loop = asyncio.get_event_loop()
        tasks = []

        for tool_call in assistant_message.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)

            if func_name in self.function_map:
                task = loop.run_in_executor(
                    self.executor,
                    lambda f=func_name, a=func_args: self.function_map[f](**a)
                )
                tasks.append((tool_call.id, func_name, func_args, task))

        # Wait for all tasks to complete
        results = []
        tool_messages = []

        for call_id, func_name, func_args, task in tasks:
            result = await task
            results.append({
                "function": func_name,
                "arguments": func_args,
                "result": result
            })
            tool_messages.append({
                "role": "tool",
                "tool_call_id": call_id,
                "content": json.dumps(result)
            })

        # Get final response
        messages.append(assistant_message)
        messages.extend(tool_messages)

        final_response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=self.tools
        )

        return {
            "response": final_response.choices[0].message.content,
            "actions_taken": results
        }

# Usage
async def main():
    rag = AsyncRAGWithActions(api_key="sk-...")
    result = await rag.process_message_async(
        "Search for laptop specs and create a ticket for my order issue"
    )
    print(result)

asyncio.run(main())

Security and Validation

Input Validation with Pydantic

DEVELOPERpython
from pydantic import BaseModel, Field, validator
from typing import Optional, Literal
import re

class TicketInput(BaseModel):
    """Validated input for ticket creation."""

    title: str = Field(..., min_length=5, max_length=200)
    description: str = Field(..., min_length=20, max_length=5000)
    priority: Literal["low", "medium", "high", "urgent"] = "medium"
    category: Literal["billing", "technical", "product", "other"] = "other"

    @validator("title")
    def title_no_html(cls, v):
        if "<" in v or ">" in v:
            raise ValueError("HTML not allowed in title")
        return v.strip()

    @validator("description")
    def description_no_scripts(cls, v):
        if re.search(r"<script|javascript:", v, re.IGNORECASE):
            raise ValueError("Scripts not allowed in description")
        return v.strip()

class CallbackInput(BaseModel):
    """Validated input for callback scheduling."""

    phone_number: str = Field(..., pattern=r"^\+?[\d\s-]{10,15}$")
    reason: str = Field(..., min_length=10, max_length=500)
    preferred_time: Optional[str] = None

    @validator("phone_number")
    def clean_phone(cls, v):
        return re.sub(r"[\s-]", "", v)

class SecureFunctionImplementations(FunctionImplementations):
    """Function implementations with Pydantic validation."""

    def create_support_ticket(self, **kwargs) -> dict:
        try:
            validated = TicketInput(**kwargs)
            return super().create_support_ticket(
                title=validated.title,
                description=validated.description,
                priority=validated.priority,
                category=validated.category
            )
        except Exception as e:
            return {"success": False, "error": str(e)}

    def schedule_callback(self, **kwargs) -> dict:
        try:
            validated = CallbackInput(**kwargs)
            return super().schedule_callback(
                phone_number=validated.phone_number,
                reason=validated.reason,
                preferred_time=validated.preferred_time
            )
        except Exception as e:
            return {"success": False, "error": str(e)}

Rate Limiting

DEVELOPERpython
from datetime import datetime, timedelta
from collections import defaultdict
import threading

class RateLimiter:
    """Rate limiter for function calls per user."""

    def __init__(self):
        self.calls = defaultdict(list)
        self.limits = {
            "search_knowledge_base": (100, 60),   # 100 calls per 60 seconds
            "create_support_ticket": (5, 3600),   # 5 tickets per hour
            "schedule_callback": (3, 86400),      # 3 callbacks per day
            "update_user_preferences": (20, 3600) # 20 updates per hour
        }
        self.lock = threading.Lock()

    def check_rate_limit(self, user_id: str, function_name: str) -> bool:
        """Check if user is within rate limits."""
        if function_name not in self.limits:
            return True

        max_calls, window_seconds = self.limits[function_name]
        key = f"{user_id}:{function_name}"
        now = datetime.now()
        window_start = now - timedelta(seconds=window_seconds)

        with self.lock:
            # Clean old entries
            self.calls[key] = [t for t in self.calls[key] if t > window_start]

            # Check limit
            if len(self.calls[key]) >= max_calls:
                return False

            # Record call
            self.calls[key].append(now)
            return True

    def get_retry_after(self, user_id: str, function_name: str) -> int:
        """Get seconds until rate limit resets."""
        if function_name not in self.limits:
            return 0

        max_calls, window_seconds = self.limits[function_name]
        key = f"{user_id}:{function_name}"

        if not self.calls[key]:
            return 0

        oldest_call = min(self.calls[key])
        reset_time = oldest_call + timedelta(seconds=window_seconds)
        return max(0, int((reset_time - datetime.now()).total_seconds()))

class RateLimitedRAG(RAGWithActions):
    """RAG system with rate limiting."""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.rate_limiter = RateLimiter()

    def process_message(self, user_message: str, user_id: str = "anonymous", **kwargs):
        # Override function map with rate-limited versions
        original_map = self.function_map.copy()

        for func_name, func in original_map.items():
            self.function_map[func_name] = self._wrap_with_rate_limit(
                func, func_name, user_id
            )

        try:
            return super().process_message(user_message, **kwargs)
        finally:
            self.function_map = original_map

    def _wrap_with_rate_limit(self, func, func_name: str, user_id: str):
        def wrapper(*args, **kwargs):
            if not self.rate_limiter.check_rate_limit(user_id, func_name):
                retry_after = self.rate_limiter.get_retry_after(user_id, func_name)
                return {
                    "success": False,
                    "error": "Rate limit exceeded",
                    "retry_after_seconds": retry_after
                }
            return func(*args, **kwargs)
        return wrapper

Monitoring and Observability

Action Logging

DEVELOPERpython
import logging
from datetime import datetime
from typing import Optional
import json

class ActionLogger:
    """Structured logging for RAG actions."""

    def __init__(self, logger_name: str = "rag_actions"):
        self.logger = logging.getLogger(logger_name)
        self.logger.setLevel(logging.INFO)

        # JSON formatter for structured logs
        handler = logging.StreamHandler()
        handler.setFormatter(logging.Formatter('%(message)s'))
        self.logger.addHandler(handler)

    def log_action(
        self,
        user_id: str,
        session_id: str,
        function_name: str,
        arguments: dict,
        result: dict,
        duration_ms: float,
        success: bool
    ):
        """Log a function call with full context."""
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "event_type": "function_call",
            "user_id": user_id,
            "session_id": session_id,
            "function": function_name,
            "arguments": self._sanitize_args(arguments),
            "success": success,
            "duration_ms": round(duration_ms, 2),
            "result_summary": self._summarize_result(result)
        }

        self.logger.info(json.dumps(log_entry))

    def _sanitize_args(self, args: dict) -> dict:
        """Remove sensitive data from arguments."""
        sensitive_keys = ["phone_number", "email", "password", "api_key"]
        sanitized = {}

        for key, value in args.items():
            if key in sensitive_keys:
                sanitized[key] = "[REDACTED]"
            else:
                sanitized[key] = value

        return sanitized

    def _summarize_result(self, result: dict) -> dict:
        """Create a summary of the result."""
        return {
            "success": result.get("success", True),
            "has_error": "error" in result,
            "result_keys": list(result.keys())
        }

class MonitoredRAG(RAGWithActions):
    """RAG system with full monitoring."""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.action_logger = ActionLogger()
        self.metrics = {
            "total_calls": 0,
            "successful_calls": 0,
            "failed_calls": 0,
            "total_duration_ms": 0,
            "calls_by_function": defaultdict(int)
        }

    def process_message(
        self,
        user_message: str,
        user_id: str = "anonymous",
        session_id: str = None,
        **kwargs
    ):
        import time
        import uuid

        session_id = session_id or str(uuid.uuid4())

        # Wrap functions with monitoring
        original_map = self.function_map.copy()

        for func_name, func in original_map.items():
            self.function_map[func_name] = self._wrap_with_monitoring(
                func, func_name, user_id, session_id
            )

        try:
            return super().process_message(user_message, **kwargs)
        finally:
            self.function_map = original_map

    def _wrap_with_monitoring(
        self,
        func,
        func_name: str,
        user_id: str,
        session_id: str
    ):
        def wrapper(*args, **kwargs):
            import time
            start = time.time()

            try:
                result = func(*args, **kwargs)
                success = result.get("success", True) if isinstance(result, dict) else True
            except Exception as e:
                result = {"success": False, "error": str(e)}
                success = False

            duration_ms = (time.time() - start) * 1000

            # Log action
            self.action_logger.log_action(
                user_id=user_id,
                session_id=session_id,
                function_name=func_name,
                arguments=kwargs,
                result=result,
                duration_ms=duration_ms,
                success=success
            )

            # Update metrics
            self.metrics["total_calls"] += 1
            self.metrics["total_duration_ms"] += duration_ms
            self.metrics["calls_by_function"][func_name] += 1

            if success:
                self.metrics["successful_calls"] += 1
            else:
                self.metrics["failed_calls"] += 1

            return result

        return wrapper

    def get_metrics(self) -> dict:
        """Get current metrics."""
        total = self.metrics["total_calls"]
        return {
            **self.metrics,
            "success_rate": self.metrics["successful_calls"] / total if total > 0 else 0,
            "avg_duration_ms": self.metrics["total_duration_ms"] / total if total > 0 else 0
        }

E-commerce Use Case

DEVELOPERpython
# Complete e-commerce assistant example
ECOMMERCE_TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "search_products",
            "description": "Search product catalog",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "category": {"type": "string"},
                    "min_price": {"type": "number"},
                    "max_price": {"type": "number"},
                    "in_stock_only": {"type": "boolean", "default": True}
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "add_to_cart",
            "description": "Add a product to the shopping cart",
            "parameters": {
                "type": "object",
                "properties": {
                    "product_id": {"type": "string"},
                    "quantity": {"type": "integer", "default": 1}
                },
                "required": ["product_id"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "check_order_status",
            "description": "Check the status of an order",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {"type": "string"}
                },
                "required": ["order_id"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "initiate_return",
            "description": "Start a return process for an order",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {"type": "string"},
                    "reason": {"type": "string"},
                    "items": {
                        "type": "array",
                        "items": {"type": "string"}
                    }
                },
                "required": ["order_id", "reason"]
            }
        }
    }
]

# Example conversation flow
"""
User: "I'm looking for wireless headphones under $100"

Assistant calls: search_products(query="wireless headphones", max_price=100)
Result: [{"id": "HP-001", "name": "Sony WH-1000", "price": 89.99}, ...]

Assistant: "I found 3 wireless headphones under $100. The Sony WH-1000 at $89.99 has excellent reviews. Would you like me to add it to your cart?"

User: "Yes, add 2 of them"

Assistant calls: add_to_cart(product_id="HP-001", quantity=2)
Result: {"success": true, "cart_total": 179.98}

Assistant: "Done\! I added 2 Sony WH-1000 headphones to your cart. Your total is $179.98."
"""

Key Takeaways

Next Steps

Function calling + RAG opens up powerful possibilities:

1. Start simple: Search + one action (like ticket creation)
2. Add validation: Use Pydantic for all inputs
3. Add monitoring: Log all actions for debugging
4. Scale gradually: Add more tools as needed

Related Guides

• RAG Agents Orchestration
• LangGraph: RAG Workflows
• AutoGen: Multi-Agent Systems
• CrewAI: Specialized Agents

---

Need to deploy a RAG system with actions? Ailog provides a complete RAG-as-a-Service platform with built-in function calling, security, and monitoring. Deploy your intelligent assistant in minutes.