KI-Agenten stellen eine neue Generation intelligenter Systeme dar, die zu autonomer Entscheidungsfindung und Handlung faehig sind. Dieser Artikel untersucht ihre Architektur, Designprinzipien und praktische Implementierungen auf Basis grosser Sprachmodelle.
Grundprinzipien der KI-Agenten-Architektur¶
KI-Agenten stellen den naechsten evolutionaeren Schritt von statischen Chatbots zu autonomen Systemen dar, die zu komplexer Entscheidungsfindung und Handlung faehig sind. Ein richtig entworfener Agent kombiniert mehrere Schluesselkomponenten: Reasoning Engine, Memory-System, Tool-Integration und Execution Layer.
Die grundlegende Architektur eines modernen KI-Agenten basiert auf dem ReAct (Reasoning + Acting)-Muster, bei dem der Agent schrittweise die Situation analysiert, Schritte plant und Aktionen ausfuehrt. Dieser Zyklus wiederholt sich, bis das Ziel erreicht oder die Ressourcen erschoepft sind.
class AIAgent:
def __init__(self, llm_client, tools, memory):
self.llm = llm_client
self.tools = tools
self.memory = memory
self.max_iterations = 10
async def execute(self, task):
for iteration in range(self.max_iterations):
# Reasoning phase
context = self.memory.get_context()
response = await self.llm.generate(
prompt=f"Task: {task}\nContext: {context}\n"
f"Available tools: {list(self.tools.keys())}"
)
# Acting phase
if self.should_use_tool(response):
result = await self.execute_tool(response)
self.memory.add_step(response, result)
else:
return response
Memory Management¶
Ein kritisches Element ist das Memory Management des Agenten. Wir unterscheiden drei Typen von Speicher: Working Memory fuer den aktuellen Kontext, Episodic Memory fuer spezifische Ereignisse und Semantic Memory fuer allgemeines Wissen.
class AgentMemory:
def __init__(self, vector_store, max_working_memory=4096):
self.vector_store = vector_store
self.working_memory = []
self.max_tokens = max_working_memory
def add_step(self, thought, action, result):
step = {
"timestamp": datetime.now(),
"thought": thought,
"action": action,
"result": result,
"embedding": self.get_embedding(f"{thought} {result}")
}
# Store in long-term memory
self.vector_store.store(step)
# Add to working memory with size management
self.working_memory.append(step)
self.manage_working_memory_size()
def retrieve_relevant_memories(self, query, top_k=3):
query_embedding = self.get_embedding(query)
return self.vector_store.similarity_search(
query_embedding, top_k
)
Tool-Integration und Function Calling¶
Moderne KI-Agenten nutzen Function Calling zur Interaktion mit externen Systemen. Entscheidend ist die Definition klarer Interfaces und Error Handling fuer jedes Tool.
class ToolRegistry:
def __init__(self):
self.tools = {}
def register_tool(self, name, func, schema):
self.tools[name] = {
"function": func,
"schema": schema,
"retry_config": {"max_retries": 3, "backoff": 2}
}
async def execute_tool(self, tool_name, parameters):
if tool_name not in self.tools:
raise ValueError(f"Tool {tool_name} not found")
tool = self.tools[tool_name]
try:
# Validate parameters against schema
self.validate_parameters(parameters, tool["schema"])
# Execute with retry logic
return await self.execute_with_retry(
tool["function"],
parameters,
tool["retry_config"]
)
except Exception as e:
return {"error": str(e), "tool": tool_name}
# KI-Agenten — Architektur und Design
@tools.register("web_search", {
"type": "function",
"parameters": {
"query": {"type": "string"},
"max_results": {"type": "integer", "default": 5}
}
})
async def web_search(query, max_results=5):
# Implementation
pass
Planung und mehrstufige Ausfuehrung¶
Fuer komplexere Aufgaben benoetigt der Agent Planungsfaehigkeiten. Wir implementieren hierarchische Planung mit der Moeglichkeit zur Anpassung waehrend der Ausfuehrung.
class TaskPlanner:
def __init__(self, llm_client):
self.llm = llm_client
async def create_plan(self, goal, available_tools):
planning_prompt = f"""
Goal: {goal}
Available tools: {list(available_tools.keys())}
Create a step-by-step plan. Each step should:
1. Be atomic and executable
2. Specify required tool if needed
3. Include success criteria
Format as JSON array of steps.
"""
response = await self.llm.generate(planning_prompt)
return self.parse_plan(response)
async def adapt_plan(self, current_plan, executed_steps, error_info):
adaptation_prompt = f"""
Original plan: {current_plan}
Executed steps: {executed_steps}
Error encountered: {error_info}
Adapt the remaining plan to handle the error and achieve the goal.
"""
response = await self.llm.generate(adaptation_prompt)
return self.parse_plan(response)
Monitoring und Observability¶
Im Produktiveinsatz ist die Ueberwachung von Leistung und Verhalten des Agenten entscheidend. Wir implementieren umfassendes Logging und Metriken fuer Debugging und Optimierung.
class AgentObserver:
def __init__(self, logger, metrics_client):
self.logger = logger
self.metrics = metrics_client
def log_agent_step(self, agent_id, step_type, details, duration):
log_entry = {
"agent_id": agent_id,
"step_type": step_type,
"details": details,
"duration_ms": duration,
"timestamp": datetime.now().isoformat()
}
self.logger.info("agent_step", extra=log_entry)
# Track metrics
self.metrics.histogram("agent.step.duration",
duration, tags={"type": step_type})
self.metrics.increment("agent.steps.total",
tags={"type": step_type})
def track_tool_usage(self, tool_name, success, error_type=None):
tags = {"tool": tool_name, "success": success}
if error_type:
tags["error_type"] = error_type
self.metrics.increment("agent.tool.calls", tags=tags)
def track_token_usage(self, prompt_tokens, completion_tokens):
self.metrics.gauge("agent.tokens.prompt", prompt_tokens)
self.metrics.gauge("agent.tokens.completion", completion_tokens)
Fehlerbehandlung und Wiederherstellung¶
Robuste Agenten muessen Fehler elegant behandeln. Wir implementieren mehrstufige Fehlerwiederherstellung mit Fallback-Strategien.
class ErrorRecoveryManager:
def __init__(self, max_recovery_attempts=3):
self.max_attempts = max_recovery_attempts
self.recovery_strategies = {
"tool_error": self.handle_tool_error,
"llm_error": self.handle_llm_error,
"timeout_error": self.handle_timeout_error
}
async def handle_error(self, error, context):
error_type = self.classify_error(error)
if error_type in self.recovery_strategies:
return await self.recovery_strategies[error_type](
error, context
)
else:
# Generic fallback
return await self.generic_recovery(error, context)
async def handle_tool_error(self, error, context):
# Try alternative tool or simplified approach
if hasattr(error, 'tool_name'):
alternative_tools = self.find_alternative_tools(
error.tool_name, context
)
if alternative_tools:
return {"strategy": "retry_with_alternative",
"tools": alternative_tools}
return {"strategy": "simplify_task"}
Sicherheit und Rate Limiting¶
Die Sicherheit von KI-Agenten erfordert einen mehrschichtigen Ansatz, einschliesslich Input-Sanitization, Output-Validierung und Ressourcenmanagement.
class AgentSecurityManager:
def __init__(self, rate_limiter, content_filter):
self.rate_limiter = rate_limiter
self.content_filter = content_filter
self.dangerous_patterns = [
r'rm\s+-rf',
r'DROP\s+TABLE',
r'eval\s*\(',
]
async def validate_input(self, user_input, user_id):
# Rate limiting
if not await self.rate_limiter.allow(user_id):
raise SecurityError("Rate limit exceeded")
# Content filtering
if self.content_filter.is_malicious(user_input):
raise SecurityError("Malicious content detected")
# Pattern matching for dangerous commands
for pattern in self.dangerous_patterns:
if re.search(pattern, user_input, re.IGNORECASE):
raise SecurityError(f"Dangerous pattern detected: {pattern}")
def sanitize_tool_parameters(self, tool_name, parameters):
# Tool-specific sanitization
if tool_name == "file_system":
parameters = self.sanitize_file_paths(parameters)
elif tool_name == "database":
parameters = self.sanitize_sql_inputs(parameters)
return parameters
Zusammenfassung¶
Eine erfolgreiche KI-Agenten-Architektur kombiniert modulares Design mit robusten Mechanismen fuer Memory Management, Tool-Integration und Fehlerwiederherstellung. Der Schluessel liegt in der Balance zwischen Agentenautonomie und Sicherheitsmassnahmen. Bei der Implementierung sollte der Fokus auf Observability und schrittweisem Testen einzelner Komponenten liegen. Moderne Agenten sind nicht nur Chatbots mit Tools – sie sind ausgefeilte Systeme, die sorgfaeltige architektonische Entscheidungen und kontinuierliches Monitoring in der Produktion erfordern.