So erstellen Sie ein benutzerdefiniertes Agenten-Geschirr

So bauen Sie ein benutzerdefiniertes Agenten-Framework

Ein benutzerdefiniertes Agenten-Framework zu entwickeln, ist ein praktischer Ansatz, um KI-Agenten zu orchestrieren, ihre Werkzeuge zu verwalten und ihre Interaktionen zu steuern. Anders als vorgefertigte Frameworks gibt Ihnen ein eigenes Framework die volle Kontrolle über das Agentenverhalten, die Kontextverarbeitung und die Fehlerbehebung. Dieser Artikel führt Sie durch eine vollständige, praxisnahe Implementierung in Python, mit Fokus auf Klarheit und Erweiterbarkeit.

Voraussetzungen

Stellen Sie vor dem Start sicher, dass Ihre Umgebung diese Anforderungen erfüllt:

• Python 3.10 oder höher auf Ihrem System installiert
• pip-Paketmanager (in Python enthalten)
• Grundlegende Vertrautheit mit Kommandozeilen-Tools und Python-Syntax
• Ein API-Schlüssel für mindestens einen LLM-Anbieter (in den Beispielen wird OpenAI verwendet)

Folgende Python-Pakete werden benötigt:

• `openai` – für LLM-API-Aufrufe
• `httpx` – für HTTP-Anfragen an Werkzeuge
• `pydantic` – für Datenvalidierung und Konfiguration

Schritt-für-Schritt-Installation

1. Python-Virtual-Environment einrichten

Isolieren Sie Abhängigkeiten, um Konflikte mit anderen Projekten zu vermeiden.

python -m venv agent-harness
source agent-harness/bin/activate  # Unter Windows: agent-harness\Scripts\activate

2. Erforderliche Pakete installieren

Installieren Sie die Kernbibliotheken für API-Kommunikation und Datenmodellierung.

pip install openai httpx pydantic

3. Projektstruktur erstellen

Organisieren Sie Ihren Code in Module für bessere Wartbarkeit.

mkdir -p agent_harness/tools
touch agent_harness/__init__.py
touch agent_harness/core.py
touch agent_harness/tools/__init__.py
touch agent_harness/tools/search.py
touch agent_harness/tools/calculator.py
touch main.py

4. Umgebungsvariablen konfigurieren

Speichern Sie Ihre API-Schlüssel sicher. Erstellen Sie eine `.env`-Datei (nicht in der Versionskontrolle) oder exportieren Sie sie direkt.

export OPENAI_API_KEY="ihr-api-schluessel-hier"

Das Kern-Agenten-Framework erstellen

Die Agent-Klasse

Das Herzstück des Frameworks ist eine `Agent`-Klasse, die den LLM, die Werkzeuge und den Gesprächsverlauf verwaltet.

# agent_harness/core.py
import os
from typing import List, Dict, Any, Callable
from openai import OpenAI

class Tool:
    """Repräsentiert ein Werkzeug, das der Agent nutzen kann."""
    def __init__(self, name: str, description: str, function: Callable):
        self.name = name
        self.description = description
        self.function = function

class Agent:
    def __init__(self, model: str = "gpt-4", tools: List[Tool] = None):
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.model = model
        self.tools = tools or []
        self.history = []
        
    def add_tool(self, tool: Tool):
        """Ein neues Werkzeug registrieren."""
        self.tools.append(tool)
        
    def build_system_prompt(self) -> str:
        """System-Prompt mit verfügbaren Werkzeugen erstellen."""
        tool_descriptions = "\n".join(
            [f"- {t.name}: {t.description}" for t in self.tools]
        )
        return f"""Sie sind ein hilfreicher Assistent mit Zugriff auf folgende Werkzeuge:
{tool_descriptions}

Wenn Sie ein Werkzeug verwenden müssen, antworten Sie mit:
TOOL: werkzeugname
ARGS: arg1=wert1, arg2=wert2

Andernfalls antworten Sie direkt."""

Werkzeugausführung hinzufügen

Das Framework muss Werkzeugaufrufe parsen und sicher ausführen.

# agent_harness/core.py (Fortsetzung)
    def run(self, user_input: str) -> str:
        """Benutzereingabe verarbeiten und Agentenantwort zurückgeben."""
        self.history.append({"role": "user", "content": user_input})
        
        # Nachrichten für LLM erstellen
        messages = [{"role": "system", "content": self.build_system_prompt()}]
        messages.extend(self.history)
        
        # LLM-Antwort abrufen
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=0.7
        )
        
        content = response.choices[0].message.content
        
        # Auf Werkzeugaufruf prüfen
        if content.startswith("TOOL:"):
            tool_name = content.split("\n")[0].replace("TOOL:", "").strip()
            args_line = content.split("\n")[1].replace("ARGS:", "").strip()
            args = self._parse_args(args_line)
            
            # Werkzeug ausführen
            for tool in self.tools:
                if tool.name == tool_name:
                    result = tool.function(**args)
                    self.history.append({"role": "assistant", "content": content})
                    self.history.append({"role": "tool", "content": str(result)})
                    return self.run(f"Werkzeugergebnis: {result}")
                    
        self.history.append({"role": "assistant", "content": content})
        return content
        
    def _parse_args(self, args_str: str) -> Dict[str, Any]:
        """Schlüssel=Wert-Argumente aus einem String parsen."""
        args = {}
        for pair in args_str.split(","):
            key, value = pair.split("=")
            args[key.strip()] = value.strip()
        return args

Beispiel-Werkzeuge erstellen

