Function calling : RAG avec actions

Le function calling permet au LLM d'appeler des fonctions externes de maniere structuree. Combine avec le RAG, cela cree des assistants qui peuvent chercher dans vos documents ET executer des actions concretes dans vos systemes.

Prerequis : Consultez notre guide sur l'orchestration d'agents RAG pour comprendre les fondamentaux.

Pourquoi combiner RAG et Function Calling ?

Du passif a l'actif

Le RAG classique est en lecture seule : il recherche et repond. Avec le function calling, l'assistant peut agir sur vos systemes.

EVOLUTION DES CAPACITES

RAG Classique          RAG + Function Calling
────────────           ──────────────────────
Question → Reponse     Question → Reponse + Actions

"Quelle est la        "Quelle est la politique de remboursement?"
politique?"           → Recherche la politique
                      → Detecte une situation de remboursement
                      → Propose de creer un ticket
                      → Execute le remboursement si approuve

Comparaison des approches

Cas d'usage

• Support client : Recherche FAQ + creation de tickets + suivi de commande
• E-commerce : Info produit + ajout au panier + verification stock
• RH : Politique interne + demande de conges + consultation solde
• IT : Documentation + creation incident + status systeme
• Finance : Procedures + soumission notes de frais + approbation

Implementation de base

Definition des fonctions (OpenAI)

DEVELOPERpython
from openai import OpenAI
import json

client = OpenAI()

# Definition des outils disponibles
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_knowledge_base",
            "description": "Recherche des documents dans la base de connaissances. Utilise cette fonction pour trouver des informations.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "La requete de recherche semantique"
                    },
                    "top_k": {
                        "type": "integer",
                        "description": "Nombre de resultats a retourner (defaut: 5)",
                        "default": 5
                    },
                    "filters": {
                        "type": "object",
                        "description": "Filtres optionnels (category, date_after, etc.)",
                        "properties": {
                            "category": {"type": "string"},
                            "date_after": {"type": "string", "format": "date"}
                        }
                    }
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "create_support_ticket",
            "description": "Cree un ticket de support. Utilise cette fonction quand l'utilisateur a un probleme qui necessite un suivi.",
            "parameters": {
                "type": "object",
                "properties": {
                    "title": {
                        "type": "string",
                        "description": "Titre court et descriptif du ticket"
                    },
                    "description": {
                        "type": "string",
                        "description": "Description detaillee du probleme"
                    },
                    "priority": {
                        "type": "string",
                        "enum": ["low", "medium", "high", "urgent"],
                        "description": "Niveau de priorite"
                    },
                    "category": {
                        "type": "string",
                        "enum": ["technical", "billing", "account", "product", "other"],
                        "description": "Categorie du ticket"
                    }
                },
                "required": ["title", "description", "priority"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "check_order_status",
            "description": "Verifie le statut d'une commande. Utilise cette fonction quand l'utilisateur demande des infos sur sa commande.",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "Identifiant de la commande"
                    },
                    "email": {
                        "type": "string",
                        "description": "Email associe a la commande (optionnel)",
                        "format": "email"
                    }
                },
                "required": ["order_id"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "initiate_refund",
            "description": "Initie une demande de remboursement. Utilise uniquement si l'utilisateur a explicitement demande un remboursement.",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "Identifiant de la commande a rembourser"
                    },
                    "reason": {
                        "type": "string",
                        "description": "Raison du remboursement"
                    },
                    "amount": {
                        "type": "number",
                        "description": "Montant a rembourser (optionnel, defaut: total commande)"
                    }
                },
                "required": ["order_id", "reason"]
            }
        }
    }
]

Implementation des fonctions

DEVELOPERpython
from typing import Dict, Any, Optional
import json

class RAGWithActions:
    """RAG avec capacite d'action via function calling."""

    def __init__(self, vector_store, ticket_system, order_system):
        self.vector_store = vector_store
        self.ticket_system = ticket_system
        self.order_system = order_system

        # Map des fonctions disponibles
        self.function_map = {
            "search_knowledge_base": self._search_kb,
            "create_support_ticket": self._create_ticket,
            "check_order_status": self._check_order,
            "initiate_refund": self._initiate_refund,
        }

    def _search_kb(self, query: str, top_k: int = 5, filters: Dict = None) -> str:
        """Recherche dans la base de connaissances."""
        results = self.vector_store.similarity_search(
            query,
            k=top_k,
            filter=filters
        )

        if not results:
            return json.dumps({
                "status": "no_results",
                "message": "Aucun document pertinent trouve"
            })

        documents = []
        for i, doc in enumerate(results):
            documents.append({
                "id": f"DOC_{i+1}",
                "content": doc.page_content[:500],
                "source": doc.metadata.get("source", "unknown"),
                "relevance_score": doc.metadata.get("score", 0)
            })

        return json.dumps({
            "status": "success",
            "count": len(documents),
            "documents": documents
        }, ensure_ascii=False)

    def _create_ticket(
        self,
        title: str,
        description: str,
        priority: str,
        category: str = "other"
    ) -> str:
        """Cree un ticket de support."""
        try:
            ticket = self.ticket_system.create(
                title=title,
                description=description,
                priority=priority,
                category=category
            )
            return json.dumps({
                "status": "success",
                "ticket_id": ticket.id,
                "message": f"Ticket #{ticket.id} cree avec succes"
            })
        except Exception as e:
            return json.dumps({
                "status": "error",
                "message": str(e)
            })

    def _check_order(self, order_id: str, email: str = None) -> str:
        """Verifie le statut d'une commande."""
        try:
            order = self.order_system.get_order(order_id, email=email)
            if not order:
                return json.dumps({
                    "status": "not_found",
                    "message": f"Commande {order_id} non trouvee"
                })

            return json.dumps({
                "status": "success",
                "order": {
                    "id": order.id,
                    "status": order.status,
                    "created_at": order.created_at.isoformat(),
                    "total": order.total,
                    "items_count": len(order.items),
                    "tracking_number": order.tracking_number,
                    "estimated_delivery": order.estimated_delivery
                }
            })
        except Exception as e:
            return json.dumps({
                "status": "error",
                "message": str(e)
            })

    def _initiate_refund(
        self,
        order_id: str,
        reason: str,
        amount: float = None
    ) -> str:
        """Initie un remboursement."""
        try:
            refund = self.order_system.request_refund(
                order_id=order_id,
                reason=reason,
                amount=amount
            )
            return json.dumps({
                "status": "success",
                "refund_id": refund.id,
                "amount": refund.amount,
                "message": f"Demande de remboursement #{refund.id} initiee"
            })
        except Exception as e:
            return json.dumps({
                "status": "error",
                "message": str(e)
            })

    def execute_function(self, name: str, arguments: Dict[str, Any]) -> str:
        """Execute une fonction par son nom."""
        if name not in self.function_map:
            return json.dumps({
                "status": "error",
                "message": f"Fonction inconnue: {name}"
            })

        func = self.function_map[name]
        return func(**arguments)

Boucle de conversation

DEVELOPERpython
def chat_with_actions(
    rag: RAGWithActions,
    user_message: str,
    conversation_history: list = None,
    max_iterations: int = 5
) -> Dict[str, Any]:
    """
    Gere une conversation avec fonction calling.

    Args:
        rag: Instance RAGWithActions
        user_message: Message de l'utilisateur
        conversation_history: Historique de la conversation
        max_iterations: Nombre max d'iterations de function calling

    Returns:
        Dict avec la reponse et les actions executees
    """
    messages = conversation_history or []
    messages.append({"role": "user", "content": user_message})

    actions_executed = []
    iteration = 0

    while iteration < max_iterations:
        iteration += 1

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

        assistant_message = response.choices[0].message

        # Si pas de function call, on a la reponse finale
        if not assistant_message.tool_calls:
            messages.append(assistant_message)
            return {
                "response": assistant_message.content,
                "actions_executed": actions_executed,
                "conversation_history": messages
            }

        # Executer les function calls
        messages.append(assistant_message)

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

            # Executer la fonction
            result = rag.execute_function(function_name, arguments)

            # Enregistrer l'action
            actions_executed.append({
                "function": function_name,
                "arguments": arguments,
                "result": json.loads(result)
            })

            # Ajouter le resultat a la conversation
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })

    # Limite d'iterations atteinte
    return {
        "response": "La requete a necessite trop d'operations. Veuillez reformuler.",
        "actions_executed": actions_executed,
        "conversation_history": messages,
        "error": "max_iterations_reached"
    }

Patterns avances

Pattern 1 : Confirmation avant action

Demander confirmation pour les actions sensibles :

DEVELOPERpython
class ConfirmableRAG(RAGWithActions):
    """RAG avec confirmation pour les actions sensibles."""

    # Actions necessitant confirmation
    SENSITIVE_ACTIONS = {"initiate_refund", "delete_account", "cancel_order"}

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

    def execute_function(self, name: str, arguments: Dict[str, Any]) -> str:
        """Execute avec confirmation pour les actions sensibles."""

        if name in self.SENSITIVE_ACTIONS:
            # Generer un ID de confirmation
            confirmation_id = str(uuid.uuid4())[:8]

            # Stocker l'action en attente
            self.pending_actions[confirmation_id] = {
                "function": name,
                "arguments": arguments,
                "created_at": datetime.now()
            }

            return json.dumps({
                "status": "confirmation_required",
                "confirmation_id": confirmation_id,
                "action": name,
                "arguments": arguments,
                "message": f"Action sensible detectee. Confirmez avec l'ID: {confirmation_id}"
            })

        return super().execute_function(name, arguments)

    def confirm_action(self, confirmation_id: str) -> str:
        """Confirme et execute une action en attente."""

        if confirmation_id not in self.pending_actions:
            return json.dumps({
                "status": "error",
                "message": "ID de confirmation invalide ou expire"
            })

        action = self.pending_actions.pop(confirmation_id)

        # Verifier l'expiration (5 minutes)
        if datetime.now() - action["created_at"] > timedelta(minutes=5):
            return json.dumps({
                "status": "error",
                "message": "Confirmation expiree. Veuillez recommencer."
            })

        # Executer l'action
        return super().execute_function(
            action["function"],
            action["arguments"]
        )

Pattern 2 : Actions composees

Chainer plusieurs actions pour des workflows complexes :

DEVELOPERpython
class WorkflowRAG(RAGWithActions):
    """RAG avec workflows composes."""

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

        # Definir les workflows disponibles
        self.workflows = {
            "full_refund_process": self._workflow_full_refund,
            "escalate_to_human": self._workflow_escalate,
        }

        # Ajouter les workflows comme fonctions
        self.function_map.update({
            f"workflow_{name}": func
            for name, func in self.workflows.items()
        })

    async def _workflow_full_refund(
        self,
        order_id: str,
        reason: str
    ) -> str:
        """Workflow complet de remboursement."""
        results = []

        # 1. Verifier la commande
        order_result = json.loads(self._check_order(order_id))
        results.append({"step": "check_order", "result": order_result})

        if order_result["status"] != "success":
            return json.dumps({
                "status": "error",
                "step": "check_order",
                "message": "Commande non trouvee"
            })

        # 2. Verifier la politique de remboursement
        policy_result = json.loads(self._search_kb(
            "politique remboursement conditions",
            top_k=2
        ))
        results.append({"step": "check_policy", "result": policy_result})

        # 3. Creer le ticket de suivi
        ticket_result = json.loads(self._create_ticket(
            title=f"Remboursement commande {order_id}",
            description=f"Raison: {reason}",
            priority="high",
            category="billing"
        ))
        results.append({"step": "create_ticket", "result": ticket_result})

        # 4. Initier le remboursement
        refund_result = json.loads(self._initiate_refund(
            order_id=order_id,
            reason=reason
        ))
        results.append({"step": "initiate_refund", "result": refund_result})

        return json.dumps({
            "status": "success",
            "workflow": "full_refund_process",
            "steps_completed": len(results),
            "results": results
        })

    async def _workflow_escalate(
        self,
        conversation_summary: str,
        priority: str = "high"
    ) -> str:
        """Workflow d'escalade vers un humain."""
        results = []

        # 1. Creer le ticket d'escalade
        ticket_result = json.loads(self._create_ticket(
            title="Escalade requise",
            description=conversation_summary,
            priority=priority,
            category="other"
        ))
        results.append({"step": "create_ticket", "result": ticket_result})

        # 2. Notifier l'equipe (simulation)
        notification = {
            "status": "success",
            "message": "Equipe support notifiee"
        }
        results.append({"step": "notify_team", "result": notification})

        return json.dumps({
            "status": "success",
            "workflow": "escalate_to_human",
            "steps_completed": len(results),
            "results": results,
            "message": "Conversation escaladee. Un agent vous contactera sous peu."
        })

Pattern 3 : Function calling parallele

Executer plusieurs fonctions en parallele :

DEVELOPERpython
import asyncio
from concurrent.futures import ThreadPoolExecutor

class ParallelRAG(RAGWithActions):
    """RAG avec execution parallele des fonctions."""

    def __init__(self, *args, max_workers: int = 4, **kwargs):
        super().__init__(*args, **kwargs)
        self.executor = ThreadPoolExecutor(max_workers=max_workers)

    async def execute_functions_parallel(
        self,
        function_calls: list
    ) -> list:
        """Execute plusieurs fonctions en parallele."""

        async def execute_one(call):
            loop = asyncio.get_event_loop()
            result = await loop.run_in_executor(
                self.executor,
                self.execute_function,
                call["name"],
                call["arguments"]
            )
            return {
                "function": call["name"],
                "result": json.loads(result)
            }

        tasks = [execute_one(call) for call in function_calls]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        return [
            r if not isinstance(r, Exception) else {"error": str(r)}
            for r in results
        ]


async def chat_with_parallel_actions(
    rag: ParallelRAG,
    user_message: str
) -> Dict:
    """Conversation avec fonction calling parallele."""

    messages = [{"role": "user", "content": user_message}]

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=tools,
        tool_choice="auto",
        parallel_tool_calls=True  # Activer le parallelisme
    )

    assistant_message = response.choices[0].message

    if assistant_message.tool_calls:
        # Preparer les appels
        calls = [
            {
                "name": tc.function.name,
                "arguments": json.loads(tc.function.arguments),
                "id": tc.id
            }
            for tc in assistant_message.tool_calls
        ]

        # Executer en parallele
        results = await rag.execute_functions_parallel(calls)

        # Continuer la conversation
        messages.append(assistant_message)
        for call, result in zip(calls, results):
            messages.append({
                "role": "tool",
                "tool_call_id": call["id"],
                "content": json.dumps(result)
            })

        # Reponse finale
        final_response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages
        )

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

    return {"response": assistant_message.content, "actions": []}

Securite et validation

Validation des entrees

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

class SearchInput(BaseModel):
    """Schema de validation pour la recherche."""
    query: str = Field(..., min_length=1, max_length=500)
    top_k: int = Field(default=5, ge=1, le=20)
    filters: Optional[dict] = None

    @validator('query')
    def sanitize_query(cls, v):
        # Supprimer les caracteres dangereux
        return v.replace('\n', ' ').strip()


class TicketInput(BaseModel):
    """Schema de validation pour les tickets."""
    title: str = Field(..., min_length=5, max_length=200)
    description: str = Field(..., min_length=10, max_length=5000)
    priority: Literal["low", "medium", "high", "urgent"]
    category: Literal["technical", "billing", "account", "product", "other"] = "other"


class RefundInput(BaseModel):
    """Schema de validation pour les remboursements."""
    order_id: str = Field(..., pattern=r'^ORD-[0-9]{8}$')
    reason: str = Field(..., min_length=10, max_length=1000)
    amount: Optional[float] = Field(default=None, ge=0)


def validated_execute(func, input_class: type, **kwargs):
    """Execute une fonction avec validation Pydantic."""
    try:
        validated = input_class(**kwargs)
        return func(**validated.dict())
    except ValidationError as e:
        return json.dumps({
            "status": "validation_error",
            "errors": e.errors()
        })

Rate limiting et quotas

DEVELOPERpython
from datetime import datetime, timedelta
from collections import defaultdict

class RateLimitedRAG(RAGWithActions):
    """RAG avec rate limiting par fonction."""

    # Limites par fonction (appels par minute)
    RATE_LIMITS = {
        "search_knowledge_base": 30,
        "create_support_ticket": 5,
        "check_order_status": 10,
        "initiate_refund": 2,
    }

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.call_history = defaultdict(list)

    def _check_rate_limit(self, function_name: str, user_id: str) -> bool:
        """Verifie si l'utilisateur a depasse la limite."""
        limit = self.RATE_LIMITS.get(function_name, 10)
        key = f"{user_id}:{function_name}"

        now = datetime.now()
        cutoff = now - timedelta(minutes=1)

        # Nettoyer les anciennes entrees
        self.call_history[key] = [
            t for t in self.call_history[key]
            if t > cutoff
        ]

        return len(self.call_history[key]) < limit

    def execute_function_with_limit(
        self,
        name: str,
        arguments: Dict[str, Any],
        user_id: str
    ) -> str:
        """Execute avec rate limiting."""

        if not self._check_rate_limit(name, user_id):
            return json.dumps({
                "status": "rate_limited",
                "message": f"Trop d'appels a {name}. Reessayez dans 1 minute."
            })

        # Enregistrer l'appel
        key = f"{user_id}:{name}"
        self.call_history[key].append(datetime.now())

        return self.execute_function(name, arguments)

Monitoring et observabilite

DEVELOPERpython
import logging
from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Dict, Any

@dataclass
class FunctionCallTrace:
    """Trace d'un appel de fonction."""
    function_name: str
    arguments: Dict[str, Any]
    result: Dict[str, Any]
    start_time: datetime
    end_time: datetime
    success: bool
    error: str = None

    @property
    def duration_ms(self) -> float:
        return (self.end_time - self.start_time).total_seconds() * 1000


class TracedRAG(RAGWithActions):
    """RAG avec tracing complet."""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.traces: List[FunctionCallTrace] = []
        self.logger = logging.getLogger("rag_function_calls")

    def execute_function(self, name: str, arguments: Dict[str, Any]) -> str:
        """Execute avec tracing."""
        start_time = datetime.now()
        error = None
        success = True

        try:
            result = super().execute_function(name, arguments)
            result_parsed = json.loads(result)
        except Exception as e:
            error = str(e)
            success = False
            result_parsed = {"error": error}
            result = json.dumps(result_parsed)

        end_time = datetime.now()

        # Creer la trace
        trace = FunctionCallTrace(
            function_name=name,
            arguments=arguments,
            result=result_parsed,
            start_time=start_time,
            end_time=end_time,
            success=success,
            error=error
        )
        self.traces.append(trace)

        # Logger
        self.logger.info(
            f"Function call: {name} | "
            f"Duration: {trace.duration_ms:.2f}ms | "
            f"Success: {success}"
        )

        return result

    def get_metrics(self) -> Dict[str, Any]:
        """Retourne les metriques de performance."""
        if not self.traces:
            return {}

        by_function = defaultdict(list)
        for trace in self.traces:
            by_function[trace.function_name].append(trace)

        metrics = {}
        for func_name, traces in by_function.items():
            durations = [t.duration_ms for t in traces]
            success_count = sum(1 for t in traces if t.success)

            metrics[func_name] = {
                "total_calls": len(traces),
                "success_rate": success_count / len(traces),
                "avg_duration_ms": sum(durations) / len(durations),
                "max_duration_ms": max(durations),
                "min_duration_ms": min(durations),
            }

        return metrics

Couts et performance

Bonnes pratiques

1. Descriptions de fonctions precises

DEVELOPERpython
# BON - Description claire avec contexte d'utilisation
{
    "name": "search_knowledge_base",
    "description": """Recherche des documents dans la base de connaissances.
    Utilise cette fonction pour:
    - Repondre aux questions factuelles
    - Trouver des procedures ou politiques
    - Verifier des informations
    NE PAS utiliser pour les questions sur les commandes (utiliser check_order_status)."""
}

# MAUVAIS - Description vague
{
    "name": "search",
    "description": "Cherche des trucs"
}

2. Gestion des erreurs gracieuse

DEVELOPERpython
# Toujours retourner un JSON structure, meme en cas d'erreur
try:
    result = do_action()
    return json.dumps({"status": "success", "data": result})
except ValidationError as e:
    return json.dumps({"status": "validation_error", "errors": e.errors()})
except PermissionError:
    return json.dumps({"status": "unauthorized", "message": "Action non autorisee"})
except Exception as e:
    return json.dumps({"status": "error", "message": str(e)})

3. Limiter le nombre d'iterations

DEVELOPERpython
# Toujours definir une limite d'iterations
MAX_FUNCTION_CALLS = 5

for i in range(MAX_FUNCTION_CALLS):
    response = call_llm()
    if not response.tool_calls:
        break
else:
    return "Limite d'operations atteinte"

Conclusion

Le function calling transforme le RAG d'un systeme de questions-reponses en un assistant capable d'agir. Les patterns de confirmation, workflows composes et execution parallele permettent de construire des assistants sophistiques et securises.

Points cles :

• Definissez des fonctions avec des descriptions precises
• Implementez la confirmation pour les actions sensibles
• Utilisez le rate limiting et la validation
• Tracez tous les appels pour le monitoring

Pour aller plus loin

• LangGraph : Workflows RAG - Orchestration complexe
• AutoGen : Multi-agents - Agents conversationnels
• CrewAI : Equipes d'agents - Collaboration d'agents

Besoin d'un RAG actionnable ? Ailog combine recherche documentaire et function calling pour des assistants IA qui agissent. Configuration simple, securite integree.