KI-Agent-Tools und externe Integration
Dieses Kapitel führt ein, wie Agenten mit der Außenwelt interagieren.
Die Fähigkeit zur Tool-Aufrufung ist der Schlüssel, um Agenten mit der externen Umgebung zu verbinden.
Durch die Integration verschiedener Tools können Agenten Code ausführen, auf APIs zugreifen, Dateisysteme bedienen und mehr.
Computer Use
Computer Use ermöglicht es Agenten, Computeroberflächen wie Menschen zu bedienen.
Dazu gehören grafische Oberflächen wie Browser und Desktop-Anwendungen.
Dies ist ein wichtiger Schritt zur Erreichung allgemeiner künstlicher Intelligenz (AGI).
Kernfähigkeiten
Screenshot-Analyse: Den Inhalt auf dem Bildschirm verstehen und interaktive Elemente identifizieren.
GUI-Elementerkennung: Schnittstellenelemente wie Schaltflächen, Eingabefelder und Menüs erkennen.
Maus-/Tastatursteuerung: Menschliche Operationen simulieren, Aktionen wie Klicken und Tippen ausführen.
Browser-Automatisierung: Web-Navigation, Formularausfüllung, Suche usw.
Funktionsprinzip
Der Workflow von Computer Use lässt sich wie folgt zusammenfassen: beobachten, verstehen, entscheiden und ausführen.
Erster Schritt, Screenshot. Den Inhalt des aktuellen Bildschirms oder Fensters erfassen.
Zweiter Schritt, visuelle Analyse. Ein Vision-Modell verwenden, um den Screenshot zu analysieren und Schnittstellenelemente und -zustände zu identifizieren.
Dritter Schritt, Entscheidungsfindung. Die nächste Aktion basierend auf dem Aufgabenziel bestimmen.
Vierter Schritt, Ausführung. Aktionen wie Mausklicks und Tastatureingaben ausführen.
Wiederhole die Schleife, bis die Aufgabe abgeschlossen ist.
Code-Implementierung
Grundlegende Implementierung eines Computer-Use-Agenten
"""
Implementierung eines Computer-Use-Agenten
Kann Computeroberflächen wie ein Mensch bedienen
"""
def __init__(self, vision_model, action_executor, planner):
# Vision-Modell: Screenshot analysieren
self.vision_model = vision_model
# Aktionsausführer: Maus- und Tastaturoperationen ausführen
self.action_executor = action_executor
# Planer: Nächste Aktion entscheiden
self.planner = planner
# Maximale Ausführungsschritte
self.max_steps = 100
def observe(self, screenshot):
"""
Screenshot analysieren, interaktive Elemente identifizieren
:param screenshot: Screenshot-Bild
:return: Liste der Schnittstellenelemente
"""
# Vision-Modell verwenden, um Screenshot zu analysieren
analysis = self.vision_model.analyze(screenshot)
# Erkannte Schnittstellenelemente zurückgeben
# Jedes Element enthält: Typ, Position, Inhalt, Interaktivität
return analysis.elements
def act(self, action):
"""
Aktion ausführen
:param action: Aktionsbeschreibung, z. B. {"type": "click", "x": 100, "y": 200}
"""
return self.action_executor.execute(action)
def run(self, task):
"""
Aufgabenschleife ausführen
:param task: Aufgabenbeschreibung
:return: Aufgabenergebnis
"""
# Aufgabenstatus initialisieren
self.planner.set_task(task)
for step in range(self.max_steps):
# Schritt 1: Aktuellen Bildschirm erfassen
screenshot = self.get_screen()
# Schritt 2: Beobachten - Schnittstellenelemente identifizieren
elements = self.observe(screenshot)
# Schritt 3: Entscheiden - Nächste Aktion bestimmen
action = self.planner.decide_action(elements)
# Prüfen, ob die Aufgabe abgeschlossen ist
if action.is_final:
return action.result
# Schritt 4: Aktion ausführen
self.act(action)
# Optional: Auf Aktualisierung der Oberfläche warten
self.wait_for_update()
return "Maximale Schrittzahl erreicht"
class VisionModel:
"""Vision-Modell: Screenshot analysieren"""
def analyze(self, screenshot):
"""
Screenshot analysieren
Liste der Oberflächenelemente zurückgeben
"""
# Multimodales Modell zur Analyse verwenden
prompt = """
Analysiere diesen Screenshot und identifiziere alle interaktiven Oberflächenelemente.
Einschließlich: Schaltflächen, Eingabefelder, Links, Menüs usw.
Bitte für jedes Element angeben:
1. Typ (Schaltfläche, Eingabefeld, Link usw.)
2. Position (Bounding-Box-Koordinaten)
3. Inhalt (Schaltflächentext, Eingabe-Platzhalter usw.)
4. Interaktivität (Sichtbarkeit, aktivierter Zustand)
"""
result = self.vision_model.analyze_image(screenshot, prompt)
return ScreenAnalysisResult(elements=result.elements)
class ActionExecutor:
"""Aktions-Executor: Maus- und Tastaturoperationen ausführen"""
def execute(self, action):
"""
Aktion ausführen
:param action: Aktionsobjekt
"""
if action.type == "click":
self.mouse.click(action.x, action.y)
elif action.type == "type":
self.keyboard.type_text(action.text)
elif action.type == "scroll":
self.mouse.scroll(action.direction, action.amount)
elif action.type == "press":
self.keyboard.press_key(action.key)
return ActionResult(success=True)
class Planner:
"""Planer: Nächste Aktion entscheiden"""
def __init__(self, llm):
self.llm = llm
self.task = None
self.history = []
def set_task(self, task):
"""Aktuelle Aufgabe festlegen"""
self.task = task
self.history = []
def decide_action(self, elements):
"""
Nächste Aktion anhand des aktuellen Oberflächenzustands bestimmen
"""
prompt = f"""
Aktuelle Aufgabe: {self.task}
Verlauf der ausgeführten Aktionen:
{self.history}
Interaktive Elemente auf der aktuellen Oberfläche:
{elements}
Bitte entscheiden Sie die nächste Aktion.
Wenn die Aufgabe abgeschlossen ist, geben Sie is_final=True zurück.
Andernfalls geben Sie die auszuführende Aktion zurück (Typ, Koordinaten, Parameter usw.).
"""
response = self.llm.generate(prompt)
return Action.parse(response)
Anwendungsszenarien
Computer Use eignet sich besonders für die folgenden Szenarien:
Web-Automatisierung: wie automatisches Ausfüllen von Formularen, Scraping von Daten und Ausführen von Web-Vorgängen.
Desktop-Anwendungsoperationen: wie das Öffnen von Dateien, das Bearbeiten von Dokumenten und das Bedienen von Software.
Testautomatisierung: wie automatisches Ausführen von UI-Tests und Regressionstests.
Hinweis: Computer Use befindet sich noch in Entwicklung und hat Probleme wie langsame Ausführung und gelegentliche Fehler. Für Szenarien mit einer klaren API ist der direkte Aufruf der API normalerweise effizienter als Computer Use.
Detaillierte Erklärung des MCP-Protokolls
MCP (Model Context Protocol) ist ein offenes Standardprotokoll.
Es ermöglicht KI-Modellen, sich sicher mit externen Tools und Datenquellen zu verbinden.
Das Designziel von MCP ist, die "USB-Schnittstelle" des KI-Bereichs zu werden.
Kern-Designprinzipien
Standardisierung: Einheitliches Protokollformat, das die Interoperabilität verschiedener Agents und Tools ermöglicht.
Sicherheit: Klare Berechtigungskontrolle, Agents können nur auf autorisierte Ressourcen zugreifen.
Erweiterbarkeit: Einfaches Hinzufügen neuer Tools und Datenquellen.
Architekturkomponenten
MCP Host: Host-Umgebung zum Ausführen von KI-Anwendungen, wie Claude Desktop, KI-Coding-Assistenten usw.
MCP Client: Ein Client, der eine 1:1-Verbindung mit dem MCP Server aufrechterhält.
MCP Server: Serverprogramm, das Tools und Ressourcen bereitstellt.
Kommunikationsablauf
Erster Schritt, Verbindungsaufbau. Der Client stellt eine Verbindung mit dem Server her und tauscht Fähigkeitsinformationen aus.
Zweiter Schritt, Tool-Erkennung. Der Client fragt den Server nach verfügbaren Tools und Ressourcen ab.
Dritter Schritt, Tool-Aufruf. Der Client sendet eine Tool-Aufrufanforderung, und der Server führt sie aus und gibt das Ergebnis zurück.
Vierter Schritt, Ressourcenzugriff. Der Client kann Ressourcen lesen und schreiben, die vom Server bereitgestellt werden.
Beispiel für eine MCP-Server-Implementierung
MCP-Server-Implementierung
from mcp.types import Tool, Resource
import asyncio
# MCP-Server-Instanz erstellen
app = Server("filesystem")
# Dateisystem-Tools definieren
@app.list_tools()
async def list_tools():
"""Alle verfügbaren Tools auflisten"""
return [
Tool(
name="read_file",
description="Dateiinhalt lesen",
inputSchema={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Dateipfad"
},
"encoding": {
"type": "string",
"description": "Dateikodierung, standardmäßig utf-8",
"default": "utf-8"
}
},
"required": ["path"]
}
),
Tool(
name="write_file",
description="Dateiinhalt schreiben",
inputSchema={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Dateipfad"
},
"content": {
"type": "string",
"description": "Dateiinhalt"
}
},
"required": ["path", "content"]
}
),
Tool(
name="list_directory",
description="Verzeichnisinhalt auflisten",
inputSchema={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Verzeichnispfad"
}
}
}
)
]
@app.call_tool()
async def call_tool(name, arguments):
"""Tool-Aufruf ausführen"""
if name == "read_file":
return await read_file(arguments["path"], arguments.get("encoding", "utf-8"))
elif name == "write_file":
return await write_file(arguments["path"], arguments["content"])
elif name == "list_directory":
return await list_directory(arguments["path"])
else:
raise ValueError(f"Unbekanntes Tool: {name}")
async def read_file(path, encoding):
"""Datei lesen"""
with open(path, "r", encoding=encoding) as f:
content = f.read()
return [{"type": "text", "text": content}]
async def write_file(path, content):
"""Datei schreiben"""
with open(path, "w", encoding="utf-8") as f:
f.write(content)
return [{"type": "text", "text": "Datei erfolgreich geschrieben"}]
async def list_directory(path):
"""Verzeichnis auflisten"""
import os
entries = os.listdir(path)
return [{"type": "text", "text": "\n".join(entries)}]
# Server ausführen
if __name__ == "__main__":
import mcp.server.stdio
asyncio.run(mcp.server.stdio.serve(app))
API-Integrationsmuster
Es gibt drei gängige Möglichkeiten für einen Agenten, externe APIs zu integrieren.
REST API ist der am häufigsten verwendete Web-API-Stil.
GraphQL bietet flexiblere Datenabfragemöglichkeiten.
gRPC eignet sich für Remote-Prozeduraufrufe in Hochleistungsszenarien.
REST-API-Integration
REST (Representational State Transfer) ist ein Web-API-Designstil.
Es verwendet HTTP-Methoden (GET, POST, PUT, DELETE) für Operationen.
REST-API-Client-Implementierung
from typing import Dict, Any, Optional
class RESTAPIClient:
"""
REST-API-Client
Kapselt HTTP-Anfragen und bietet eine prägnante Aufrufschnittstelle
"""
def __init__(self, base_url: str, headers: Optional[Dict] = None):
"""
API-Client initialisieren
:param base_url: Basis-URL der API
:param headers: Standard-Anfrage-Header
"""
self.base_url = base_url.rstrip("/")
self.session = requests.Session()
# Standard-Anfrage-Header festlegen
default_headers = {
"Content-Type": "application/json",
"Accept": "application/json"
}
if headers:
default_headers.update(headers)
self.session.headers.update(default_headers)
def call(
self,
method: str,
endpoint: str,
params: Optional[Dict] = None,
data: Optional[Dict] = None,
**kwargs
) -> Dict[Any, Any]:
"""
API-Anfrage senden
:param method: HTTP-Methode (GET, POST, PUT, DELETE)
:param endpoint: API-Endpunkt
:param params: URL-Abfrageparameter
:param data: Daten des Anfragekörpers
:return: Antwortdaten
"""
url = f"{self.base_url}/{endpoint.lstrip('/')}"
response = self.session.request(
method=method.upper(),
url=url,
params=params,
json=data,
**kwargs
)
# HTTP-Statuscode prüfen
response.raise_for_status()
# Antwort parsen
if response.content:
return response.json()
return {}
def get(self, endpoint: str, params: Optional[Dict] = None):
"""GET-Anfrage"""
return self.call("GET", endpoint, params=params)
def post(self, endpoint: str, data: Dict):
"""POST-Anfrage"""
return self.call("POST", endpoint, data=data)
def put(self, endpoint: str, data: Dict):
"""PUT-Anfrage"""
return self.call("PUT", endpoint, data=data)
def delete(self, endpoint: str, params: Optional[Dict] = None):
"""DELETE-Anfrage"""
return self.call("DELETE", endpoint, params=params)
class APIAgent:
"""
API-Agent
Externe APIs über natürliche Sprache aufrufen
"""
def __init__(self, api_client: RESTAPIClient):
self.api_client = api_client
self.tool_descriptions = self.define_tools()
def define_tools(self):
"""
Verfügbare Tools für den Agenten definieren
Toolbeschreibungen zurückgeben, damit das LLM versteht, wie es sie aufrufen kann
"""
return [
{
"name": "get_weather",
"description": "Wetterinformationen für eine angegebene Stadt abrufen",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "Stadtname, z. B. Peking, Shanghai"
}
},
"required": ["city"]
}
},
{
"name": "get_forecast",
"description": "Wettervorhersage für eine angegebene Stadt abrufen",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "Stadtname"
},
"days": {
"type": "integer",
"description": "Vorhersagetage, Standardwert 3",
"default": 3
}
},
"required": ["city"]
}
}
]
def execute_tool(self, tool_name: str, arguments: Dict):
"""
Tool-Aufruf ausführen
:param tool_name: Tool-Name
:param arguments: Tool-Argumente
:return: Ausführungsergebnis
"""
if tool_name == "get_weather":
return self.api_client.get(f"/weather/{arguments['city']}")
elif tool_name == "get_forecast":
params = {"days": arguments.get("days", 3)}
return self.api_client.get(f"/forecast/{arguments['city']}", params=params)
else:
raise ValueError(f"Unbekanntes Tool: {tool_name}")
Best Practices für API-Integration
Fehlerbehandlung: Netzwerkfehler, Timeouts, Dienstnichtverfügbarkeit usw. elegant behandeln.
Wiederholungsmechanismus: Bei vorübergehenden Fehlern einen Wiederholungsversuch mit exponentiellem Backoff implementieren.
Ratenbegrenzung: Respektiere die Ratenlimits der API, um eine Sperre zu vermeiden.
Caching-Strategie: Häufig angeforderte Daten zwischenspeichern, um API-Aufrufe zu reduzieren.
Sicherheitsüberlegungen: Sensible Informationen wie API-Schlüssel nicht fest im Code hinterlegen; verwenden Sie Umgebungsvariablen, um sie zu verwalten.
Kapitelzusammenfassung
Dieses Kapitel führt die Kerntechnologien ein, mit denen Agenten mit der Außenwelt interagieren.
Computer UseErmöglicht Agenten, GUI-Oberflächen zu bedienen, geeignet für Szenarien, die eine Interaktion mit grafischen Anwendungen erfordern.
MCP-ProtokollBietet eine standardisierte Möglichkeit, Tools zu integrieren, und fungiert als "USB-Schnittstelle" des Agenten-Zeitalters.
API-IntegrationIst die gebräuchlichste Methode zur Integration externer Systeme, einschließlich REST, GraphQL, gRPC usw.
Die Wahl der geeigneten Integrationsmethode hängt vom konkreten Szenario ab.
Für grafische Schnittstellenoperationen ist Computer Use eine universelle Lösung.
Für standardisierte Tool-Integration ist MCP die zukünftige Entwicklungsrichtung.
Für die Web-API-Integration bleibt REST API die gängigste Wahl.
Andere Erweiterungen