🚀 企业级 AI Agent 系统级 Agent 架构设计

from typing import TypedDict, List, Optional from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import InMemorySaver from enum import Enum class TicketPriority(Enum): P0 = "P0_CRITICAL" # 严重故障，立即处理 P1 = "P1_HIGH" # 高优先级，24 小时内 P2 = "P2_MEDIUM" # 中优先级，3 天内 P3 = "P3_LOW" # 低优先级，1 周内 class TicketStatus(Enum): NEW = "NEW" ANALYZING = "ANALYZING" PLANNING = "PLANNING" DISPATCHED = "DISPATCHED" IN_PROGRESS = "IN_PROGRESS" VERIFYING = "VERIFYING" CLOSED = "CLOSED" class SystemAgentState(TypedDict): # 问题工单信息 ticket_id: str title: str description: str priority: TicketPriority status: TicketStatus region: str # domestic/overseas # 分析结果 root_cause: Optional[str] affected_services: List[str] similar_cases: List[dict] # 处理方案 solution_plan: Optional[dict] assigned_to: Optional[str] estimated_time: Optional[int] # 分钟 # 处理过程 progress_updates: List[dict] handler_feedback: Optional[str] # 验证结果 verification_result: Optional[bool] closure_notes: Optional[str] # 定义处理节点 async def analyze_ticket(state: SystemAgentState): """分析工单，识别根因""" # 调用日志分析 Agent # 更新 state['root_cause'], state['similar_cases'] return {"status": TicketStatus.ANALYZING} async def create_solution(state: SystemAgentState): """基于分析结果制定处理方案""" # 调用规划 Agent + RAG 检索 # 更新 state['solution_plan'], state['estimated_time'] return {"status": TicketStatus.PLANNING} async def dispatch_task(state: SystemAgentState): """智能分派任务给合适人员""" # 技能匹配 + 负载均衡 # 更新 state['assigned_to'], state['status'] return {"status": TicketStatus.DISPATCHED} async def track_progress(state: SystemAgentState): """跟踪处理进度""" # WebSocket 推送进度更新 return {"status": TicketStatus.IN_PROGRESS} async def verify_fix(state: SystemAgentState): """验证修复效果""" # 自动化测试 + 人工确认 # 更新 state['verification_result'] return {"status": TicketStatus.VERIFYING} # 构建状态图 workflow = StateGraph(SystemAgentState) # 添加节点 workflow.add_node("analyze", analyze_ticket) workflow.add_node("plan", create_solution) workflow.add_node("dispatch", dispatch_task) workflow.add_node("track", track_progress) workflow.add_node("verify", verify_fix) # 定义边 workflow.add_edge(START, "analyze") workflow.add_edge("analyze", "plan") workflow.add_edge("plan", "dispatch") workflow.add_edge("dispatch", "track") workflow.add_edge("track", "verify") workflow.add_edge("verify", END) # 编译带检查点的图 memory = InMemorySaver() app = workflow.compile(checkpointer=memory)

from langchain.chat_models import init_chat_model from langchain.prompts import ChatPromptTemplate # 日志分析 Agent 提示词模板 LOG_ANALYSIS_PROMPT = ChatPromptTemplate.from_messages([ ("system", """你是一个资深系统运维专家。请分析以下日志片段，识别： 1. 异常模式和错误类型 2. 可能的根本原因 3. 影响的服务范围 4. 建议的紧急处理措施 5. 长期优化建议 请使用结构化格式输出。"""), ("user", """日志内容： {log_content} 时间范围：{time_range} 服务名称：{service_name} 历史相似问题：{similar_cases}""") ]) # 初始化分析模型 llm = init_chat_model("anthropic:claude-sonnet-4-20250514", temperature=0.3) analysis_chain = LOG_ANALYSIS_PROMPT | llm async def analyze_log_stream(log_batch: dict) -> dict: """分析日志批次，生成建议""" response = await analysis_chain.ainvoke({ "log_content": log_batch["logs"], "time_range": log_batch["time_range"], "service_name": log_batch["service"], "similar_cases": await fetch_similar_cases(log_batch) }) return { "anomalies": extract_anomalies(response), "suggestions": extract_suggestions(response), "priority": calculate_priority(response) }

from pydantic import BaseModel, Field from enum import Enum class Region(Enum): DOMESTIC = "domestic" OVERSEAS = "overseas" class BugReport(BaseModel): ticket_id: str = Field(..., description="工单唯一标识") region: Region = Field(..., description="反馈来源区域") reporter_id: str = Field(..., description="报告人 ID") title: str = Field(..., min_length=5, max_length=200) description: str = Field(..., min_length=20) severity: str = Field(..., pattern="^(P0|P1|P2|P3)$") # 国内特有字段 phone_number: Optional[str] = Field(None, pattern="^1[3-9]\\d{9}$") wechat_id: Optional[str] = None dingtalk_id: Optional[str] = None # 国际特有字段 email: Optional[str] = Field(None, pattern="^[^@]+@[^@]+\\.[^@]+$") slack_user_id: Optional[str] = None # 通用字段 attachments: List[str] = [] environment: dict = {} # 环境信息：OS、浏览器、版本号等 steps_to_reproduce: List[str] = [] expected_behavior: str actual_behavior: str class Config: use_enum_values = True class NotificationService: async def send_notification(self, region: Region, **kwargs): """根据区域选择通知服务""" if region == Region.DOMESTIC: await self._send_via_wechat(**kwargs) await self._send_via_dingtalk(**kwargs) else: await self._send_via_email(**kwargs) await self._send_via_slack(**kwargs)

from fastapi import FastAPI, WebSocket, WebSocketDisconnect from typing import Dict, List import json app = FastAPI() class ConnectionManager: def __init__(self): self.active_connections: Dict[str, List[WebSocket]] = {} async def connect(self, websocket: WebSocket, user_id: str): await websocket.accept() if user_id not in self.active_connections: self.active_connections[user_id] = [] self.active_connections[user_id].append(websocket) async def disconnect(self, websocket: WebSocket, user_id: str): self.active_connections[user_id].remove(websocket) if not self.active_connections[user_id]: del self.active_connections[user_id] async def send_personal_message(self, message: dict, user_id: str): if user_id in self.active_connections: for connection in self.active_connections[user_id]: await connection.send_json(message) async def broadcast(self, message: dict): for connections in self.active_connections.values(): for connection in connections: await connection.send_json(message) manager = ConnectionManager() @app.websocket("/ws/{user_id}") async def websocket_endpoint(websocket: WebSocket, user_id: str): await manager.connect(websocket, user_id) try: while True: data = await websocket.receive_text() # 处理客户端消息 message = json.loads(data) await handle_client_message(message, user_id) except WebSocketDisconnect: await manager.disconnect(websocket, user_id) async def stream_ticket_updates(ticket_id: str): """流式推送工单状态更新""" async for update in ticket_update_generator(ticket_id): await manager.broadcast({ "type": "ticket_update", "ticket_id": ticket_id, "update": update })

-- 问题工单表 CREATE TABLE tickets ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), ticket_number VARCHAR(20) UNIQUE NOT NULL, title VARCHAR(200) NOT NULL, description TEXT NOT NULL, priority VARCHAR(10) NOT NULL CHECK (priority IN ('P0', 'P1', 'P2', 'P3')), status VARCHAR(20) NOT NULL DEFAULT 'NEW', region VARCHAR(20) NOT NULL CHECK (region IN ('domestic', 'overseas')), # 报告人信息 reporter_id UUID NOT NULL, reporter_type VARCHAR(20) NOT NULL, reporter_contact JSONB, # 分析结果 root_cause TEXT, affected_services JSONB, similar_case_ids UUID[], # 处理方案 solution_plan JSONB, assigned_to UUID, estimated_minutes INTEGER, # 时间戳 created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), resolved_at TIMESTAMP WITH TIME ZONE, closed_at TIMESTAMP WITH TIME ZONE, # 索引 INDEX idx_tickets_status (status), INDEX idx_tickets_priority (priority), INDEX idx_tickets_assigned (assigned_to), INDEX idx_tickets_created (created_at) ); -- 处理进度表 CREATE TABLE ticket_progress ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), ticket_id UUID NOT NULL REFERENCES tickets(id), handler_id UUID NOT NULL, progress_percentage INTEGER NOT NULL CHECK (progress_percentage BETWEEN 0 AND 100), feedback_text TEXT, attachments JSONB, created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), INDEX idx_progress_ticket (ticket_id), INDEX idx_progress_handler (handler_id) ); -- 技能图谱表 CREATE TABLE skill_profiles ( user_id UUID PRIMARY KEY, skills JSONB NOT NULL, expertise_levels JSONB, current_load INTEGER DEFAULT 0, max_load INTEGER DEFAULT 10, available_hours JSONB, timezone VARCHAR(50) DEFAULT 'UTC', INDEX idx_skills ((skills->>'categories')) ); -- 历史案例库 CREATE TABLE historical_cases ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), title VARCHAR(200) NOT NULL, description TEXT NOT NULL, root_cause TEXT NOT NULL, solution TEXT NOT NULL, resolution_time_minutes INTEGER, services_affected JSONB, tags TEXT[], embedding vector(1536), # 用于语义检索 created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), INDEX idx_cases_tags USING GIN (tags), INDEX idx_cases_embedding USING ivfflat (embedding vector_cosine_ops) );

# docker-compose.yml 核心配置 version: '3.8' services: # FastAPI 后端 api-server: build: ./backend ports: - "8000:8000" environment: - DATABASE_URL=postgresql://user:pass@db:5432/system_agent - REDIS_URL=redis://redis:6379 - LANGCHAIN_API_KEY=${LANGCHAIN_API_KEY} depends_on: - db - redis deploy: replicas: 3 # React 前端 frontend: build: ./frontend ports: - "3000:80" depends_on: - api-server # PostgreSQL 数据库 db: image: postgres:15 volumes: - postgres_data:/var/lib/postgresql/data environment: - POSTGRES_DB=system_agent - POSTGRES_USER=user - POSTGRES_PASSWORD=pass # Redis 缓存 redis: image: redis:7-alpine volumes: - redis_data:/data # Elasticsearch 日志 elasticsearch: image: elasticsearch:8.11.0 environment: - discovery.type=single-node - xpack.security.enabled=false volumes: - es_data:/usr/share/elasticsearch/data # 消息队列 rabbitmq: image: rabbitmq:3-management ports: - "15672:15672" volumes: - rabbitmq_data:/var/lib/rabbitmq volumes: postgres_data: redis_data: es_data: rabbitmq_data: