AI agenti — architektura a design
AI agenti představují novou generaci inteligentních systémů schopných autonomního rozhodování a jednání. Tento článek prozkoumává jejich architekturu, principy designu a praktické implementace založené na velkých jazykových modelech.
Základní principy architektury AI agentů
AI agenti představují další evoluční stupeň od statických chatbotů k autonomním systémům schopným komplexního rozhodování a jednání. Správně navržený agent kombinuje několik klíčových komponent: reasoning engine, memory systém, tool integration a execution layer.
Základní architektura moderního AI agenta vychází ze vzoru ReAct (Reasoning + Acting), kde agent postupně analyzuje situaci, plánuje kroky a provádí akce. Tento cyklus se opakuje až do dosažení cíle nebo vyčerpání zdrojů.
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
Kritickým prvkem je správa paměti agenta. Rozlišujeme tři typy paměti: working memory pro aktuální kontext, episodic memory pro konkrétní události a semantic memory pro obecné znalosti.
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 a Function Calling
Moderní AI agenti využívají function calling pro interakci s externími systémy. Klíčové je definovat jasné interface a error handling pro každý 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}
# Example tool registration
@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
Planning a Multi-step Execution
Pro složitější úkoly agent potřebuje schopnost plánování. Implementujeme hierarchické plánování s možností adaptace během provádění.
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 a Observability
V produkčním nasazení je klíčové sledování výkonnosti a chování agenta. Implementujeme comprehensive logging a metriky pro debugging a optimalizaci.
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)
Error Handling a Recovery
Robustní agenti musí elegantně zvládat selhání. Implementujeme multi-level error recovery s fallback strategiemi.
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"}
Security a Rate Limiting
Bezpečnost AI agentů vyžaduje vícevrstvý přístup zahrnující input sanitization, output validation a resource management.
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
Shrnutí
Úspěšná architektura AI agenta kombinuje modulární design s robustními mechanismy pro memory management, tool integration a error recovery. Klíčové je najít rovnováhu mezi autonomií agenta a bezpečnostními opatřeními. Při implementaci dbejte na observability a postupné testování jednotlivých komponent. Moderní agenti nejsou jen chattboti s tools - jsou to sofistikované systémy vyžadující pečlivé architektonické rozhodnutí a kontinuální monitoring v produkci.