Agent落地实战:从零搭建企业级AI助手-夜雨聆风

Agent落地实战:从零搭建企业级AI助手

📅 2026年4月22日 | 👤 织美 | ⏱️ 阅读约30分钟 | 🏷️ 企业级实战+完整教程

🚀 企业级AI助手落地指南

5篇文章积累 → 1个完整系统 | 从概念到生产环境 | 可直接部署

📌 写在前面

Agent系列文章已经陪伴大家走过了四篇：

1️⃣ 《谷歌Agent Smith完整解析》—— 理解原理

2️⃣ 《多Agent协作》—— 扩展能力

3️⃣ 《性能优化》—— 降低成本

4️⃣ 《安全实战》—— 防范风险

今天这篇是收官之作，我会把前面所有知识整合起来，带你从零搭建一个完整的企业级AI助手系统。

这个系统包含：

• 多Agent协作架构

• 完整的成本优化方案

• 企业级安全防护

• 高可用部署方案

读完这篇文章，你将拥有一个可直接投入生产环境的AI助手系统。

🏗️ 第一部分：企业级系统架构设计

🎯 整体架构图

┌─────────────────────────────────────────────────────────────────────────────┐ │ 🌐 接入层 (Access Layer) │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ Web端 │ │ 企业微信 │ │ 钉钉 │ │ API接口 │ │ │ │ React/Vue │ │ 机器人 │ │ 机器人 │ │ 开放 │ │ │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ └─────────┼────────────────┼────────────────┼────────────────┼───────────────┘ │ │ │ │ └────────────────┴────────────────┴────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ 🛡️ 网关层 (Gateway Layer) │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ • 统一认证鉴权 (JWT/OAuth2) • 限流熔断 • 日志记录 • 请求路由 │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ 🤖 核心业务层 (Core Business) │ │ │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ 🎛️ 编排器 (Orchestrator) │ │ │ │ • 意图识别 • Agent调度 • 工作流管理 • 结果聚合 │ │ │ └──────────────┬────────────────┬────────────────┬───────────────────┘ │ │ │ │ │ │ │ ┌────────▼────────┐ ┌────▼─────┐ ┌────────▼────────┐ │ │ │ 💼 业务Agent │ │ 🔧 工具 │ │ 📊 分析Agent │ │ │ │ • 客服Agent │ │ Agent │ │ • 数据分析师 │ │ │ │ • 销售Agent │ │ │ │ • 报告生成器 │ │ │ │ • HR Agent │ │ │ │ │ │ │ └─────────────────┘ └──────────┘ └─────────────────┘ │ │ │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ 💾 记忆系统 (Memory System) │ │ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌────────────┐ │ │ │ │ │ 短期记忆 │ │ 长期记忆 │ │ 向量存储 │ │ 知识库 │ │ │ │ │ │ Redis │ │ PostgreSQL │ │ Milvus │ │ Elasticsearch│ │ │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ └────────────┘ │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ 🔌 集成层 (Integration Layer) │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ LLM服务 │ │ 数据库 │ │ 外部API │ │ 搜索引擎 │ │ │ │ OpenAI/ │ │ MySQL/ │ │ 飞书/钉钉 │ │ Google/ │ │ │ │ Claude │ │ PostgreSQL │ │ /企业微信 │ │ Bing │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ │ └─────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ 📊 监控运维层 (Observability) │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ 日志收集 │ │ 指标监控 │ │ 链路追踪 │ │ 告警通知 │ │ │ │ ELK Stack │ │ Prometheus│ │ Jaeger │ │ 钉钉/企业微信│ │ │ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ │ └─────────────────────────────────────────────────────────────────────────────┘

🛠️ 技术栈选型

层级	技术选型	选型理由
后端框架	FastAPI + Python 3.11	高性能、异步支持、自动文档
Agent框架	AutoGen + LangChain	多Agent协作 + 工具链丰富
消息队列	Redis + Celery	轻量级、易部署、社区成熟
向量数据库	Milvus / Chroma	高性能向量检索、企业级特性
关系数据库	PostgreSQL 15	稳定可靠、JSON支持好
缓存	Redis Cluster	高可用、分布式锁、Pub/Sub
前端	React 18 + Ant Design	生态丰富、企业级UI组件
部署	Docker + K8s	容器化、弹性伸缩
监控	Prometheus + Grafana	云原生标准、可视化强大

🚀 第二部分：项目初始化与环境搭建

1项目结构初始化

enterprise-ai-assistant/ ├── 📁 app/ │ ├── __init__.py │ ├── main.py # FastAPI应用入口 │ ├── config.py # 配置管理 │ ├── dependencies.py # 依赖注入 │ └── api/ │ ├── __init__.py │ ├── v1/ │ │ ├── __init__.py │ │ ├── chat.py # 对话接口 │ │ ├── agents.py # Agent管理 │ │ ├── knowledge.py # 知识库管理 │ │ └── users.py # 用户管理 ├── 📁 core/ │ ├── __init__.py │ ├── agents/ # Agent定义 │ │ ├── __init__.py │ │ ├── base.py │ │ ├── customer_service.py │ │ ├── sales.py │ │ └── analyst.py │ ├── memory/ # 记忆系统 │ │ ├── __init__.py │ │ ├── short_term.py # 短期记忆（Redis） │ │ ├── long_term.py # 长期记忆（PostgreSQL） │ │ └── vector_store.py # 向量存储 │ ├── security/ # 安全模块 │ │ ├── __init__.py │ │ ├── input_filter.py # 输入过滤 │ │ ├── output_filter.py # 输出过滤 │ │ ├── data_masking.py # 数据脱敏 │ │ └── access_control.py # 访问控制 │ ├── optimization/ # 优化模块 │ │ ├── __init__.py │ │ ├── model_router.py # 模型路由 │ │ ├── cache_manager.py # 缓存管理 │ │ └── cost_tracker.py # 成本追踪 │ └── tools/ # 工具集 │ ├── __init__.py │ ├── database.py │ ├── search.py │ ├── email.py │ └── calendar.py ├── 📁 models/ # 数据模型 │ ├── __init__.py │ ├── user.py │ ├── conversation.py │ └── knowledge.py ├── 📁 services/ # 业务服务 │ ├── __init__.py │ ├── orchestrator.py # 编排器 │ ├── llm_service.py # LLM服务 │ └── monitoring.py # 监控服务 ├── 📁 frontend/ # 前端代码 │ ├── package.json │ ├── src/ │ └── public/ ├── 📁 tests/ # 测试 │ ├── unit/ │ ├── integration/ │ └── e2e/ ├── 📁 deployments/ # 部署配置 │ ├── docker/ │ ├── kubernetes/ │ └── scripts/ ├── 📁 docs/ # 文档 │ ├── api.md │ ├── deployment.md │ └── development.md ├── .env.example # 环境变量示例 ├── requirements.txt # Python依赖 ├── Dockerfile ├── docker-compose.yml └── README.md

