模块化架构示例

class AIAutomationSystem:
def __init__(self):
self.llm_client = LLMClient()
self.memory_store = VectorMemory()
self.tool_registry = ToolRegistry()
self.task_orchestrator = TaskOrchestrator()
self.audit_logger = AuditLogger()

async def process_request(self, user_input: str) -> Response:
# 1. 意图识别
intent = await self.llm_client.analyze_intent(user_input)

# 2. 上下文检索
context = await self.memory_store.retrieve_relevant(user_input)

# 3. 任务规划
plan = await self.task_orchestrator.create_plan(intent, context)

# 4. 执行与监控
result = await self._execute_plan(plan)

# 5. 记录审计日志
await self.audit_logger.log_execution(user_input, result)

// 记录请求入口
logger.info(‘AI_REQUEST_START’, {
traceId,
userId: context.userId,
input: sanitizeInput(context.input),
timestamp: new Date().toISOString()
});

try {
const result = await next();

// 记录成功响应
logger.info(‘AI_REQUEST_SUCCESS’, {
traceId,
duration: Date.now()

• startTime,

tokensUsed: result.tokenCount,
toolCalls: result.toolCalls?.length
});

容错处理示例

from tenacity import retry, stop_after_attempt, wait_exponential

LLM 网关配置示例

llm_gateway:
providers:

• name: qwen

endpoint: https://dashscope.aliyuncs.com/api/v1
models:

• qwen-plus

• qwen-max

rate_limit: 100 req/min
cost_per_1k_tokens: 0.002

• name: deepseek

endpoint: https://api.deepseek.com/v1
models:

• deepseek-chat

rate_limit: 50 req/min
cost_per_1k_tokens: 0.001

routing_rules:

• condition: “token_count > 10000”

route_to: qwen-max

• condition: “latency_sensitive == true”

route_to: qwen-plus

• default: deepseek-chat

// 长期记忆：向量存储的知识
longTerm: VectorStore;

// 工作记忆：当前任务的临时状态
working: TaskState;
}

class HybridMemorySystem implements MemoryManager {
async retrieveContext(query: string, sessionId: string): Promise {
// 1. 获取会话历史（最近 N 条消息）
const recentHistory = await this.shortTerm.get(sessionId, limit=10);

// 2. 向量检索相关记忆
const relevantMemories = await this.longTerm.search(query, topK=5);

// 3. 获取当前任务状态
const taskState = await this.working.get(sessionId);

工具注册与执行

from dataclasses import dataclass
from typing import Any, Callable, Dict

@dataclass
class ToolDefinition:
name: str
description: str
parameters: Dict[str, Any]
handler: Callable
timeout_seconds: int = 30
requires_approval: bool = False

class ToolExecutionEngine:
def __init__(self):
self.tools: Dict[str, ToolDefinition] = {}
self.execution_log = []

def register_tool(self, tool: ToolDefinition):
self.tools[tool.name] = tool
logger.info(f”Registered tool: {tool.name}”)

async def execute(self, tool_name: str, arguments: Dict) -> Any:
if tool_name not in self.tools:
raise ToolNotFoundError(f”Unknown tool: {tool_name}”)

tool = self.tools[tool_name]

# 安全检查
if tool.requires_approval:
await self.request_approval(tool_name, arguments)

def render(self, template_name: str, variables) -> str:
template = self.templates.get(template_name)
if not template:
raise TemplateNotFoundError(template_name)

# 注入标准变量
variables.update({
‘current_date’: datetime.now().isoformat(),
‘system_version’: ‘2.1.0’,
‘safety_guidelines’: self._get_safety_guidelines()
})

return template.render(**variables)

def load_templates(self):
self.templates[‘system_prompt’] = Template(“””
你是一个专业的 AI 助手，专注于{{ domain }}领域的任务。

## 能力范围
– {{ capabilities | join(‘, ‘) }}

约束条件

• 不执行未经批准的外部操作
• 不处理敏感个人信息

– 所有操作必须可审计

AI 系统测试框架

import pytest
from unittest.mock import AsyncMock

class TestAIAutomationSystem:
@pytest.fixture
def mock_llm(self):
llm = AsyncMock()
llm.generate.return_value = {
‘content’: ‘Expected response’,
‘tool_calls’: []
}
return llm

@pytest.mark.asyncio
async def test_intent_classification(self, mock_llm):
system = AIAutomationSystem(llm_client=mock_llm)

result = await system.classify_intent(“帮我查询昨天的天气”)

assert result.category == ‘information_retrieval’
assert result.domain == ‘weather’
assert result.confidence > 0.8

@pytest.mark.asyncio
async def test_tool_execution_safety(self):
system = AIAutomationSystem()

• INTERVAL ’30 days’

GROUP BY DATE(request_timestamp), model_name
ORDER BY date DESC, daily_cost DESC;

输入安全过滤器

import re
from typing import Optional

class InputSecurityFilter:
# 危险模式检测
DANGEROUS_PATTERNS = [
r'(?i)ignore\s+(previous|above)\s+instructions’,
r'(?i)system\s+prompt’,
r'(?i)bypass\s+(safety|security)’,
r'(?i)developer\s+mode’,
r'(?i)role\s+play\s+as\s+(admin|system|developer)’,
]

def validate(self, user_input: str) -> ValidationResult:
issues = []

# 检查注入尝试
for pattern in self.DANGEROUS_PATTERNS:
if re.search(pattern, user_input):
issues.append(f”Detected potential prompt injection: {pattern}”)

# 检查长度限制
if len(user_input) > self.max_length:
issues.append(f”Input exceeds maximum length ({len(user_input)} > {self.max_length})”)

# 检查敏感信息
if self.contains_sensitive_info(user_input):
issues.append(“Input may contain sensitive information”)

审计日志系统

import json
from datetime import datetime

class AuditLogger:
def __init__(self, storage_backend: StorageBackend):
self.storage = storage_backend
self.buffer = []

async def log(self, event: AuditEvent):
log_entry = {
‘timestamp’: datetime.utcnow().isoformat(),
‘event_type’: event.type,
‘user_id’: event.user_id,
‘session_id’: event.session_id,
‘action’: event.action,
‘input_hash’: self.hash_sensitive(event.input),
‘output_hash’: self.hash_sensitive(event.output),
‘tools_used’: event.tools_used,
‘tokens_consumed’: event.token_count,
‘cost_usd’: event.cost,
‘risk_score’: event.risk_assessment,
‘metadata’: event.metadata
}

# 异步写入存储
await self.storage.append(log_entry)

Prometheus 告警规则

groups:

• name: ai_automation_alerts

rules:

• alert: HighErrorRate

expr: rate(ai_request_errors_total[5m]) > 0.05
for: 2m
labels:
severity: critical
annotations:
summary: “AI 系统错误率超过 5%”

• alert: HighLatency

expr: histogram_quantile(0.99, rate(ai_request_latency_bucket[5m])) > 5000
for: 5m
labels:
severity: warning
annotations:
summary: “P99 延迟超过 5 秒”

• alert: CostSpike

// 1. 安全检查
const securityCheck = await this.securityFilter.validate(message.content);
if (!securityCheck.passed) {
return this.createBlockedResponse(securityCheck.reason);
}

// 2. 上下文构建
const context = await this.memoryManager.buildContext({
sessionId: session.id,
userId: message.userId,
recentMessages: session.history.slice(-10),
relevantMemories: await this.vectorStore.search(message.content, 5)
});

// 3. LLM 推理
const llmResponse = await this.llmGateway.generate({
model: this.selectModel(message.complexity),
messages: [
{ role: ‘system’, content: this.systemPrompt },
…context,
{ role: ‘user’, content: message.content }
],
tools: this.toolRegistry.getAvailableTools(session.capabilities),
temperature: this.calculateTemperature(message.type)
});

// 4. 工具执行（如果有）
if (llmResponse.toolCalls) {
const toolResults = await this.executeTools(llmResponse.toolCalls, session);
return await this.processToolResults(toolResults, session);
}

// 5. 响应生成
await this.auditLogger.log({
type: ‘message_processed’,
sessionId: session.id,
tokensUsed: llmResponse.usage.total_tokens,
latency: llmResponse.latency
});