🚀 企业级 AI Agent 系统级 Agent

from typing import TypedDict, List, Optional, Dict, Any from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.postgres import PostgresSaver from enum import Enum from datetime import datetime from pydantic import BaseModel, Field # ========== 枚举定义 ========= class TicketPriority(Enum): """工单优先级""" P0 = "P0_CRITICAL" # 严重故障，系统不可用，15 分钟响应 P1 = "P1_HIGH" # 高优先级，核心功能受损，2 小时响应 P2 = "P2_MEDIUM" # 中优先级，部分功能受损，24 小时响应 P3 = "P3_LOW" # 低优先级，轻微问题，72 小时响应 class TicketStatus(Enum): """工单状态""" NEW = "NEW" # 新建 ANALYZING = "ANALYZING" # 分析中 PLANNING = "PLANNING" # 方案制定中 DISPATCHED = "DISPATCHED" # 已分派 IN_PROGRESS = "IN_PROGRESS" # 处理中 PENDING_REVIEW = "PENDING_REVIEW" # 待验证 VERIFIED = "VERIFIED" # 已验证 CLOSED = "CLOSED" # 已关闭 REOPENED = "REOPENED" # 重新打开 class Region(Enum): """区域标识""" DOMESTIC = "domestic" OVERSEAS = "overseas" # ========== 数据模型 ========= class Ticket(BaseModel): """工单数据模型""" id: str = Field(..., description="工单唯一标识") ticket_number: str = Field(..., description="工单号") title: str = Field(..., min_length=5, max_length=200) description: str = Field(..., min_length=20) priority: TicketPriority status: TicketStatus region: Region # 报告人信息 reporter_id: str reporter_type: str # user/system/im reporter_contact: Dict[str, str] = {} # 分析结果 root_cause: Optional[str] = None affected_services: List[str] = [] similar_cases: List[Dict[str, Any]] = [] log_analysis: Optional[Dict[str, Any]] = None # 处理方案 solution_plan: Optional[Dict[str, Any]] = None assigned_to: Optional[str] = None assigned_team: Optional[str] = None estimated_minutes: Optional[int] = None sla_deadline: Optional[datetime] = None # 时间戳 created_at: datetime = Field(default_factory=datetime.utcnow) updated_at: datetime = Field(default_factory=datetime.utcnow) resolved_at: Optional[datetime] = None closed_at: Optional[datetime] = None # ========== LangGraph 状态定义 ========= class SystemAgentState(TypedDict): """系统级 Agent 状态定义""" # 当前工单 ticket: Ticket # 分析上下文 logs_context: Optional[Dict[str, Any]] metrics_context: Optional[Dict[str, Any]] # 处理过程 progress_updates: List[Dict[str, Any]] handler_feedback: Optional[str] # 验证结果 verification_result: Optional[bool] verification_details: Optional[Dict[str, Any]] closure_notes: Optional[str] # 消息历史（用于流式输出） messages: List[Dict[str, str]]

from langchain.chat_models import init_chat_model from langchain.prompts import ChatPromptTemplate import json # ========== 初始化 LLM ========= llm = init_chat_model( "anthropic:claude-sonnet-4-20250514", temperature=0.3 ) # ========== 分析 Agent ========= ANALYZE_PROMPT = ChatPromptTemplate.from_messages([ ("system", """你是资深系统运维专家。请分析以下工单和日志信息： 任务： 1. 识别异常模式和错误类型 2. 推断可能的根本原因 3. 评估影响的服务范围 4. 检索相似历史案例 5. 给出紧急处理建议 请以 JSON 格式输出分析结果。"""), ("user", """工单信息： 标题：{title} 描述：{description} 优先级：{priority} 日志摘要： {log_summary} 相关指标： {metrics} 历史相似案例： {similar_cases}""") ]) async def analyze_ticket_node(state: SystemAgentState) -> Dict: """分析工单节点""" ticket = state["ticket"] # 获取日志上下文 logs_context = await fetch_logs_for_ticket(ticket.id) metrics_context = await fetch_metrics_for_services(ticket.affected_services) similar_cases = await search_similar_cases(ticket.description) # 调用 LLM 分析 response = await ANALYZE_PROMPT.ainvoke({ "title": ticket.title, "description": ticket.description, "priority": ticket.priority.value, "log_summary": json.dumps(logs_context, ensure_ascii=False), "metrics": json.dumps(metrics_context, ensure_ascii=False), "similar_cases": json.dumps(similar_cases, ensure_ascii=False) }) analysis_result = parse_analysis_response(response.content) return { "ticket": ticket.model_copy(update={ "root_cause": analysis_result["root_cause"], "affected_services": analysis_result["affected_services"], "similar_cases": analysis_result["similar_cases"], "log_analysis": analysis_result, "status": TicketStatus.ANALYZING }), "logs_context": logs_context, "metrics_context": metrics_context, "messages": [{"role": "assistant", "content": "✅ 分析完成，已识别根因"}] } # ========== 规划 Agent ========= PLAN_PROMPT = ChatPromptTemplate.from_messages([ ("system", """你是技术负责人。请基于分析结果制定详细处理方案： 方案内容： 1. 具体处理步骤（逐步列出） 2. 所需技能和资源 3. 风险评估和回滚方案 4. 时间估算 5. 建议的处理人员/团队 请以 JSON 格式输出。"""), ("user", """根因分析： {root_cause} 影响服务： {affected_services} 相似案例解决方案： {similar_solutions}""") ]) async def plan_solution_node(state: SystemAgentState) -> Dict: """制定处理方案节点""" ticket = state["ticket"] # 获取相似案例的解决方案 similar_solutions = await fetch_solutions_from_cases(ticket.similar_cases) response = await PLAN_PROMPT.ainvoke({ "root_cause": ticket.root_cause, "affected_services": ", ".join(ticket.affected_services), "similar_solutions": json.dumps(similar_solutions, ensure_ascii=False) }) plan_result = parse_plan_response(response.content) # 计算 SLA 截止时间 sla_hours = get_sla_hours(ticket.priority) sla_deadline = datetime.utcnow() + timedelta(hours=sla_hours) return { "ticket": ticket.model_copy(update={ "solution_plan": plan_result, "estimated_minutes": plan_result["estimated_minutes"], "sla_deadline": sla_deadline, "status": TicketStatus.PLANNING }), "messages": [{"role": "assistant", "content": "📋 方案已制定，准备分派任务"}] } # ========== 分发 Agent ========= async def dispatch_task_node(state: SystemAgentState) -> Dict: """任务分发节点""" ticket = state["ticket"] plan = ticket.solution_plan # 技能匹配 required_skills = plan.get("required_skills", []) candidates = await find_available_handlers( skills=required_skills, region=ticket.region, priority=ticket.priority ) # 选择最佳处理人 best_handler = select_best_handler(candidates, ticket.priority) # 发送通知 await send_assignment_notification( handler_id=best_handler.id, ticket=ticket, channel=get_notification_channel(ticket.region, best_handler) ) return { "ticket": ticket.model_copy(update={ "assigned_to": best_handler.id, "assigned_team": best_handler.team, "status": TicketStatus.DISPATCHED }), "messages": [{"role": "assistant", "content": f"📤 已分派给 {best_handler.name}"}] } # ========== 跟踪 Agent ========= async def track_progress_node(state: SystemAgentState) -> Dict: """进度跟踪节点""" ticket = state["ticket"] # 检查是否超时 if is_overdue(ticket.sla_deadline): await escalate_ticket(ticket.id) return { "messages": [{"role": "assistant", "content": "⚠️ SLA 超时，已升级"}] } # 获取最新进度 progress = await fetch_handler_progress(ticket.id) return { "progress_updates": state["progress_updates"] + [progress], "handler_feedback": progress.get("feedback"), "messages": [{"role": "assistant", "content": f"📊 进度更新：{progress.get('percentage')}%"] } # ========== 验证 Agent ========= VERIFY_PROMPT = ChatPromptTemplate.from_messages([ ("system", """请验证修复是否有效： 验证项： 1. 自动化测试结果 2. 监控指标恢复情况 3. 用户反馈（如有） 给出是否可以通过验证的结论。"""), ("user", """修复说明： {fix_description} 测试结果： {test_results} 监控指标： {metrics_after}""") ]) async def verify_fix_node(state: SystemAgentState) -> Dict: """修复验证节点""" ticket = state["ticket"] # 运行自动化测试 test_results = await run_automated_tests(ticket.affected_services) # 检查监控指标 metrics_after = await fetch_metrics_after_fix(ticket.affected_services) response = await VERIFY_PROMPT.ainvoke({ "fix_description": ticket.solution_plan.get("fix_description"), "test_results": json.dumps(test_results), "metrics_after": json.dumps(metrics_after) }) verification = parse_verification_response(response.content) return { "ticket": ticket.model_copy(update={ "status": TicketStatus.VERIFIED if verification["passed"] else TicketStatus.IN_PROGRESS, "resolved_at": datetime.utcnow() if verification["passed"] else None }), "verification_result": verification["passed"], "verification_details": verification, "messages": [{"role": "assistant", "content": "✅ 验证通过" if verification["passed"] else "❌ 验证未通过，需重新修复"] } # ========== 构建状态图 ========= def create_system_agent_graph(): workflow = StateGraph(SystemAgentState) # 添加节点 workflow.add_node("analyze", analyze_ticket_node) workflow.add_node("plan", plan_solution_node) workflow.add_node("dispatch", dispatch_task_node) workflow.add_node("track", track_progress_node) workflow.add_node("verify", verify_fix_node) # 定义边 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) # 编译（带持久化） connection_string = "postgresql://user:pass@localhost:5432/system_agent" memory = PostgresSaver.from_conn_string(connection_string) return workflow.compile(checkpointer=memory) # 创建 Agent 应用 app = create_system_agent_graph()