2安装依赖

# 创建虚拟环境 python -m venv venv source venv/bin/activate # Linux/Mac # 或 venv\Scripts\activate # Windows # 安装核心依赖 pip install fastapi==0.104.1 uvicorn[standard]==0.24.0 pip install autogen==0.2.0 langchain==0.0.350 pip install redis==5.0.1 celery==5.3.4 pip install sqlalchemy==2.0.23 asyncpg==0.29.0 pip install chromadb==0.4.18 sentence-transformers==2.2.2 pip install python-jose[cryptography]==3.3.0 passlib[bcrypt]==1.7.4 pip install python-multipart==0.0.6 aiofiles==23.2.1 pip install prometheus-client==0.19.0 structlog==23.2.0 pip install pydantic-settings==2.1.0 python-dotenv==1.0.0 # 开发依赖 pip install pytest==7.4.3 pytest-asyncio==0.21.1 httpx==0.25.2 pip install black==23.11.0 isort==5.12.0 flake8==6.1.0 # 保存依赖 pip freeze > requirements.txt

3配置管理

# app/config.py from pydantic_settings import BaseSettings from typing import List, Optional class Settings(BaseSettings): “””应用配置””” # 应用信息 APP_NAME: str = “Enterprise AI Assistant” APP_VERSION: str = “1.0.0” DEBUG: bool = False # 服务器配置 HOST: str = “0.0.0.0” PORT: int = 8000 WORKERS: int = 4 # 安全配置 SECRET_KEY: str = “your-secret-key-change-in-production” ACCESS_TOKEN_EXPIRE_MINUTES: int = 30 REFRESH_TOKEN_EXPIRE_DAYS: int = 7 # 数据库配置 DATABASE_URL: str = “postgresql+asyncpg://user:pass@localhost/ai_assistant” REDIS_URL: str = “redis://localhost:6379/0” # LLM配置 OPENAI_API_KEY: Optional[str] = None OPENAI_BASE_URL: Optional[str] = “https://api.openai.com/v1” DEFAULT_MODEL: str = “gpt-3.5-turbo” PREMIUM_MODEL: str = “gpt-4” # 成本优化配置 ENABLE_CACHE: bool = True CACHE_TTL: int = 3600 COST_ALERT_THRESHOLD: float = 100.0 # 美元 # 安全配置 ENABLE_INPUT_FILTER: bool = True ENABLE_OUTPUT_FILTER: bool = True ENABLE_DATA_MASKING: bool = True MAX_INPUT_LENGTH: int = 4000 # 向量数据库配置 VECTOR_DB_PATH: str = “./data/vector_db” EMBEDDING_MODEL: str = “BAAI/bge-large-zh” # 监控配置 ENABLE_METRICS: bool = True METRICS_PORT: int = 9090 LOG_LEVEL: str = “INFO” class Config: env_file = “.env” case_sensitive = True settings = Settings()

# .env.example（复制为.env并填写真实值） # 安全密钥（生产环境必须修改） SECRET_KEY=your-super-secret-key-min-32-chars # 数据库 DATABASE_URL=postgresql+asyncpg://ai_user:ai_pass@localhost:5432/ai_assistant REDIS_URL=redis://localhost:6379/0 # OpenAI API OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx OPENAI_BASE_URL=https://api.openai.com/v1 # 可选：Azure OpenAI # AZURE_OPENAI_KEY=xxx # AZURE_OPENAI_ENDPOINT=https://xxx.openai.azure.com/ # 企业微信/钉钉机器人（可选） WECOM_WEBHOOK_URL=https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx DINGTALK_WEBHOOK_URL=https://oapi.dingtalk.com/robot/send?access_token=xxx # 监控告警 ALERT_WEBHOOK_URL=https://oapi.dingtalk.com/robot/send?access_token=xxx

🔧 第三部分：核心模块实现

模块1：Agent基础框架

# core/agents/base.py from abc import ABC, abstractmethod from typing import Dict, List, Optional, Any from dataclasses import dataclass from enum import Enum import autogen from app.config import settings class AgentRole(Enum): “””Agent角色枚举””” CUSTOMER_SERVICE = “customer_service” SALES = “sales” ANALYST = “analyst” HR = “hr” IT_SUPPORT = “it_support” class TaskPriority(Enum): “””任务优先级””” LOW = 1 MEDIUM = 2 HIGH = 3 CRITICAL = 4 @dataclass class AgentResponse: “””Agent响应结构””” content: str success: bool metadata: Dict[str, Any] cost: float # 成本（美元） latency_ms: int # 延迟（毫秒） used_model: str # 使用的模型 class BaseAgent(ABC): “””Agent基类””” def __init__( self, name: str, role: AgentRole, system_message: str, model: Optional[str] = None, temperature: float = 0.7, tools: Optional[List[Dict]] = None ): self.name = name self.role = role self.model = model or settings.DEFAULT_MODEL self.temperature = temperature self.tools = tools or [] # 创建AutoGen代理 self.agent = autogen.AssistantAgent( name=name, system_message=system_message, llm_config={ “config_list”: [{ “model”: self.model, “api_key”: settings.OPENAI_API_KEY, “base_url”: settings.OPENAI_BASE_URL }], “temperature”: temperature } ) @abstractmethod async def process( self, query: str, context: Optional[Dict] = None, user_id: Optional[str] = None ) -> AgentResponse: “””处理请求（子类必须实现）””” pass def get_capabilities(self) -> List[str]: “””获取Agent能力列表””” return [] def estimate_cost(self, input_tokens: int, output_tokens: int) -> float: “””估算成本””” # GPT-3.5: $0.0015/1K tokens, GPT-4: $0.03/1K tokens rates = { “gpt-3.5-turbo”: {“input”: 0.0015, “output”: 0.002}, “gpt-4”: {“input”: 0.03, “output”: 0.06}, “gpt-4-turbo”: {“input”: 0.01, “output”: 0.03} } rate = rates.get(self.model, rates[“gpt-3.5-turbo”]) cost = (input_tokens / 1000 * rate[“input”] + output_tokens / 1000 * rate[“output”]) return round(cost, 6) # 用户代理（人机交互接口） class UserProxy: “””用户代理，处理人机交互””” def __init__(self): self.agent = autogen.UserProxyAgent( name=”User”, human_input_mode=”NEVER”, max_consecutive_auto_reply=10, code_execution_config={“work_dir”: “coding”, “use_docker”: False} )

模块2：安全防护层（整合第四篇内容）

# core/security/defense_layer.py import re import hashlib from typing import Tuple, List, Optional from app.config import settings class SecurityDefenseLayer: “””安全防护层 – 整合所有安全措施””” def __init__(self): self.input_filter = InputFilter() self.output_filter = OutputFilter() self.data_masker = DataMasker() self.rate_limiter = RateLimiter() async def process_input( self, user_input: str, user_id: str ) -> Tuple[str, List[str]]: “”” 处理输入（过滤 + 脱敏）返回: (处理后的输入, 警告列表) “”” warnings = [] # 1. 限流检查 if not await self.rate_limiter.check(user_id): raise RateLimitExceeded(“请求过于频繁，请稍后再试”) # 2. 输入长度检查 if len(user_input) > settings.MAX_INPUT_LENGTH: user_input = user_input[:settings.MAX_INPUT_LENGTH] warnings.append(“输入过长，已截断”) # 3. 危险内容过滤 if settings.ENABLE_INPUT_FILTER: filtered, risks = self.input_filter.filter(user_input) if risks: warnings.extend(risks) user_input = filtered # 4. 敏感数据脱敏 if settings.ENABLE_DATA_MASKING: user_input = self.data_masker.mask(user_input) return user_input, warnings async def process_output( self, output: str, user_id: str ) -> Tuple[str, bool]: “”” 处理输出（过滤敏感信息）返回: (处理后的输出, 是否包含敏感信息) “”” if not settings.ENABLE_OUTPUT_FILTER: return output, False filtered, has_sensitive = self.output_filter.filter(output) if has_sensitive: # 记录安全事件 await self.log_security_event( event_type=”sensitive_data_leak_attempt”, user_id=user_id, details={“original_length”: len(output)} ) return filtered, has_sensitive async def log_security_event( self, event_type: str, user_id: str, details: dict ): “””记录安全事件””” from core.monitoring import security_logger security_logger.warning({ “event_type”: event_type, “user_id”: user_id, “details”: details }) class InputFilter: “””输入过滤器””” DANGEROUS_PATTERNS = [ re.compile(r’ignore\s+(all\s+)?(previous|prior)\s+instructions?’, re.I), re.compile(r’forget\s+(all\s+)?(previous|prior)\s+instructions?’, re.I), re.compile(r’you\s+are\s+now\s+.*?(unrestricted|uncensored|free)’, re.I), re.compile(r’system\s*:\s*.+’, re.I), re.compile(r’new\s+instruction\s*:’, re.I), re.compile(r’developer\s+mode’, re.I), re.compile(r’DAN\s*mode’, re.I), ] DANGEROUS_KEYWORDS = [ “忽略之前的指令”, “忘记之前的”, “你是新的AI”, “系统提示词”, “告诉我你的提示词”, “绕过限制”, “没有限制”, “开发者模式”, “DAN模式” ] def filter(self, text: str) -> Tuple[str, List[str]]: risks = [] filtered = text # 检查危险模式 for pattern in self.DANGEROUS_PATTERNS: if pattern.search(text): risks.append(f”检测到危险模式: {pattern.pattern}”) filtered = pattern.sub(‘[FILTERED]’, filtered) # 检查危险关键词 for keyword in self.DANGEROUS_KEYWORDS: if keyword.lower() in text.lower(): risks.append(f”检测到危险关键词: {keyword}”) filtered = re.sub( re.escape(keyword), ‘[FILTERED]’, filtered, flags=re.I ) return filtered, risks class DataMasker: “””数据脱敏器””” PATTERNS = { ‘phone’: (r’1[3-9]\d{9}’, lambda m: m.group()[:3] + ‘****’ + m.group()[-4:]), ’email’: (r'[\w.-]+@[\w.-]+\.\w+’, lambda m: m.group() + ‘***@***.’ + m.group().split(‘.’)[-1]), ‘id_card’: (r’\d{17}[\dXx]’, lambda m: m.group()[:6] + ‘********’ + m.group()[-4:]), ‘api_key’: (r’sk-[a-zA-Z0-9]{48}’, ‘sk-********************************’), ‘password’: (r'(password|pwd|passwd)[\s]*[=:][\s]*[^\s]+’, ‘password=[MASKED]’), } def mask(self, text: str) -> str: for data_type, (pattern, replacement) in self.PATTERNS.items(): if callable(replacement): text = re.sub(pattern, replacement, text) else: text = re.sub(pattern, replacement, text) return text class OutputFilter: “””输出过滤器””” SENSITIVE_PATTERNS = [ re.compile(r’sk-[a-zA-Z0-9]{48}’, re.I), # API密钥 re.compile(r’password[\s]*[=:][\s]*[^\s]+’, re.I), re.compile(r’secret[\s]*[=:][\s]*[^\s]+’, re.I), re.compile(r’\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\:\d+\b’), # IP:Port ] FORBIDDEN_CONTENT = [ “数据库密码”, “root密码”, “管理员账号”, “connection string”, “私钥”, “private key”, “系统配置”, “服务器IP” ] def filter(self, text: str) -> Tuple[str, bool]: has_sensitive = False filtered = text # 过滤敏感模式 for pattern in self.SENSITIVE_PATTERNS: if pattern.search(text): has_sensitive = True filtered = pattern.sub(‘[SENSITIVE]’, filtered) # 检查禁用内容 for keyword in self.FORBIDDEN_CONTENT: if keyword in text: has_sensitive = True return filtered, has_sensitive class RateLimiter: “””速率限制器（基于Redis）””” def __init__(self): self.redis_client = None # 初始化时连接 async def check(self, user_id: str, max_requests: int = 60) -> bool: “””检查是否超过速率限制””” # 实现：使用Redis计数器 # 每用户每分钟最多60次请求 return True # 简化实现

模块3：成本优化层（整合第三篇内容）

# core/optimization/cost_optimizer.py import hashlib import json from typing import Optional, Dict, Any from datetime import datetime, timedelta from enum import Enum from app.config import settings class TaskComplexity(Enum): “””任务复杂度””” SIMPLE = “simple” # GPT-3.5 STANDARD = “standard” # GPT-4-Turbo COMPLEX = “complex” # GPT-4 class ModelRouter: “””智能模型路由器””” MODEL_MAP = { TaskComplexity.SIMPLE: “gpt-3.5-turbo”, TaskComplexity.STANDARD: “gpt-4-turbo-preview”, TaskComplexity.COMPLEX: “gpt-4” } COST_RATES = { “gpt-3.5-turbo”: {“input”: 0.0015, “output”: 0.002}, “gpt-4-turbo-preview”: {“input”: 0.01, “output”: 0.03}, “gpt-4”: {“input”: 0.03, “output”: 0.06} } # 简单任务关键词 SIMPLE_KEYWORDS = [ “翻译”, “摘要”, “格式化”, “转换”, “提取”, “hello”, “hi”, “你好”, “谢谢”, “再见” ] # 复杂任务关键词 COMPLEX_KEYWORDS = [ “架构”, “设计”, “优化”, “调试”, “分析”, “architecture”, “design”, “optimize”, “debug” ] def route(self, query: str) -> str: “””根据查询路由到合适的模型””” complexity = self._classify_complexity(query) return self.MODEL_MAP[complexity] def _classify_complexity(self, query: str) -> TaskComplexity: “””分类任务复杂度””” query_lower = query.lower() # 检查简单关键词 if any(kw in query_lower for kw in self.SIMPLE_KEYWORDS): return TaskComplexity.SIMPLE # 检查复杂关键词 if any(kw in query_lower for kw in self.COMPLEX_KEYWORDS): return TaskComplexity.COMPLEX # 根据长度判断 if len(query) < 50: return TaskComplexity.SIMPLE elif len(query) > 500: return TaskComplexity.COMPLEX return TaskComplexity.STANDARD def estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float: “””估算成本””” rates = self.COST_RATES.get(model, self.COST_RATES[“gpt-3.5-turbo”]) cost = (input_tokens / 1000 * rates[“input”] + output_tokens / 1000 * rates[“output”]) return round(cost, 6) class SmartCache: “””智能缓存系统””” def __init__(self, redis_client): self.redis = redis_client self.default_ttl = settings.CACHE_TTL def _generate_key(self, query: str, model: str) -> str: “””生成缓存key””” content = f”{model}:{query}” return f”llm:cache:{hashlib.md5(content.encode()).hexdigest()}” async def get(self, query: str, model: str) -> Optional[Dict]: “””获取缓存””” if not settings.ENABLE_CACHE: return None key = self._generate_key(query, model) cached = await self.redis.get(key) if cached: return json.loads(cached) return None async def set( self, query: str, model: str, response: Dict, ttl: Optional[int] = None ): “””设置缓存””” if not settings.ENABLE_CACHE: return key = self._generate_key(query, model) await self.redis.setex( key, ttl or self.default_ttl, json.dumps(response, ensure_ascii=False) ) class CostTracker: “””成本追踪器””” def __init__(self, redis_client): self.redis = redis_client self.daily_budget = settings.COST_ALERT_THRESHOLD async def track( self, user_id: str, model: str, input_tokens: int, output_tokens: int, cost: float ): “””记录成本””” today = datetime.now().strftime(“%Y-%m-%d”) # 记录到Redis key = f”cost:{today}:{user_id}” await self.redis.hincrbyfloat(key, “total_cost”, cost) await self.redis.hincrby(key, “request_count”, 1) await self.redis.expire(key, 86400 * 7) # 保留7天 # 检查预算 daily_cost = await self.redis.hget(key, “total_cost”) if daily_cost and float(daily_cost) > self.daily_budget: await self._send_budget_alert(user_id, float(daily_cost)) async def _send_budget_alert(self, user_id: str, cost: float): “””发送预算告警””” # 集成钉钉/企业微信告警 pass async def get_daily_report(self, user_id: str) -> Dict: “””获取日报””” today = datetime.now().strftime(“%Y-%m-%d”) key = f”cost:{today}:{user_id}” data = await self.redis.hgetall(key) return { “date”: today, “user_id”: user_id, “total_cost”: float(data.get(“total_cost”, 0)), “request_count”: int(data.get(“request_count”, 0)), “budget”: self.daily_budget, “budget_usage”: float(data.get(“total_cost”, 0)) / self.daily_budget * 100 } class ContextCompressor: “””上下文压缩器””” def __init__(self, max_tokens: int = 3000): self.max_tokens = max_tokens def compress(self, messages: list) -> list: “”” 压缩对话历史策略：保留系统消息 + 最近的N条消息 + 历史摘要 “”” if not messages: return messages # 分离系统消息 system_msgs = [m for m in messages if m.get(“role”) == “system”] other_msgs = [m for m in messages if m.get(“role”) != “system”] # 如果消息不多，直接返回 if len(other_msgs) <= 6: return messages # 保留最近6条，其余生成摘要 recent_msgs = other_msgs[-6:] old_msgs = other_msgs[:-6] # 生成摘要（这里简化处理，实际可以用LLM生成） summary = { “role”: “system”, “content”: f”[历史对话摘要：之前共有{len(old_msgs)}轮对话]” } return system_msgs + [summary] + recent_msgs

🎛️ 第四部分：编排器与业务Agent实现

编排器（Orchestrator）核心实现

# services/orchestrator.py from typing import Dict, List, Optional, Any import asyncio from core.agents.base import BaseAgent, AgentResponse from core.agents.customer_service import CustomerServiceAgent from core.agents.sales import SalesAgent from core.agents.analyst import AnalystAgent from core.security.defense_layer import SecurityDefenseLayer from core.optimization.cost_optimizer import ModelRouter, SmartCache, CostTracker, ContextCompressor from core.memory.short_term import ShortTermMemory from core.memory.vector_store import VectorStore from app.config import settings class Orchestrator: “”” 编排器 – 系统的”大脑” 负责任务分发、Agent调度、结果整合 “”” def __init__(self): # 初始化所有Agent self.agents: Dict[str, BaseAgent] = { “customer_service”: CustomerServiceAgent(), “sales”: SalesAgent(), “analyst”: AnalystAgent() } # 初始化安全层 self.security = SecurityDefenseLayer() # 初始化优化层 self.model_router = ModelRouter() self.cache = SmartCache(redis_client=None) # 实际使用时注入 self.cost_tracker = CostTracker(redis_client=None) self.context_compressor = ContextCompressor() # 初始化记忆系统 self.short_term_memory = ShortTermMemory() self.vector_store = VectorStore() async def process_request( self, query: str, user_id: str, session_id: str, context: Optional[Dict] = None ) -> Dict[str, Any]: “”” 处理用户请求的主入口流程： 1. 安全过滤（输入） 2. 意图识别 3. 缓存检查 4. Agent调度 5. 执行处理 6. 安全过滤（输出） 7. 成本记录 8. 记忆更新 “”” start_time = asyncio.get_event_loop().time() warnings = [] try: # Step 1: 安全过滤（输入） safe_input, input_warnings = await self.security.process_input( query, user_id ) warnings.extend(input_warnings) # Step 2: 意图识别 + Agent选择 intent = await self._classify_intent(safe_input) agent = self._select_agent(intent) # Step 3: 检查缓存 model = self.model_router.route(safe_input) cached_response = await self.cache.get(safe_input, model) if cached_response: return { “success”: True, “response”: cached_response[“content”], “from_cache”: True, “agent_type”: agent.role.value, “warnings”: warnings, “cost”: 0, “latency_ms”: 0 } # Step 4: 获取对话上下文 conversation_history = await self.short_term_memory.get(session_id) compressed_context = self.context_compressor.compress(conversation_history) # Step 5: Agent执行 agent_response: AgentResponse = await agent.process( query=safe_input, context={ “history”: compressed_context, “user_id”: user_id, “session_id”: session_id, **(context or {}) } ) # Step 6: 安全过滤（输出） safe_output, has_sensitive = await self.security.process_output( agent_response.content, user_id ) if has_sensitive: warnings.append(“输出内容已被安全过滤”) # Step 7: 记录成本 await self.cost_tracker.track( user_id=user_id, model=agent_response.used_model, input_tokens=agent_response.metadata.get(“input_tokens”, 0), output_tokens=agent_response.metadata.get(“output_tokens”, 0), cost=agent_response.cost ) # Step 8: 更新记忆 await self.short_term_memory.add(session_id, { “role”: “user”, “content”: safe_input }) await self.short_term_memory.add(session_id, { “role”: “assistant”, “content”: safe_output }) # Step 9: 写入缓存 await self.cache.set( safe_input, model, {“content”: safe_output} ) latency_ms = int((asyncio.get_event_loop().time() – start_time) * 1000) return { “success”: agent_response.success, “response”: safe_output, “from_cache”: False, “agent_type”: agent.role.value, “used_model”: agent_response.used_model, “warnings”: warnings, “cost”: agent_response.cost, “latency_ms”: latency_ms } except Exception as e: return { “success”: False, “response”: f”处理请求时出错: {str(e)}”, “error”: str(e), “warnings”: warnings } async def _classify_intent(self, query: str) -> str: “””意图识别””” # 简单的关键词匹配，实际可以用分类模型 query_lower = query.lower() if any(kw in query_lower for kw in [“订单”, “退款”, “售后”, “客服”]): return “customer_service” elif any(kw in query_lower for kw in [“价格”, “购买”, “优惠”, “产品”]): return “sales” elif any(kw in query_lower for kw in [“分析”, “报表”, “数据”, “统计”]): return “analyst” return “customer_service” # 默认 def _select_agent(self, intent: str) -> BaseAgent: “””根据意图选择Agent””” return self.agents.get(intent, self.agents[“customer_service”]) async def get_cost_report(self, user_id: str) -> Dict: “””获取用户成本报告””” return await self.cost_tracker.get_daily_report(user_id)

客服Agent实现示例

# core/agents/customer_service.py from typing import Dict, Optional from core.agents.base import BaseAgent, AgentResponse, AgentRole from core.tools.database import DatabaseTool from core.tools.search import SearchTool class CustomerServiceAgent(BaseAgent): “””客服Agent – 处理用户咨询、订单查询、售后等””” def __init__(self): system_message = “””你是专业的客服助手，帮助用户解决以下问题： 1. 订单查询 – 查询订单状态、物流信息 2. 售后服务 – 处理退款、退货、换货申请 3. 产品咨询 – 解答产品功能、使用方法 4. 账户问题 – 密码重置、信息修改原则： – 礼貌、耐心、专业 – 不清楚的问题，不要编造 – 涉及敏感操作，需要确认用户身份 – 复杂问题及时转人工””” super().__init__( name=”CustomerServiceAgent”, role=AgentRole.CUSTOMER_SERVICE, system_message=system_message, tools=[ DatabaseTool(), SearchTool() ] ) self.db_tool = DatabaseTool() self.search_tool = SearchTool() async def process( self, query: str, context: Optional[Dict] = None, user_id: Optional[str] = None ) -> AgentResponse: “””处理客服请求””” import time start = time.time() # 根据查询类型路由到不同处理逻辑 if “订单” in query or “查询” in query: response = await self._handle_order_query(query, context, user_id) elif “退款” in query or “退货” in query: response = await self._handle_refund(query, context, user_id) elif “产品” in query or “功能” in query: response = await self._handle_product_inquiry(query) else: # 通用对话 response = await self._general_chat(query, context) latency = int((time.time() – start) * 1000) return AgentResponse( content=response, success=True, metadata={ “agent_type”: “customer_service”, “input_tokens”: len(query) // 4, # 估算 “output_tokens”: len(response) // 4 }, cost=self.estimate_cost(len(query) // 4, len(response) // 4), latency_ms=latency, used_model=self.model ) async def _handle_order_query( self, query: str, context: Optional[Dict], user_id: Optional[str] ) -> str: “””处理订单查询””” # 提取订单号（简化示例） import re order_match = re.search(r’\d{10,}’, query) if order_match: order_id = order_match.group() # 查询数据库 order_info = await self.db_tool.query_order(order_id, user_id) if order_info: return f”订单 {order_id} 信息：\n状态：{order_info[‘status’]}\n金额：{order_info[‘amount’]}\n创建时间：{order_info[‘created_at’]}” else: return f”未找到订单 {order_id}，请确认订单号是否正确” else: return “请提供订单号，我可以帮您查询订单状态” async def _handle_refund( self, query: str, context: Optional[Dict], user_id: Optional[str] ) -> str: “””处理退款申请””” # 安全：退款操作需要身份验证 if not user_id: return “退款申请需要先登录账户” return “我已收到您的退款申请，客服将在24小时内审核并联系您” async def _handle_product_inquiry(self, query: str) -> str: “””处理产品咨询””” # 搜索知识库 results = await self.search_tool.search_knowledge_base(query) if results: return f”根据您的问题，我找到了以下信息：\n\n{results[‘content’]}” else: return “抱歉，关于这个问题我暂时没有找到相关信息，建议您联系人工客服” async def _general_chat(self, query: str, context: Optional[Dict]) -> str: “””通用对话””” # 使用AutoGen代理生成回复 # 这里简化处理，实际调用self.agent return f”感谢您的咨询！我是客服助手，请问有什么可以帮您？” def get_capabilities(self) -> List[str]: return [ “订单查询”, “退款退货处理”, “产品咨询”, “账户问题” ]

🌐 第五部分：API接口与前端集成

FastAPI接口实现

# app/api/v1/chat.py from fastapi import APIRouter, Depends, HTTPException, Request from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from pydantic import BaseModel from typing import Optional, List import time from services.orchestrator import Orchestrator from app.dependencies import get_current_user, get_orchestrator from core.monitoring import metrics router = APIRouter(prefix=”/chat”, tags=[“对话”]) security = HTTPBearer() class ChatRequest(BaseModel): “””对话请求””” message: str session_id: Optional[str] = None context: Optional[dict] = None class ChatResponse(BaseModel): “””对话响应””” success: bool response: str agent_type: str used_model: Optional[str] = None from_cache: bool = False cost: float = 0.0 latency_ms: int = 0 warnings: List[str] = [] @router.post(“/send”, response_model=ChatResponse) async def send_message( request: ChatRequest, current_user: dict = Depends(get_current_user), orchestrator: Orchestrator = Depends(get_orchestrator) ): “”” 发送消息并获取回复 – **message**: 用户消息 – **session_id**: 会话ID（用于保持上下文） – **context**: 额外上下文信息 “”” start_time = time.time() try: # 生成或复用session_id session_id = request.session_id or f”{current_user[‘id’]}_{int(start_time)}” # 处理请求 result = await orchestrator.process_request( query=request.message, user_id=current_user[“id”], session_id=session_id, context=request.context ) # 记录指标 metrics.record_request( endpoint=”/chat/send”, latency=result.get(“latency_ms”, 0), success=result[“success”] ) return ChatResponse( success=result[“success”], response=result[“response”], agent_type=result.get(“agent_type”, “unknown”), used_model=result.get(“used_model”), from_cache=result.get(“from_cache”, False), cost=result.get(“cost”, 0), latency_ms=result.get(“latency_ms”, 0), warnings=result.get(“warnings”, []) ) except Exception as e: metrics.record_error(“/chat/send”, str(e)) raise HTTPException(status_code=500, detail=str(e)) @router.get(“/history/{session_id}”) async def get_chat_history( session_id: str, current_user: dict = Depends(get_current_user) ): “””获取对话历史””” from core.memory.short_term import ShortTermMemory memory = ShortTermMemory() history = await memory.get(session_id) return {“session_id”: session_id, “messages”: history} @router.get(“/cost-report”) async def get_cost_report( current_user: dict = Depends(get_current_user), orchestrator: Orchestrator = Depends(get_orchestrator) ): “””获取用户成本报告””” report = await orchestrator.get_cost_report(current_user[“id”]) return report # app/main.py – 应用入口 from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.config import settings from app.api.v1 import chat, agents, knowledge, users app = FastAPI( title=settings.APP_NAME, version=settings.APP_VERSION, description=”企业级AI助手系统” ) # CORS配置 app.add_middleware( CORSMiddleware, allow_origins=[“*”], # 生产环境应限制域名 allow_credentials=True, allow_methods=[“*”], allow_headers=[“*”], ) # 注册路由 app.include_router(chat.router, prefix=”/api/v1″) app.include_router(agents.router, prefix=”/api/v1″) app.include_router(knowledge.router, prefix=”/api/v1″) app.include_router(users.router, prefix=”/api/v1″) @app.get(“/health”) async def health_check(): “””健康检查””” return {“status”: “healthy”, “version”: settings.APP_VERSION} @app.get(“/”) async def root(): “””根路径””” return { “name”: settings.APP_NAME, “version”: settings.APP_VERSION, “docs”: “/docs” }

前端界面示例（React）

// frontend/src/components/Chat.jsx import React, { useState, useRef, useEffect } from ‘react’; import { Send, Bot, User, AlertCircle } from ‘lucide-react’; import { sendMessage, getChatHistory } from ‘../api/chat’; const Chat = ({ userToken }) => { const [messages, setMessages] = useState([]); const [input, setInput] = useState(”); const [loading, setLoading] = useState(false); const [sessionId, setSessionId] = useState(null); const messagesEndRef = useRef(null); const scrollToBottom = () => { messagesEndRef.current?.scrollIntoView({ behavior: ‘smooth’ }); }; useEffect(() => { scrollToBottom(); }, [messages]); const handleSend = async () => { if (!input.trim()) return; const userMessage = { role: ‘user’, content: input }; setMessages(prev => […prev, userMessage]); setInput(”); setLoading(true); try { const response = await sendMessage({ message: input, session_id: sessionId }, userToken); if (response.session_id) { setSessionId(response.session_id); } const assistantMessage = { role: ‘assistant’, content: response.response, metadata: { agentType: response.agent_type, model: response.used_model, cost: response.cost, fromCache: response.from_cache, warnings: response.warnings } }; setMessages(prev => […prev, assistantMessage]); } catch (error) { setMessages(prev => […prev, { role: ‘error’, content: ‘发送消息失败，请重试’ }]); } finally { setLoading(false); } }; return (

{/* 消息列表 */}

{messages.map((msg, idx) => (

{msg.role === ‘user’ ?:} {msg.role === ‘user’ ? ‘我’ : ‘AI助手’}

{msg.content}

{/* 元信息 */} {msg.metadata && (

模型: {msg.metadata.model} | 成本: ${msg.metadata.cost?.toFixed(4)} {msg.metadata.fromCache && (缓存)}

)} {/* 警告 */} {msg.metadata?.warnings?.length > 0 && (

{msg.metadata.warnings}

)}

))}{/* 输入框 */}

setInput(e.target.value)} onKeyPress={(e) => e.key === ‘Enter’ && handleSend()} placeholder=”输入消息…” className=”flex-1 border rounded-lg px-4 py-2 focus:outline-none focus:ring-2 focus:ring-blue-500″ disabled={loading} />

); }; export default Chat;

🚀 第六部分：生产环境部署

Docker Compose部署配置

# docker-compose.yml version: ‘3.8’ services: # 主应用 app: build: . ports: – “8000:8000” environment: – DATABASE_URL=postgresql+asyncpg://postgres:postgres@db:5432/ai_assistant – REDIS_URL=redis://redis:6379/0 – OPENAI_API_KEY=${OPENAI_API_KEY} depends_on: – db – redis – milvus volumes: – ./data:/app/data restart: unless-stopped deploy: resources: limits: cpus: ‘2’ memory: 4G # 任务队列Worker worker: build: . command: celery -A tasks worker –loglevel=info environment: – DATABASE_URL=postgresql+asyncpg://postgres:postgres@db:5432/ai_assistant – REDIS_URL=redis://redis:6379/0 depends_on: – db – redis restart: unless-stopped # 定时任务 scheduler: build: . command: celery -A tasks beat –loglevel=info environment: – REDIS_URL=redis://redis:6379/0 depends_on: – redis restart: unless-stopped # PostgreSQL数据库 db: image: postgres:15-alpine environment: – POSTGRES_USER=postgres – POSTGRES_PASSWORD=postgres – POSTGRES_DB=ai_assistant volumes: – postgres_data:/var/lib/postgresql/data ports: – “5432:5432” restart: unless-stopped # Redis缓存 redis: image: redis:7-alpine ports: – “6379:6379” volumes: – redis_data:/data restart: unless-stopped # Milvus向量数据库 milvus: image: milvusdb/milvus:v2.3.3 ports: – “19530:19530” – “9091:9091” volumes: – milvus_data:/var/lib/milvus restart: unless-stopped # Nginx反向代理 nginx: image: nginx:alpine ports: – “80:80” – “443:443” volumes: – ./deployments/nginx/nginx.conf:/etc/nginx/nginx.conf – ./deployments/nginx/ssl:/etc/nginx/ssl depends_on: – app restart: unless-stopped # 前端 frontend: build: ./frontend ports: – “3000:80” depends_on: – app restart: unless-stopped volumes: postgres_data: redis_data: milvus_data:

# Dockerfile FROM python:3.11-slim WORKDIR /app # 安装系统依赖 RUN apt-get update && apt-get install -y \ gcc \ postgresql-client \ && rm -rf /var/lib/apt/lists/* # 安装Python依赖 COPY requirements.txt . RUN pip install –no-cache-dir -r requirements.txt # 复制代码 COPY app/ ./app/ COPY core/ ./core/ COPY services/ ./services/ COPY models/ ./models/ # 非root用户运行 RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app USER appuser EXPOSE 8000 CMD [“uvicorn”, “app.main:app”, “–host”, “0.0.0.0”, “–port”, “8000”, “–workers”, “4”]

监控与告警配置

# core/monitoring/metrics.py from prometheus_client import Counter, Histogram, Gauge, Info import time # 定义指标 REQUEST_COUNT = Counter( ‘http_requests_total’, ‘Total HTTP requests’, [‘method’, ‘endpoint’, ‘status’] ) REQUEST_LATENCY = Histogram( ‘http_request_duration_seconds’, ‘HTTP request latency’, [‘method’, ‘endpoint’] ) LLM_COST = Counter( ‘llm_cost_dollars_total’, ‘Total LLM API cost’, [‘model’, ‘user_id’] ) LLM_TOKENS = Counter( ‘llm_tokens_total’, ‘Total LLM tokens used’, [‘model’, ‘type’] # type: input/output ) CACHE_HIT = Counter( ‘cache_hits_total’, ‘Total cache hits’, [‘cache_type’] ) SECURITY_EVENTS = Counter( ‘security_events_total’, ‘Total security events’, [‘event_type’, ‘severity’] ) ACTIVE_SESSIONS = Gauge( ‘active_sessions’, ‘Number of active sessions’ ) APP_INFO = Info(‘app’, ‘Application information’) class MetricsCollector: “””指标收集器””” @staticmethod def record_request(endpoint: str, latency_ms: int, success: bool): “””记录请求指标””” REQUEST_COUNT.labels( method=’POST’, endpoint=endpoint, status=’success’ if success else ‘error’ ).inc() REQUEST_LATENCY.labels( method=’POST’, endpoint=endpoint ).observe(latency_ms / 1000) @staticmethod def record_llm_usage(model: str, user_id: str, input_tokens: int, output_tokens: int, cost: float): “””记录LLM使用情况””” LLM_COST.labels(model=model, user_id=user_id).inc(cost) LLM_TOKENS.labels(model=model, type=’input’).inc(input_tokens) LLM_TOKENS.labels(model=model, type=’output’).inc(output_tokens) @staticmethod def record_cache_hit(cache_type: str = ‘llm’): “””记录缓存命中””” CACHE_HIT.labels(cache_type=cache_type).inc() @staticmethod def record_security_event(event_type: str, severity: str): “””记录安全事件””” SECURITY_EVENTS.labels(event_type=event_type, severity=severity).inc() @staticmethod def record_error(endpoint: str, error_type: str): “””记录错误””” REQUEST_COUNT.labels( method=’POST’, endpoint=endpoint, status=’error’ ).inc() # 初始化应用信息 APP_INFO.info({‘version’: ‘1.0.0’, ‘name’: ‘Enterprise AI Assistant’})

✍️ 写在最后：完整系列回顾

至此，Agent系列文章正式完结。五篇文章，带你从0到1构建企业级AI助手：

📚 系列文章回顾：

1️⃣ 《谷歌Agent Smith完整解析》 —— 理解Agent原理，激发想象力

2️⃣ 《多Agent协作》 —— 从单兵作战到团队协作，能力倍增

3️⃣ 《性能优化》 —— 降低80%成本，让项目可持续发展

4️⃣ 《安全实战》 —— 防范Prompt注入与数据泄露，守住底线

5️⃣ 《落地实战》 —— 从零搭建企业级系统，真正投入使用

核心收获：

✅ 掌握了Agent系统的完整架构设计

✅ 学会了多Agent协作的实现方法

✅ 了解了成本优化的7大策略

✅ 建立了安全防护体系

✅ 拥有了一套可直接部署的代码

“AI时代已经到来，Agent是连接大模型与业务的桥梁。希望这套教程能帮助你在这个时代抢占先机。”

后续规划：

• 开源代码仓库（GitHub）：整理后开放完整代码

• 视频教程：配合代码讲解

• 社区交流：建立技术交流群

如果你有任何问题或建议，欢迎在评论区留言！

👇 完结撒花 👇

5篇文章，你最喜欢哪一篇？你在搭建过程中遇到什么困难？或者，你已经成功部署了自己的Agent系统？

💬 写评论👍 点赞⭐ 收藏系列

📌 感谢陪伴！如果这个系列对你有帮助，欢迎分享给更多朋友🔔 未来会持续分享AI实战干货，保持关注哦～

— 本文由织美原创出品，转载请联系授权 —📅 2026年4月22日