Function calling : RAG mit Aktionen
Vollständiger Leitfaden zum Kombinieren von RAG und function calling: Agenten, die recherchieren UND handeln, Integration externer APIs, automatisierte Aktionen und interaktive Workflows.
Function calling : RAG mit Aktionen
Function Calling ermöglicht es dem LLM, externe Funktionen strukturiert aufzurufen. In Kombination mit RAG entstehen so Assistenten, die in Ihren Dokumenten recherchieren UND konkrete Aktionen in Ihren Systemen ausführen können.
Voraussetzungen : Lesen Sie unseren Leitfaden zur Orchestrierung von RAG-Agenten, um die Grundlagen zu verstehen.
Warum RAG und Function Calling kombinieren?
Vom Passiven zum Aktiven
Klassisches RAG ist rein lesend: Es recherchiert und antwortet. Mit Function Calling kann der Assistent auf Ihren Systemen aktiv werden.
ENTWICKLUNG DER FÄHIGKEITEN
Klassisches RAG RAG + Function Calling
──────────── ──────────────────────
Frage → Antwort Frage → Antwort + Aktionen
"Wie lautet die "Wie lautet die Rückerstattungsrichtlinie?"
Richtlinie?" → Sucht die Richtlinie
→ Erkennt eine Rückerstattungssituation
→ Schlägt vor, ein Ticket zu erstellen
→ Führt die Rückerstattung bei Genehmigung aus
Vergleich der Ansätze
| Aspekt | Nur RAG | RAG + Function Calling |
|---|---|---|
| Modus | Nur lesend | Lesen + Schreiben |
| Antwort | Informativ | Informativ + Aktionen |
| Interaktion | Frage/Antwort | Interaktiver Workflow |
| Integration | Wissensdatenbank | KB + APIs + Systeme |
| Mehrwert | Information | Automatisierung |
Anwendungsfälle
- Kundensupport: FAQ-Suche + Ticket-Erstellung + Bestellverfolgung
- E-Commerce: Produktinfo + Warenkorb-Hinzufügung + Bestandsprüfung
- HR: Interne Richtlinie + Urlaubsantrag + Saldoabfrage
- IT: Dokumentation + Vorfallerstellung + Systemstatus
- Finanzen: Verfahren + Spesenabrechnung + Genehmigung
Grundimplementierung
Definition der Funktionen (OpenAI)
DEVELOPERpythonfrom openai import OpenAI import json client = OpenAI() # Definition der verfügbaren Tools tools = [ { "type": "function", "function": { "name": "search_knowledge_base", "description": "Durchsucht Dokumente in der Wissensdatenbank. Diese Funktion nutzen, um Informationen zu finden.", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Die semantische Suchanfrage" }, "top_k": { "type": "integer", "description": "Anzahl der zurückzugebenden Ergebnisse (Standard: 5)", "default": 5 }, "filters": { "type": "object", "description": "Optionale Filter (category, date_after usw.)", "properties": { "category": {"type": "string"}, "date_after": {"type": "string", "format": "date"} } } }, "required": ["query"] } } }, { "type": "function", "function": { "name": "create_support_ticket", "description": "Erstellt ein Support-Ticket. Diese Funktion nutzen, wenn der Nutzer ein Problem hat, das nachverfolgt werden muss.", "parameters": { "type": "object", "properties": { "title": { "type": "string", "description": "Kurzer, aussagekräftiger Titel des Tickets" }, "description": { "type": "string", "description": "Detaillierte Beschreibung des Problems" }, "priority": { "type": "string", "enum": ["low", "medium", "high", "urgent"], "description": "Prioritätsstufe" }, "category": { "type": "string", "enum": ["technical", "billing", "account", "product", "other"], "description": "Kategorie des Tickets" } }, "required": ["title", "description", "priority"] } } }, { "type": "function", "function": { "name": "check_order_status", "description": "Prüft den Status einer Bestellung. Diese Funktion nutzen, wenn der Nutzer Informationen zu seiner Bestellung anfragt.", "parameters": { "type": "object", "properties": { "order_id": { "type": "string", "description": "Bestell-ID" }, "email": { "type": "string", "description": "Mit der Bestellung verknüpfte E-Mail (optional)", "format": "email" } }, "required": ["order_id"] } } }, { "type": "function", "function": { "name": "initiate_refund", "description": "Leitet eine Rückerstattungsanfrage ein. Nur verwenden, wenn der Nutzer ausdrücklich eine Rückerstattung angefragt hat.", "parameters": { "type": "object", "properties": { "order_id": { "type": "string", "description": "ID der zu erstattenden Bestellung" }, "reason": { "type": "string", "description": "Grund der Rückerstattung" }, "amount": { "type": "number", "description": "Zu erstattender Betrag (optional, Standard: Gesamtbetrag der Bestellung)" } }, "required": ["order_id", "reason"] } } } ]
Implementierung der Funktionen
DEVELOPERpythonfrom typing import Dict, Any, Optional import json class RAGWithActions: """RAG mit Aktionsfähigkeit 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: """Durchsucht die Wissensdatenbank.""" results = self.vector_store.similarity_search( query, k=top_k, filter=filters ) if not results: return json.dumps({ "status": "no_results", "message": "Kein relevantes Dokument gefunden" }) 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: """Erstellt ein Support-Ticket.""" 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} erfolgreich erstellt" }) except Exception as e: return json.dumps({ "status": "error", "message": str(e) }) def _check_order(self, order_id: str, email: str = None) -> str: """Prüft den Status einer Bestellung.""" try: order = self.order_system.get_order(order_id, email=email) if not order: return json.dumps({ "status": "not_found", "message": f"Bestellung {order_id} nicht gefunden" }) 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: """Leitet eine Rückerstattung ein.""" 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"Rückerstattungsanfrage #{refund.id} eingeleitet" }) except Exception as e: return json.dumps({ "status": "error", "message": str(e) }) def execute_function(self, name: str, arguments: Dict[str, Any]) -> str: """Führt eine Funktion anhand ihres Namens aus.""" if name not in self.function_map: return json.dumps({ "status": "error", "message": f"Unbekannte Funktion: {name}" }) func = self.function_map[name] return func(**arguments)
Konversationsschleife
DEVELOPERpythondef chat_with_actions( rag: RAGWithActions, user_message: str, conversation_history: list = None, max_iterations: int = 5 ) -> Dict[str, Any]: """ Verwaltet eine Konversation mit Function Calling. Args: rag: RAGWithActions-Instanz user_message: Nachricht des Nutzers conversation_history: Konversationsverlauf max_iterations: Maximale Anzahl an Function-Calling-Iterationen Returns: Dict mit der Antwort und den ausgeführten Aktionen """ 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": "Die Anfrage erforderte zu viele Vorgänge. Bitte formulieren Sie sie um.", "actions_executed": actions_executed, "conversation_history": messages, "error": "max_iterations_reached" }
Fortgeschrittene Patterns
Pattern 1: Bestätigung vor Aktion
Bestätigung für sensible Aktionen anfordern:
DEVELOPERpythonclass ConfirmableRAG(RAGWithActions): """RAG mit Bestätigung für sensible Aktionen.""" # 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: """Führt mit Bestätigung für sensible Aktionen aus.""" 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"Sensible Aktion erkannt. Bestätigen Sie mit der ID: {confirmation_id}" }) return super().execute_function(name, arguments) def confirm_action(self, confirmation_id: str) -> str: """Bestätigt und führt eine ausstehende Aktion aus.""" if confirmation_id not in self.pending_actions: return json.dumps({ "status": "error", "message": "Ungültige oder abgelaufene Bestätigungs-ID" }) 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": "Bestätigung abgelaufen. Bitte beginnen Sie erneut." }) # Aktion ausführen return super().execute_function( action["function"], action["arguments"] )
Pattern 2: Zusammengesetzte Aktionen
Mehrere Aktionen für komplexe Workflows verketten:
DEVELOPERpythonclass WorkflowRAG(RAGWithActions): """RAG mit zusammengesetzten Workflows.""" 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: """Vollständiger Rückerstattungs-Workflow.""" 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": "Bestellung nicht gefunden" }) # 2. Rückerstattungsrichtlinie prüfen policy_result = json.loads(self._search_kb( "Rückerstattungsrichtlinie Bedingungen", top_k=2 )) results.append({"step": "check_policy", "result": policy_result}) # 3. Folgeticket erstellen ticket_result = json.loads(self._create_ticket( title=f"Rückerstattung Bestellung {order_id}", description=f"Grund: {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: """Eskalations-Workflow an einen Menschen.""" results = [] # 1. Eskalationsticket erstellen ticket_result = json.loads(self._create_ticket( title="Eskalation erforderlich", description=conversation_summary, priority=priority, category="other" )) results.append({"step": "create_ticket", "result": ticket_result}) # 2. Team benachrichtigen (Simulation) notification = { "status": "success", "message": "Support-Team benachrichtigt" } results.append({"step": "notify_team", "result": notification}) return json.dumps({ "status": "success", "workflow": "escalate_to_human", "steps_completed": len(results), "results": results, "message": "Konversation eskaliert. Ein Mitarbeiter wird sich in Kürze bei Ihnen melden." })
Pattern 3: Paralleles Function Calling
Mehrere Funktionen parallel ausführen:
DEVELOPERpythonimport asyncio from concurrent.futures import ThreadPoolExecutor class ParallelRAG(RAGWithActions): """RAG mit paralleler Ausführung der Funktionen.""" 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: """Führt mehrere Funktionen parallel aus.""" 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: """Konversation mit parallelem Function Calling.""" 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
DEVELOPERpythonfrom pydantic import BaseModel, Field, validator from typing import Optional, Literal class SearchInput(BaseModel): """Validierungsschema für die Suche.""" 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): """Validierungsschema für 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): """Validierungsschema für Rückerstattungen.""" 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): """Führt eine Funktion mit Pydantic-Validierung aus.""" 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
DEVELOPERpythonfrom datetime import datetime, timedelta from collections import defaultdict class RateLimitedRAG(RAGWithActions): """RAG mit Rate Limiting pro Funktion.""" # 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: """Prüft, ob der Nutzer das Limit überschritten hat.""" 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: """Führt mit Rate Limiting aus.""" if not self._check_rate_limit(name, user_id): return json.dumps({ "status": "rate_limited", "message": f"Zu viele Aufrufe von {name}. Versuchen Sie es in 1 Minute erneut." }) # Den Aufruf speichern key = f"{user_id}:{name}" self.call_history[key].append(datetime.now()) return self.execute_function(name, arguments)
Monitoring und Observability
DEVELOPERpythonimport logging from dataclasses import dataclass, field from datetime import datetime from typing import List, Dict, Any @dataclass class FunctionCallTrace: """Trace eines Funktionsaufrufs.""" 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 mit vollständigem Tracing.""" 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: """Führt mit Tracing aus.""" 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]: """Gibt die Performance-Metriken zurück.""" 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
| Aktion | Geschätzte Tokens | Geschätzte Kosten | Latenz |
|---|---|---|---|
| Nur Suche | ~1500 | ~$0.015 | 1-2s |
| Suche + 1 Aktion | ~2500 | ~$0.025 | 2-4s |
| Suche + 3 Aktionen | ~4000 | ~$0.04 | 4-8s |
| Vollständiger Workflow (5 Aktionen) | ~6000 | ~$0.06 | 8-15s |
Bewährte Praktiken
1. Präzise Funktionsbeschreibungen
DEVELOPERpython# GUT - Klare Beschreibung mit Nutzungskontext { "name": "search_knowledge_base", "description": """Durchsucht Dokumente in der Wissensdatenbank. Diese Funktion nutzen für: - Beantwortung von Sachfragen - Auffinden von Verfahren oder Richtlinien - Überprüfung von Informationen NICHT verwenden für Fragen zu Bestellungen (dafür check_order_status verwenden).""" } # SCHLECHT - Vage Beschreibung { "name": "search", "description": "Sucht irgendwas" }
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": "Aktion nicht autorisiert"}) 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 "Limit an Vorgängen erreicht"
Fazit
Function Calling verwandelt RAG von einem reinen Frage-Antwort-System in einen handlungsfähigen Assistenten. Die Patterns Bestätigung, zusammengesetzte Workflows und parallele Ausführung ermöglichen den Aufbau anspruchsvoller und sicherer Assistenten.
Hauptpunkte:
- Definieren Sie Funktionen mit präzisen Beschreibungen
- Implementieren Sie eine Bestätigung für sensible Aktionen
- Nutzen Sie Rate Limiting und Validierung
- Verfolgen Sie alle Aufrufe für das Monitoring
FAQ
Weiterführende Links
- LangGraph: RAG-Workflows - Komplexe Orchestrierung
- AutoGen: Multi-Agenten - Konversationelle Agenten
- CrewAI: Agenten-Teams - Zusammenarbeit von Agenten
Benötigen Sie ein RAG, das Aktionen ausführen kann? Ailog kombiniert Dokumentensuche und Function Calling für KI-Assistenten, die handeln. Einfache Einrichtung, integrierte Sicherheit.
Tags
Verwandte Artikel
RAG-Agenten: Orchestrierung von Multi-Agenten-Systemen
Konzipieren Sie RAG-basierte Multi-Agenten-Systeme: Orchestrierung, Spezialisierung, Zusammenarbeit und Fehlerbehandlung für komplexe Assistenten.
Agentic RAG 2025: Aufbau autonomer KI-Agenten (Kompletter Leitfaden)
Kompletter Agentic RAG-Leitfaden: Architektur, Design Patterns, autonome Agenten mit dynamischem Retrieval, Multi-Tool-Orchestrierung. Mit Beispielen LangGraph und CrewAI.
Konversationelles RAG: Gedächtnis und Kontext über mehrere Sitzungen
Implementieren Sie ein RAG mit konversationellem Gedächtnis: Verwaltung des Kontexts, Verlauf über mehrere Sitzungen und Personalisierung der Antworten.