from fastapi import APIRouter, HTTPException, status from pydantic import BaseModel, Field from typing import Optional, List router = APIRouter(prefix="/api/v1/tickets", tags=["工单管理"]) class CreateTicketRequest(BaseModel): title: str = Field(..., min_length=5, max_length=200) description: str = Field(..., min_length=20) priority: str = Field(..., pattern="^(P0|P1|P2|P3)$") region: str = Field(..., pattern="^(domestic|overseas)$") reporter_contact: Dict[str, str] = {} attachments: List[str] = [] class TicketResponse(BaseModel): id: str ticket_number: str title: str status: str priority: str assigned_to: Optional[str] created_at: datetime sla_deadline: Optional[datetime] @router.post("", response_model=TicketResponse, status_code=status.HTTP_201_CREATED) async def create_ticket(request: CreateTicketRequest): """创建新工单""" ticket = await ticket_service.create(request) # 触发系统级 Agent 处理流程 await system_agent_app.ainvoke({ "ticket": ticket, "messages": [] }) return ticket @router.get("/{ticket_id}") async def get_ticket(ticket_id: str): """获取工单详情""" ticket = await ticket_service.get_by_id(ticket_id) if not ticket: raise HTTPException(status_code=404, detail="工单不存在") return ticket @router.put("/{ticket_id}/status") async def update_ticket_status(ticket_id: str, status_update: StatusUpdateRequest): """更新工单状态""" ticket = await ticket_service.update_status(ticket_id, status_update) # 通过 WebSocket 推送状态更新 await websocket_manager.broadcast({ "type": "ticket_status_update", "ticket_id": ticket_id, "new_status": ticket.status }) return ticket

-- ========== 工单表 ========= CREATE TABLE tickets ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), ticket_number VARCHAR(30) 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(30) NOT NULL DEFAULT 'NEW', region VARCHAR(20) NOT NULL CHECK (region IN ('domestic', 'overseas')), -- 报告人信息 reporter_id VARCHAR(100) NOT NULL, reporter_type VARCHAR(20) NOT NULL, reporter_contact JSONB DEFAULT '{}', -- 分析结果 root_cause TEXT, affected_services TEXT[] DEFAULT '{}', similar_case_ids UUID[] DEFAULT '{}', log_analysis JSONB, -- 处理方案 solution_plan JSONB, assigned_to VARCHAR(100), assigned_team VARCHAR(100), estimated_minutes INTEGER, sla_deadline TIMESTAMPTZ, -- 时间戳 created_at TIMESTAMPTZ DEFAULT NOW(), updated_at TIMESTAMPTZ DEFAULT NOW(), resolved_at TIMESTAMPTZ, closed_at TIMESTAMPTZ, -- 索引 INDEX idx_tickets_status (status), INDEX idx_tickets_priority (priority), INDEX idx_tickets_assigned (assigned_to), INDEX idx_tickets_created (created_at DESC), INDEX idx_tickets_region (region) ); -- ========== 处理进度表 ========= CREATE TABLE ticket_progress ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), ticket_id UUID NOT NULL REFERENCES tickets(id) ON DELETE CASCADE, handler_id VARCHAR(100) NOT NULL, progress_percentage INTEGER NOT NULL CHECK (progress_percentage BETWEEN 0 AND 100), feedback_text TEXT, attachments JSONB DEFAULT '[]', created_at TIMESTAMPTZ DEFAULT NOW(), INDEX idx_progress_ticket (ticket_id), INDEX idx_progress_handler (handler_id), INDEX idx_progress_created (created_at DESC) ); -- ========== 技能图谱表 ========= CREATE TABLE skill_profiles ( user_id VARCHAR(100) PRIMARY KEY, username VARCHAR(100) NOT NULL, skills JSONB NOT NULL, expertise_levels JSONB NOT NULL, current_load INTEGER DEFAULT 0, max_load INTEGER DEFAULT 10, available_hours JSONB, timezone VARCHAR(50) DEFAULT 'UTC', region VARCHAR(20) DEFAULT 'domestic', updated_at TIMESTAMPTZ DEFAULT NOW(), 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 TEXT[] DEFAULT '{}', tags TEXT[] DEFAULT '{}', priority VARCHAR(10), region VARCHAR(20), -- 向量嵌入（用于语义检索） embedding vector(1536), created_at TIMESTAMPTZ DEFAULT NOW(), INDEX idx_cases_tags USING GIN (tags), INDEX idx_cases_embedding USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100) ); -- ========== 通知记录表 ========= CREATE TABLE notifications ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), user_id VARCHAR(100) NOT NULL, ticket_id UUID REFERENCES tickets(id), type VARCHAR(30) NOT NULL, channel VARCHAR(30) NOT NULL, title VARCHAR(200) NOT NULL, content TEXT NOT NULL, is_read BOOLEAN DEFAULT FALSE, sent_at TIMESTAMPTZ DEFAULT NOW(), read_at TIMESTAMPTZ, INDEX idx_notifications_user (user_id), INDEX idx_notifications_unread (user_id, is_read) WHERE is_read = FALSE ); -- ========== Agent 执行日志表 ========= CREATE TABLE agent_execution_logs ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), ticket_id UUID NOT NULL, agent_name VARCHAR(50) NOT NULL, action VARCHAR(50) NOT NULL, input_data JSONB, output_data JSONB, execution_time_ms INTEGER, status VARCHAR(20) DEFAULT 'success', error_message TEXT, created_at TIMESTAMPTZ DEFAULT NOW(), INDEX idx_agent_logs_ticket (ticket_id), INDEX idx_agent_logs_agent (agent_name), INDEX idx_agent_logs_created (created_at DESC) );

src/ ├── components/ │ ├── common/ # 通用组件 │ │ ├── Header.tsx │ │ ├── Sidebar.tsx │ │ ├── Loading.tsx │ │ └── ErrorBoundary.tsx │ ├── tickets/ # 工单相关组件 │ │ ├── TicketList.tsx │ │ ├── TicketDetail.tsx │ │ ├── TicketForm.tsx │ │ └── TicketTimeline.tsx │ ├── dashboard/ # 仪表盘组件 │ │ ├── MetricsCard.tsx │ │ ├── ChartPanel.tsx │ │ └── AlertPanel.tsx │ └── agents/ # Agent 状态组件 │ ├── AgentStatus.tsx │ └── AgentLog.tsx ├── hooks/ │ ├── useWebSocket.ts # WebSocket Hook │ ├── useTickets.ts # 工单数据 Hook │ └── useNotifications.ts ├── services/ │ ├── api.ts # API 客户端 │ ├── websocket.ts # WebSocket 服务 │ └── auth.ts # 认证服务 ├── store/ │ ├── index.ts # Redux store │ ├── slices/ │ │ ├── ticketSlice.ts │ │ ├── notificationSlice.ts │ │ └── userSlice.ts ├── types/ │ ├── ticket.ts │ ├── user.ts │ └── api.ts ├── utils/ │ ├── formatters.ts │ └── validators.ts ├── pages/ │ ├── Dashboard.tsx │ ├── Tickets.tsx │ ├── Analytics.tsx │ └── Settings.tsx └── App.tsx

