Function calling : RAG mit Aktionen

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.

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

Pourquoi combiner RAG et Function Calling ?

Vom Passiven zum Aktiven

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

Anwendungsfälle

• 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

Grundimplementierung

Definition der Funktionen (OpenAI)

DEVELOPERpython
from openai import OpenAI
import json

client = OpenAI()

# Definition der verfügbaren Tools
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

        # Mapping der verfügbaren Funktionen
        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

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

        assistant_message = response.choices[0].message

        # Wenn kein function call, ist dies die endgültige Antwort
        if not assistant_message.tool_calls:
            messages.append(assistant_message)
            return {
                "response": assistant_message.content,
                "actions_executed": actions_executed,
                "conversation_history": messages
            }

        # Führe die function calls aus
        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)

            # Führ die Funktion aus
            result = rag.execute_function(function_name, arguments)

            # Aktion protokollieren
            actions_executed.append({
                "function": function_name,
                "arguments": arguments,
                "result": json.loads(result)
            })

            # Ergebnis zur Konversation hinzufügen
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })

    # Max. Iterationen erreicht
    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."""

    # Aktionen, die une Bestätigung erfordern
    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:
            # Bestätigungs-ID generieren
            confirmation_id = str(uuid.uuid4())[:8]

            # Aktion in Wartestellung speichern
            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)

        # Ablauf prüfen (5 Minuten)
        if datetime.now() - action["created_at"] > timedelta(minutes=5):
            return json.dumps({
                "status": "error",
                "message": "Confirmation expiree. Veuillez recommencer."
            })

        # Aktion ausführen
        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)

        # Verfügbare Workflows definieren
        self.workflows = {
            "full_refund_process": self._workflow_full_refund,
            "escalate_to_human": self._workflow_escalate,
        }

        # Workflows als Funktionen hinzufügen
        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. Bestellung prüfen
        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. Rückerstattungsrichtlinie prüfen
        policy_result = json.loads(self._search_kb(
            "politique remboursement conditions",
            top_k=2
        ))
        results.append({"step": "check_policy", "result": policy_result})

        # 3. Folgeticket erstellen
        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. Rückerstattung einleiten
        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. Eskalationsticket erstellen
        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. Team benachrichtigen (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  # Parallelität aktivieren
    )

    assistant_message = response.choices[0].message

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

        # Parallel ausführen
        results = await rag.execute_functions_parallel(calls)

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

        # Endgültige Antwort
        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": []}

Sicherheit und Validierung

Validierung der Eingaben

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):
        # Gefährliche Zeichen entfernen
        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 und Quotas

DEVELOPERpython
from datetime import datetime, timedelta
from collections import defaultdict

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

    # Limits pro Funktion (Aufrufe pro 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)

        # Alte Einträge bereinigen
        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."
            })

        # Den Aufruf speichern
        key = f"{user_id}:{name}"
        self.call_history[key].append(datetime.now())

        return self.execute_function(name, arguments)

Monitoring und Observability

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()

        # Trace erstellen
        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)

        # Loggen
        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

Kosten und Performance

Bewährte Praktiken

1. Präzise Funktionsbeschreibungen

DEVELOPERpython
# GUT - Klare Beschreibung mit Nutzungskontext
{
    "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)."""
}

# SCHLECHT - Vage Beschreibung
{
    "name": "search",
    "description": "Cherche des trucs"
}

2. Robuste Fehlerbehandlung

DEVELOPERpython
# Immer eine strukturierte JSON-Antwort zurückgeben, auch im Fehlerfall
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. Anzahl der Iterationen begrenzen

DEVELOPERpython
# Immer ein Limit für Iterationen definieren
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"

Fazit

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.

Hauptpunkte :

• 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

Weiterführende Links

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

Benötigen Sie ein RAG, das Aktionen ausführen kann ? Ailog combine recherche documentaire et function calling pour des assistants IA qui agissent. Configuration simple, securite integree.