Websuche-Werkzeug

Ein einfaches Werkzeug, das eine simulierte Websuche durchführt. In der Produktion durch eine echte API ersetzen.

# agent_harness/tools/search.py
import httpx

def web_search(query: str) -> str:
    """Eine simulierte Websuche durchführen."""
    # In der Produktion eine echte Such-API wie SerpAPI oder Bing verwenden
    return f"Simulierte Suchergebnisse für '{query}': 3 relevante Dokumente gefunden."

Rechner-Werkzeug

Ein sicherer Rechner mit Pythons `eval` und eingeschränkten Globals.

# agent_harness/tools/calculator.py
def calculator(expression: str) -> str:
    """Einen mathematischen Ausdruck sicher auswerten."""
    allowed_names = {"abs", "round", "min", "max", "sum", "pow"}
    try:
        # Nur grundlegende mathematische Operationen erlauben
        result = eval(expression, {"__builtins__": {}}, {name: __builtins__.__dict__[name] for name in allowed_names})
        return f"Ergebnis: {result}"
    except Exception as e:
        return f"Fehler: {str(e)}"

Anwendungsbeispiele

Das Framework ausführen

Erstellen Sie ein Hauptskript, um den Agenten in Aktion zu zeigen.

# main.py
from agent_harness.core import Agent, Tool
from agent_harness.tools.search import web_search
from agent_harness.tools.calculator import calculator

def main():
    # Agent initialisieren
    agent = Agent(model="gpt-4")
    
    # Werkzeuge registrieren
    agent.add_tool(Tool(
        name="web_search",
        description="Im Internet nach Informationen suchen",
        function=web_search
    ))
    agent.add_tool(Tool(
        name="calculator",
        description="Mathematische Ausdrücke auswerten",
        function=calculator
    ))
    
    # Interaktive Schleife
    print("Agenten-Framework bereit. 'exit' zum Beenden.")
    while True:
        user_input = input("\nSie: ")
        if user_input.lower() == "exit":
            break
            
        response = agent.run(user_input)
        print(f"Agent: {response}")

if __name__ == "__main__":
    main()

Führen Sie das Framework aus:

python main.py

Beispiel-Interaktion:

Agenten-Framework bereit. 'exit' zum Beenden.

Sie: Was ist 2 + 2?
Agent: 4

Sie: Suche nach neuesten KI-Nachrichten
Agent: Simulierte Suchergebnisse für 'neueste KI-Nachrichten': 3 relevante Dokumente gefunden.

Erweiterte Konfiguration

Unterstützung für Gesprächsspeicher und Werkzeug-Timeout hinzufügen.

# agent_harness/core.py (zur Agent-Klasse hinzufügen)
    def clear_history(self):
        """Gesprächskontext zurücksetzen."""
        self.history = []
        
    def run_with_timeout(self, user_input: str, timeout: int = 30) -> str:
        """Mit Timeout ausführen, um Sicherheit zu gewährleisten."""
        import signal
        
        def handler(signum, frame):
            raise TimeoutError("Agentenausführung abgelaufen")
            
        signal.signal(signal.SIGALRM, handler)
        signal.alarm(timeout)
        
        try:
            return self.run(user_input)
        finally:
            signal.alarm(0)

Best Practices

Fehlerbehandlung

Werkzeugausführung in try-except-Blöcke einwickeln, um Agentenabstürze zu verhindern.

# In Agent.run()
try:
    result = tool.function(**args)
except Exception as e:
    result = f"Werkzeugausführung fehlgeschlagen: {str(e)}"

Ratenbegrenzung

Exponentielles Backoff für API-Aufrufe implementieren.

import time
from openai import RateLimitError

def run_with_retry(self, user_input: str, max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            return self.run(user_input)
        except RateLimitError:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)

Protokollierung

Strukturierte Protokollierung für Debugging und Überwachung hinzufügen.

import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# In Agent.run()
logger.info(f"Benutzereingabe: {user_input}")
logger.info(f"Werkzeug aufgerufen: {tool_name}")

Fazit

Ein benutzerdefiniertes Agenten-Framework zu entwickeln, gibt Ihnen die vollständige Kontrolle darüber, wie Ihre KI-Agenten arbeiten. Diese Implementierung bietet eine solide Grundlage mit:

• Einer modularen Architektur, die Kernlogik von Werkzeugen trennt
• Sicherer Werkzeugausführung mit Fehlerbehandlung
• Sauberer Verwaltung des Gesprächsverlaufs
• Einfacher Erweiterbarkeit für neue Funktionen

Das Framework kann um Funktionen wie parallele Werkzeugausführung, Streaming-Antworten oder persistenten Zustand erweitert werden. Beginnen Sie mit dieser Basis und fügen Sie dann Werkzeuge hinzu, die für Ihre Domäne spezifisch sind – sei es für Datenbankabfragen, API-Integrationen oder benutzerdefinierte Datenverarbeitung. Der Schlüssel liegt darin, die Trennung zwischen der Argumentation des Agenten und der Werkzeugausführung beizubehalten, was Debugging und Skalierung vereinfacht.

Wenn Sie komplexere Agenten entwickeln, denken Sie daran, Sicherheit durch Eingabevalidierung, Timeout-Mechanismen und Ressourcenüberwachung zu priorisieren. Ein gut gebautes Framework wird das zuverlässige Rückgrat Ihrer KI-Anwendungen bilden, egal ob Sie einen Chatbot, einen Forschungsassistenten oder ein automatisiertes Workflow-System betreiben.