import { useEffect, useRef, useCallback, useState } from 'react'; interface WebSocketMessage { type: string; payload: any; timestamp: string; } export function useWebSocket(userId: string) { const wsRef = useRef(null); const [isConnected, setIsConnected] = useState(false); const [messages, setMessages] = useState([]); const connect = useCallback(() => { const wsUrl = `ws://${window.location.host}/ws/${userId}`; wsRef.current = new WebSocket(wsUrl); wsRef.current.onopen = () => { console.log('WebSocket connected'); setIsConnected(true); }; wsRef.current.onmessage = (event) => { const message = JSON.parse(event.data); setMessages(prev => [...prev, message]); // 根据消息类型处理 handleMessage(message); }; wsRef.current.onclose = () => { console.log('WebSocket disconnected'); setIsConnected(false); // 3 秒后重连 setTimeout(connect, 3000); }; wsRef.current.onerror = (error) => { console.error('WebSocket error:', error); }; }, [userId]); const sendMessage = useCallback((data: any) => { if (wsRef.current && isConnected) { wsRef.current.send(JSON.stringify(data)); } }, [isConnected]); useEffect(() => { connect(); return () => { wsRef.current?.close(); }; }, [connect]); return { isConnected, messages, sendMessage }; }

import React, { useEffect, useState } from 'react'; import { useTickets } from '../hooks/useTickets'; import { useWebSocket } from '../hooks/useWebSocket'; export const TicketList: React.FC = () => { const { tickets, loading, error } = useTickets(); const { isConnected, messages } = useWebSocket(currentUserId); const [filter, setFilter] = useState({ status: 'all', priority: 'all' }); // 处理 WebSocket 实时更新 useEffect(() => { messages.forEach(msg => { if (msg.type === 'ticket_update') { // 更新本地工单状态 updateTicketInCache(msg.payload); } }); }, [messages]); const filteredTickets = tickets.filter(ticket => { if (filter.status !== 'all' && ticket.status !== filter.status) return false; if (filter.priority !== 'all' && ticket.priority !== filter.priority) return false; return true; }); if (loading) return ; if (error) return ; return (

setFilter({...filter, status: e.target.value})}> 全部状态 新建 处理中 已关闭 {isConnected ? '🟢 实时连接' : '🔴 断开'}

{filteredTickets.map(ticket => ( ))}

工单号	标题	优先级	状态	处理人	创建时间	SLA 截止

); };

from fastapi import FastAPI, WebSocket, WebSocketDisconnect from typing import Dict, List, Set import json import asyncio app = FastAPI() class ConnectionManager: """WebSocket 连接管理器""" def __init__(self): # user_id -> [WebSocket connections] self.active_connections: Dict[str, List[WebSocket]] = {} # ticket_id -> set of user_ids (订阅该工单的用户) self.ticket_subscribers: Dict[str, Set[str]] = {} 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) # 发送欢迎消息 await self.send_personal_message({ "type": "connected", "message": "✅ 已连接到实时通知系统" }, user_id) async def disconnect(self, websocket: WebSocket, user_id: str): if user_id in self.active_connections: 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]: try: await connection.send_json(message) except: pass # 连接已断开，忽略 async def broadcast_to_ticket_subscribers(self, ticket_id: str, message: dict): """广播给订阅某工单的所有用户""" if ticket_id in self.ticket_subscribers: for user_id in self.ticket_subscribers[ticket_id]: await self.send_personal_message(message, user_id) async def subscribe_to_ticket(self, user_id: str, ticket_id: str): """订阅工单更新""" if ticket_id not in self.ticket_subscribers: self.ticket_subscribers[ticket_id] = set() self.ticket_subscribers[ticket_id].add(user_id) 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) # 处理客户端消息 if message["type"] == "subscribe_ticket": await manager.subscribe_to_ticket(user_id, message["ticket_id"]) elif message["type"] == "update_progress": await handle_progress_update(message) except WebSocketDisconnect: await manager.disconnect(websocket, user_id) # ========== 后台任务：推送工单更新 ========= async def push_ticket_update(ticket_id: str, update_data: dict): """推送工单更新给所有订阅者""" await manager.broadcast_to_ticket_subscribers(ticket_id, { "type": "ticket_update", "ticket_id": ticket_id, "payload": update_data, "timestamp": datetime.utcnow().isoformat() })

from langchain.callbacks import AsyncIteratorCallbackHandler import asyncio async def stream_agent_response(ticket_id: str, user_id: str): """流式推送 Agent 处理过程""" callback = AsyncIteratorCallbackHandler() async def agent_task(): return await system_agent_app.ainvoke( {"ticket": ticket, "messages": []}, config={"callbacks": [callback]} ) # 启动 Agent 任务 task = asyncio.create_task(agent_task()) # 流式输出 async for chunk in callback.aiter(): await manager.send_personal_message({ "type": "agent_stream", "ticket_id": ticket_id, "content": chunk }, user_id) await task

from sklearn.ensemble import IsolationForest import numpy as np class LogAnomalyDetector: """日志异常检测器""" def __init__(self): self.model = IsolationForest( contamination=0.1, random_state=42 ) self.is_fitted = False def extract_features(self, log_entry: dict) -> np.ndarray: """从日志条目提取特征""" features = [ log_entry.get("response_time", 0), log_entry.get("status_code", 200), log_entry.get("error_count", 0), log_entry.get("request_size", 0), # ... 更多特征 ] return np.array(features).reshape(1, -1) def detect(self, log_entries: List[dict]) -> List[dict]: """检测异常日志""" features = np.vstack([self.extract_features(entry) for entry in log_entries]) if not self.is_fitted: self.model.fit(features) self.is_fitted = True predictions = self.model.predict(features) # -1 表示异常，1 表示正常 anomalies = [ {**log_entries[i], "anomaly_score": score} for i, pred in enumerate(predictions) if pred == -1 ] return anomalies # 使用示例 detector = LogAnomalyDetector() anomalies = detector.detect(recent_logs) for anomaly in anomalies: await create_alert(anomaly)

version: '3.8' services: # API 网关 api-gateway: build: ./services/api-gateway ports: - "8000:8000" environment: - AUTH_SERVICE_URL=http://auth-service:8001 - RATE_LIMIT_ENABLED=true depends_on: - auth-service deploy: replicas: 3 # Agent 核心服务 agent-core: build: ./services/agent-core ports: - "8001:8001" 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: 2 # PostgreSQL db: image: postgres:15 volumes: - postgres_data:/var/lib/postgresql/data environment: - POSTGRES_DB=system_agent - POSTGRES_USER=user - POSTGRES_PASSWORD=${DB_PASSWORD} healthcheck: test: ["CMD-SHELL", "pg_isready -U user"] interval: 10s timeout: 5s retries: 5 # Redis redis: image: redis:7-alpine command: redis-server --appendonly yes volumes: - redis_data:/data healthcheck: test: ["CMD", "redis-cli", "ping"] interval: 10s timeout: 5s retries: 5 # Elasticsearch elasticsearch: image: elasticsearch:8.11.0 environment: - discovery.type=single-node - xpack.security.enabled=false - "ES_JAVA_OPTS=-Xms1g -Xmx1g" volumes: - es_data:/usr/share/elasticsearch/data ulimits: memlock: soft: -1 hard: -1 # WebSocket Hub websocket-hub: build: ./services/websocket-hub ports: - "8005:8005" environment: - REDIS_URL=redis://redis:6379 depends_on: - redis volumes: postgres_data: redis_data: es_data: