江门市网站建设_网站建设公司_VS Code_seo优化

1. 项目概述与架构设计

1.1 系统核心设计理念

本系统采用“AI优先，人工兜底”的混合交互模式，构建一个能够自动处理80%常规查询，同时无缝转接复杂问题给人工坐席的智能客服系统。系统基于微服务架构，确保高可用性、可扩展性和模块化设计。

1.2 整体架构图

┌─────────────────────────────────────────────────────────────┐ │ 客户端层 (多渠道接入) │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ │ Web聊天 │ │ 移动APP │ │ 微信 │ │ 电话接口│ │ │ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │ └────────────────────────┬────────────────────────────────────┘ │ HTTPS/WebSocket ┌─────────────────────────────────────────────────────────────┐ │ API网关层 (统一接入) │ │ ┌─────────────────────────────────────────────────────┐ │ │ │ 认证授权 │ 流量控制 │ 协议转换 │ 请求路由 │ 负载均衡 │ │ │ └─────────────────────────────────────────────────────┘ │ └────────────────────────┬────────────────────────────────────┘ │ 内部RPC/gRPC ┌─────────────────────────────────────────────────────────────┐ │ 业务逻辑层 (核心服务) │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ │对话管理 │ │意图识别 │ │知识检索 │ │坐席协作 │ │ │ │ 服务 │ │ 服务 │ │ 服务 │ │ 服务 │ │ │ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │ └────────────────────────┬────────────────────────────────────┘ │ 消息队列/数据库访问 ┌─────────────────────────────────────────────────────────────┐ │ 数据与AI层 (能力支撑) │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ │向量数据库│ │关系数据库│ │缓存服务 │ │AI模型 │ │ │ │ (FAISS/ │ │(PostgreSQL│ │(Redis) │ │服务 │ │ │ │Chroma) │ │ /MySQL) │ │ │ │ │ │ │ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │ └─────────────────────────────────────────────────────────────┘

1.3 技术栈选择

2. 环境搭建与项目初始化

2.1 开发环境配置

# 1. 创建项目目录结构mkdir-p ai-human-customer-servicecdai-human-customer-service# 2. 创建Python虚拟环境python3.10 -m venv venvsourcevenv/bin/activate# Linux/Mac# venv\Scripts\activate # Windows# 3. 安装基础依赖pipinstall--upgrade pip pipinstallfastapi==0.104.1 pipinstalluvicorn[standard]==0.24.0 pipinstallsqlalchemy==2.0.23 pipinstallasyncpg==0.29.0 pipinstallredis==5.0.1 pipinstallpydantic==2.5.0 pipinstallpydantic-settings==2.1.0 pipinstallpython-multipart==0.0.6 pipinstallwebsockets==12.0pipinstallaiohttp==3.9.1 pipinstalljwt==1.3.1 pipinstallpython-dateutil==2.8.2 pipinstallloguru==0.7.2# 4. 创建项目结构mkdir-p{src,tests,docs,scripts,deploy}mkdir-p src/{api,core,models,schemas,services,utils,ai_components}mkdir-p src/api/{endpoints,middleware}mkdir-p src/core/{config,database,security}mkdir-p src/ai_components/{nlp,retrieval,models}mkdir-p tests/{unit,integration}

2.2 配置文件设计

创建src/core/config.py：

""" 系统配置管理模块 使用pydantic-settings进行配置管理，支持环境变量覆盖 """fromtypingimportOptional,List,Dict,Anyfrompydantic_settingsimportBaseSettingsfrompydanticimportField,validatorimportsecretsclassSettings(BaseSettings):"""应用配置类"""# 应用基础配置APP_NAME:str="AI-Human Customer Service"APP_VERSION:str="1.0.0"DEBUG:bool=FalseENVIRONMENT:str="development"# development, staging, production# API配置API_V1_STR:str="/api/v1"PROJECT_NAME:str="AI Human Customer Service"BACKEND_CORS_ORIGINS:List[str]=["http://localhost:3000"]# 安全配置SECRET_KEY:str=Field(default_factory=lambda:secrets.token_urlsafe(32))ALGORITHM:str="HS256"ACCESS_TOKEN_EXPIRE_MINUTES:int=60*24*7# 7天# 数据库配置POSTGRES_SERVER:str="localhost"POSTGRES_USER:str="postgres"POSTGRES_PASSWORD:str="password"POSTGRES_DB:str="customer_service"POSTGRES_PORT:str="5432"DATABASE_URL:Optional[str]=None@validator("DATABASE_URL",pre=True)defassemble_db_connection(cls,v:Optional[str],values:Dict[str,Any])->Any:"""构建数据库连接URL"""ifisinstance(v,str):returnvreturn(f"postgresql+asyncpg://{values.get('POSTGRES_USER')}:"f"{values.get('POSTGRES_PASSWORD')}@{values.get('POSTGRES_SERVER')}:"f"{values.get('POSTGRES_PORT')}/{values.get('POSTGRES_DB')}")# Redis配置REDIS_HOST:str="localhost"REDIS_PORT:int=6379REDIS_DB:int=0REDIS_PASSWORD:Optional[str]=NoneREDIS_URL:Optional[str]=None@validator("REDIS_URL",pre=True)defassemble_redis_connection(cls,v:Optional[str],values:Dict[str,Any])->Any:"""构建Redis连接URL"""ifisinstance(v,str):returnv password=values.get("REDIS_PASSWORD")auth_part=f":{password}@"ifpasswordelse""return(f"redis://{auth_part}{values.get('REDIS_HOST')}:"f"{values.get('REDIS_PORT')}/{values.get('REDIS_DB')}")# AI模型配置AI_MODEL_PATH:str="./models"SENTENCE_TRANSFORMER_MODEL:str="all-MiniLM-L6-v2"BERT_MODEL_PATH:str="bert-base-uncased"GPT_MODEL_PATH:str="gpt2"USE_GPU:bool=TrueAI_MODEL_CACHE_SIZE:int=100# 对话管理配置MAX_CONVERSATION_HISTORY:int=20SESSION_TIMEOUT_MINUTES:int=30DEFAULT_AI_CONFIDENCE_THRESHOLD:float=0.7# 坐席配置MAX_AGENTS_PER_QUEUE:int=10AGENT_HEARTBEAT_INTERVAL:int=30# 秒AGENT_AWAY_TIMEOUT:int=300# 秒# 消息队列配置RABBITMQ_HOST:str="localhost"RABBITMQ_PORT:int=5672RABBITMQ_USER:str="guest"RABBITMQ_PASSWORD:str="guest"# 监控配置ENABLE_METRICS:bool=TrueMETRICS_PORT:int=9090LOG_LEVEL:str="INFO"classConfig:"""Pydantic配置"""env_file=".env"case_sensitive=Trueextra="ignore"settings=Settings()# 创建环境文件模板ENV_TEMPLATE="""# 应用配置 ENVIRONMENT=development DEBUG=True # 数据库配置 POSTGRES_SERVER=localhost POSTGRES_USER=postgres POSTGRES_PASSWORD=your_password POSTGRES_DB=customer_service POSTGRES_PORT=5432 # Redis配置 REDIS_HOST=localhost REDIS_PORT=6379 REDIS_DB=0 # AI模型配置 SENTENCE_TRANSFORMER_MODEL=all-MiniLM-L6-v2 USE_GPU=False # 安全配置 SECRET_KEY={secret_key} """# 生成.env文件defcreate_env_file():"""创建环境变量文件"""withopen(".env","w")asf:secret_key=secrets.token_urlsafe(32)f.write(ENV_TEMPLATE.format(secret_key=secret_key))print("✅ .env文件创建成功")if__name__=="__main__":create_env_file()

3. 核心数据模型设计

3.1 数据库模型定义

创建src/models/base.py：

""" 数据库基础模型定义 使用SQLAlchemy ORM进行数据建模 """fromdatetimeimportdatetimefromtypingimportAny,Dict,OptionalfromsqlalchemyimportColumn,DateTime,Integer,String,Boolean,Text,JSONfromsqlalchemy.ext.declarativeimportdeclarative_basefromsqlalchemy.sqlimportfunc Base=declarative_base()classTimestampMixin:"""时间戳混入类"""created_at=Column(DateTime,default=func.now(),nullable=False)updated_at=Column(DateTime,default=func.now(),onupdate=func.now(),nullable=False)deleted_at=Column(DateTime,nullable=True)classCustomer(Base,TimestampMixin):"""客户模型"""__tablename__="customers"id=Column(String(36),primary_key=True,index=True)external_id=Column(String(100),unique=True,nullable=True,index=True)name=Column(String(200),nullable=True)email=Column(String(254),nullable=True,index=True)phone=Column(String(50),nullable=True,index=True)metadata=Column(JSON,nullable=True,default=dict)tags=Column(JSON,nullable=True,default=list)segment=Column(String(50),nullable=True)lifetime_value=Column(Integer,default=0)last_interaction_at=Column(DateTime,nullable=True)is_active=Column(Boolean,default=True)notes=Column(Text,nullable=True)defto_dict(self)->Dict[str,Any]:"""转换为字典"""return{"id":self.id,"external_id":self.external_id,"name":self.name,"email":self.email,"phone":self.phone,"metadata":self.metadataor{},"tags":self.tagsor[],"segment":self.segment,"lifetime_value":self.lifetime_value,"last_interaction_at":self.last_interaction_at.isoformat()ifself.last_interaction_atelseNone,"is_active":self.is_active,"notes":self.notes,"created_at":self.created_at.isoformat(),"updated_at":self.updated_at.isoformat()}classConversation(Base,TimestampMixin):"""对话模型"""__tablename__="conversations"id=Column(String(36),primary_key=True,index=True)customer_id=Column(String(36),index=True,nullable=False)channel=Column(String(50),nullable=False)# web, mobile, wechat, etc.status=Column(String(20),default="active")# active, pending, closed, transferredassigned_agent_id=Column(String(36),nullable=True,index=True)assigned_ai_model=Column(String(100),nullable=True)metadata=Column(JSON,nullable=True,default=dict)sentiment_score=Column(Integer,nullable=True)# -100 to 100satisfaction_score=Column(Integer,nullable=True)# 0-100closed_at=Column(DateTime,nullable=True)close_reason=Column(String(100),nullable=True)defto_dict(self)->Dict[str,Any]:"""转换为字典"""return{"id":self.id,"customer_id":self.customer_id,"channel":self.channel,"status":self.status,"assigned_agent_id":self.assigned_agent_id,"assigned_ai_model":self.assigned_ai_model,"metadata":self.metadataor{},"sentiment_score":self.sentiment_score,"satisfaction_score":self.satisfaction_score,"closed_at":self.closed_at.isoformat()ifself.closed_atelseNone,"close_reason":self.close_reason,"created_at":self.created_at.isoformat(),"updated_at":self.updated_at.isoformat()}classMessage(Base,TimestampMixin):"""消息模型"""__tablename__="messages"id=Column(String(36),primary_key=True,index=True)conversation_id=Column(String(36),index=True,nullable=False)sender_type=Column(String(20),nullable=False)# customer, agent, aisender_id=Column(String(36),nullable=True)content=Column(Text,nullable=False)content_type=Column(String(20),default="text")# text, image, file, etc.metadata=Column(JSON,nullable=True,default=dict)intent=Column(String(100),nullable=True)confidence=Column(Integer,nullable=True)# 0-100is_read=Column(Boolean,default=False)read_at=Column(DateTime,nullable=True)# AI相关字段ai_model_used=Column(String(100),nullable=True)ai_confidence=Column(Integer,nullable=True)ai_metadata=Column(JSON,nullable=True,default=dict)defto_dict(self)->Dict[str,Any]:"""转换为字典"""return{"id":self.id,"conversation_id":self.conversation_id,"sender_type":self.sender_type,"sender_id":self.sender_id,"content":self.content,"content_type":self.content_type,"metadata":self.metadataor{},"intent":self.intent,"confidence":self.confidence,"is_read":self.is_read,"read_at":self.read_at.isoformat()ifself.read_atelseNone,"ai_model_used":self.ai_model_used,"ai_confidence":self.ai_confidence,"ai_metadata":self.ai_metadataor{},"created_at":self.created_at.isoformat()}classAgent(Base,TimestampMixin):"""坐席模型"""__tablename__="agents"id=Column(String(36),primary_key=True,index=True)user_id=Column(String(36),unique=True,nullable=False)name=Column(String(200),nullable=False)email=Column(String(254),unique=True,nullable=False)status=Column(String(20),default="offline")# online, away, offline, busyskills=Column(JSON,nullable=True,default=list)current_conversation_ids=Column(JSON,nullable=True,default=list)max_concurrent_chats=Column(Integer,default=3)metadata=Column(JSON,nullable=True,default=dict)last_active_at=Column(DateTime,nullable=True)availability_schedule=Column(JSON,nullable=True)# 排班信息performance_score=Column(Integer,default=0)is_active=Column(Boolean,default=True)defto_dict(self)->Dict[str,Any]:"""转换为字典"""return{"id":self.id,"user_id":self.user_id,"name":self.name,"email":self.email,"status":self.status,"skills":self.skillsor[],"current_conversation_ids":self.current_conversation_idsor[],"max_concurrent_chats":self.max_concurrent_chats,"metadata":self.metadataor{},"last_active_at":self.last_active_at.isoformat()ifself.last_active_atelseNone,"availability_schedule":self.availability_scheduleor{},"performance_score":self.performance_score,"is_active":self.is_active,"created_at":self.created_at.isoformat(),"updated_at":self.updated_at.isoformat()}classKnowledgeBase(Base,TimestampMixin):"""知识库模型"""__tablename__="knowledge_base"id=Column(String(36),primary_key=True,index=True)title=Column(String(500),nullable=False)content=Column(Text,nullable=False)category=Column(String(100),nullable=True,index=True)tags=Column(JSON,nullable=True,default=list)language=Column(String(10),default="zh")is_active=Column(Boolean,default=True)vector_embedding=Column(JSON,nullable=True)# 存储向量嵌入metadata=Column(JSON,nullable=True,default=dict)usage_count=Column(Integer,default=0)last_used_at=Column(DateTime,nullable=True)defto_dict(self)->Dict[str,Any]:"""转换为字典"""return{"id":self.id,"title":self.title,"content":self.content,"category":self.category,"tags":self.tagsor[],"language":self.language,"is_active":self.is_active,"metadata":self.metadataor{},"usage_count":self.usage_count,"last_used_at":self.last_used_at.isoformat()ifself.last_used_atelseNone,"created_at":self.created_at.isoformat(),"updated_at":self.updated_at.isoformat()}classIntent(Base,TimestampMixin):"""意图模型"""__tablename__="intents"id=Column(String(36),primary_key=True,index=True)name=Column(String(100),nullable=False,unique=True)description=Column(Text,nullable=True)examples=Column(JSON,nullable=True,default=list)handler_type=Column(String(50),default="ai")# ai, agent, hybridconfidence_threshold=Column(Integer,default=70)metadata=Column(JSON,nullable=True,default=dict)is_active=Column(Boolean,default=True)defto_dict(self)->Dict[str,Any]:"""转换为字典"""return{"id":self.id,"name":self.name,"description":self.description,"examples":self.examplesor[],"handler_type":self.handler_type,"confidence_threshold":self.confidence_threshold,"metadata":self.metadataor{},"is_active":self.is_active,"created_at":self.created_at.isoformat(),"updated_at":self.updated_at.isoformat()}classAIResponseLog(Base,TimestampMixin):"""AI响应日志"""__tablename__="ai_response_logs"id=Column(String(36),primary_key=True,index=True)conversation_id=Column(String(36),index=True,nullable=False)message_id=Column(String(36),index=True,nullable=False)ai_model=Column(String(100),nullable=False)prompt=Column(Text,nullable=False)response=Column(Text,nullable=False)confidence=Column(Integer,nullable=True)processing_time_ms=Column(Integer,nullable=True)tokens_used=Column(Integer,nullable=True)metadata=Column(JSON,nullable=True,default=dict)defto_dict(self)->Dict[str,Any]:"""转换为字典"""return{"id":self.id,"conversation_id":self.conversation_id,"message_id":self.message_id,"ai_model":self.ai_model,"prompt":self.prompt,"response":self.response,"confidence":self.confidence,"processing_time_ms":self.processing_time_ms,"tokens_used":self.tokens_used,"metadata":self.metadataor{},"created_at":self.created_at.isoformat()}

3.2 数据库初始化脚本

创建scripts/init_database.py：

#!/usr/bin/env python3""" 数据库初始化脚本 创建数据库表结构和初始数据 """importasyncioimportsysfrompathlibimportPath# 添加项目根目录到Python路径sys.path.append(str(Path(__file__).parent.parent))fromsqlalchemy.ext.asyncioimportcreate_async_engine,AsyncSessionfromsqlalchemy.ormimportsessionmakerfromsqlalchemyimporttextfromsrc.core.configimportsettingsfromsrc.models.baseimportBasefromsrc.modelsimport*# 导入所有模型asyncdefinit_database():"""初始化数据库"""print("🚀 开始初始化数据库...")# 创建异步引擎database_url=settings.DATABASE_URL.replace("postgresql+asyncpg","postgresql")sync_engine=create_async_engine(database_url,echo=settings.DEBUG,pool_pre_ping=True,pool_size=20,max_overflow=30)# 创建所有表print("📦 创建数据库表...")asyncwithsync_engine.begin()asconn:awaitconn.run_sync(Base.metadata.create_all)print("✅ 数据库表创建完成")# 创建初始数据print("📝 创建初始数据...")# 使用异步会话async_session=sessionmaker(sync_engine,class_=AsyncSession,expire_on_commit=False)asyncwithasync_session()assession:try:# 创建示例意图sample_intents=[{"id":"greeting","name":"greeting","description":"客户打招呼","examples":["你好","嗨","早上好","在吗","有人吗"],"handler_type":"ai","confidence_threshold":60},{"id":"product_inquiry","name":"product_inquiry","description":"产品咨询","examples":["这个产品多少钱","有什么功能","怎么使用","有什么优惠","什么时候发货"],"handler_type":"ai","confidence_threshold":70},{"id":"technical_support","name":"technical_support","description":"技术支持","examples":["无法登录","系统错误","怎么重置密码","连接失败","闪退问题"],"handler_type":"hybrid","confidence_threshold":65},{"id":"complaint","name":"complaint","description":"投诉建议","examples":["我要投诉","服务太差","质量有问题","退款","赔偿"],"handler_type":"agent","confidence_threshold":75},{"id":"payment_issue","name":"payment_issue","description":"支付问题","examples":["付款失败","重复扣款","退款未到账","支付方式","发票问题"],"handler_type":"hybrid","confidence_threshold":70}]# 插入意图数据forintent_datainsample_intents:awaitsession.execute(text(""" INSERT INTO intents (id, name, description, examples, handler_type, confidence_threshold, created_at, updated_at) VALUES (:id, :name, :description, :examples, :handler_type, :confidence_threshold, NOW(), NOW()) ON CONFLICT (id) DO NOTHING """),intent_data)# 创建示例知识库条目knowledge_entries=[{"id":"kb_welcome","title":"欢迎信息","content":"欢迎使用我们的客服系统！我是AI助手，可以帮您解答问题。如需人工服务，请告诉我。","category":"general","tags":["welcome","greeting"],"language":"zh"},{"id":"kb_product_info","title":"产品基本信息","content":"我们的产品提供以下功能：1. 智能对话 2. 多语言支持 3. 24小时在线 4. 人工转接。价格根据套餐不同有所差异。","category":"product","tags":["product","features","pricing"],"language":"zh"},{"id":"kb_technical_help","title":"常见技术问题","content":"如果您遇到技术问题：1. 尝试刷新页面 2. 清除缓存 3. 检查网络连接 4. 重启应用。如果问题持续，请联系技术支持。","category":"technical","tags":["troubleshooting","help","support"],"language":"zh"},{"id":"kb_refund_policy","title":"退款政策","content":"我们的退款政策：购买后30天内可申请退款，需提供订单号和退款原因。退款将在7-14个工作日内处理。","category":"policy","tags":["refund","policy","money"],"language":"zh"}]forkb_datainknowledge_entries:awaitsession.execute(text(""" INSERT INTO knowledge_base (id, title, content, category, tags, language, is_active, created_at, updated_at) VALUES (:id, :title, :content, :category, :tags, :language, true, NOW(), NOW()) ON CONFLICT (id) DO NOTHING """),kb_data)awaitsession.commit()print("✅ 初始数据创建完成")exceptExceptionase:awaitsession.rollback()print(f"❌ 创建初始数据失败:{e}")raiseprint("🎉 数据库初始化完成！")asyncdefcheck_database_connection():"""检查数据库连接"""print("🔍 检查数据库连接...")try:database_url=settings.DATABASE_URL.replace("postgresql+asyncpg","postgresql")engine=create_async_engine(database_url)asyncwithengine.connect()asconn:result=awaitconn.execute(text("SELECT version()"))version=result.scalar()print(f"✅ 数据库连接成功:{version}")awaitengine.dispose()returnTrueexceptExceptionase:print(f"❌ 数据库连接失败:{e}")returnFalseasyncdefmain():"""主函数"""print("="*50)print("数据库初始化工具")print("="*50)# 检查连接ifnotawaitcheck_database_connection():print("请检查数据库配置和连接")return# 初始化数据库awaitinit_database()if__name__=="__main__":asyncio.run(main())

4. AI核心组件实现

4.1 NLP处理器（意图识别与情感分析）

创建src/ai_components/nlp/processor.py：

""" NLP处理器 负责意图识别、情感分析、实体提取等自然语言处理任务 """importasyncioimportloggingfromtypingimportDict,List,Optional,Tuple,Anyfromdataclassesimportdataclassimportnumpyasnpfromdatetimeimportdatetime# 导入必要的NLP库try:importjiebaimportjieba.analyse JIEBA_AVAILABLE=TrueexceptImportError:JIEBA_AVAILABLE=Falseprint("⚠️ jieba未安装，中文分词功能将受限")try:fromtransformersimport(AutoTokenizer,AutoModelForSequenceClassification,pipeline,BertTokenizer,BertModel)fromsentence_transformersimportSentenceTransformer TRANSFORMERS_AVAILABLE=TrueexceptImportError:TRANSFORMERS_AVAILABLE=Falseprint("⚠️ transformers/sentence-transformers未安装")fromsrc.core.configimportsettings@dataclassclassIntentResult:"""意图识别结果"""intent:strconfidence:floatentities:Dict[str,Any]alternatives:List[Dict[str,float]]@dataclassclassSentimentResult:"""情感分析结果"""sentiment:str# positive, negative, neutralscore:floatconfidence:floatemotions:Dict[str,float]# 细分情绪@dataclassclassNLPAnalysis:"""NLP分析综合结果"""text:strintent_result:IntentResult sentiment_result:SentimentResult entities:Dict[str,Any]keywords:List[str]language:strprocessing_time_ms:floatclassNLPProcessor:"""NLP处理器主类"""def__init__(self):self.logger=logging.getLogger(__name__)self.initialized=Falseself.models={}self.tokenizers={}# 配置self.confidence_threshold=0.6self.max_sequence_length=512asyncdefinitialize(self):"""初始化NLP模型"""ifself.initialized:returnself.logger.info("初始化NLP处理器...")try:# 初始化分词器ifJIEBA_AVAILABLE:# 加载自定义词典jieba.set_dictionary('./data/dict.txt')jieba.initialize()# 初始化句子编码器ifTRANSFORMERS_AVAILABLE:self.logger.info("加载Sentence Transformer模型...")self.sentence_model=SentenceTransformer(settings.SENTENCE_TRANSFORMER_MODEL,cache_folder=settings.AI_MODEL_PATH)# 加载情感分析模型self.logger.info("加载情感分析模型...")self.sentiment_analyzer=pipeline("sentiment-analysis",model="nlptown/bert-base-multilingual-uncased-sentiment",device=0ifsettings.USE_GPUelse-1)# 加载BERT模型用于特征提取self.logger.info("加载BERT模型...")self.bert_tokenizer=BertTokenizer.from_pretrained(settings.BERT_MODEL_PATH,cache_dir=settings.AI_MODEL_PATH)self.bert_model=BertModel.from_pretrained(settings.BERT_MODEL_PATH,cache_dir=settings.AI_MODEL_PATH)ifsettings.USE_GPU:self.bert_model.cuda()# 加载意图分类模型（示例，实际需要训练）self.intent_patterns=self._load_intent_patterns()self.initialized=Trueself.logger.info("NLP处理器初始化完成")exceptExceptionase:self.logger.error(f"NLP处理器初始化失败:{e}")raisedef_load_intent_patterns(self)->Dict[str,List[str]]:"""加载意图模式（实际应从数据库加载）"""return{"greeting":["你好","嗨","早上好","晚上好","在吗","有人吗"],"farewell":["再见","拜拜","谢谢","结束","好了"],"product_inquiry":["产品","价格","功能","特性","规格","多少钱"],"technical_support":["问题","错误","故障","无法","不能","帮助"],"complaint":["投诉","不满","差评","退款","赔偿","生气"],"payment_issue":["支付","付款","扣款","退款","发票","账单"],"schedule":["时间","预约","安排","什么时候","多久"],"location":["地址","位置","在哪里","怎么去","门店"],"account":["登录","注册","密码","账户","会员"],"other":[]}asyncdefanalyze_text(self,text:str,context:Optional[Dict]=None)->NLPAnalysis:""" 综合分析文本 Args: text: 待分析文本 context: 上下文信息 Returns: NLPAnalysis对象 """start_time=datetime.now()ifnotself.initialized:awaitself.initialize()# 检测语言language=self._detect_language(text)# 并行执行各种分析intent_task=asyncio.create_task(self.detect_intent(text,context))sentiment_task=asyncio.create_task(self.analyze_sentiment(text))entity_task=asyncio.create_task(self.extract_entities(text))keyword_task=asyncio.create_task(self.extract_keywords(text,language))# 等待所有任务完成intent_result=awaitintent_task sentiment_result=awaitsentiment_task entities=awaitentity_task keywords=awaitkeyword_task# 计算处理时间processing_time=(datetime.now()-start_time).total_seconds()*1000returnNLPAnalysis(text=text,intent_result=intent_result,sentiment_result=sentiment_result,entities=entities,keywords=keywords,language=language,processing_time_ms=processing_time)asyncdefdetect_intent(self,text:str,context:Optional[Dict]=None)->IntentResult:""" 检测用户意图 Args: text: 用户输入文本 context: 对话上下文 Returns: IntentResult对象 """try:# 方法1：基于规则的匹配（快速，准确率高）rule_based_result=self._rule_based_intent_detection(text)# 方法2：基于相似度的匹配（使用句子编码器）similarity_based_result=awaitself._similarity_based_intent_detection(text)# 方法3：基于机器学习的分类（如果有训练好的模型）# ml_based_result = await self._ml_based_intent_detection(text)# 融合结果（加权平均）final_intent,final_confidence,alternatives=self._fuse_intent_results(rule_based_result,similarity_based_result)# 提取实体entities=awaitself.extract_entities(text)# 考虑上下文ifcontext:final_intent=self._apply_context_rules(final_intent,context)returnIntentResult(intent=final_intent,confidence=final_confidence,entities=entities,alternatives=alternatives)exceptExceptionase:self.logger.error(f"意图检测失败:{e}")returnIntentResult(intent="unknown",confidence=0.0,entities={},alternatives=[])def_rule_based_intent_detection(self,text:str)->Tuple[str,float]:"""基于规则的意图检测"""text_lower=text.lower()best_intent="other"best_score=0.0forintent,patternsinself.intent_patterns.items():score=0forpatterninpatterns:ifpatternintext_lower:score+=1ifscore>0:pattern_count=len(patterns)normalized_score=score/pattern_countifnormalized_score>best_score:best_score=normalized_score best_intent=intentreturnbest_intent,best_scoreasyncdef_similarity_based_intent_detection(self,text:str)->Tuple[str,float]:"""基于相似度的意图检测"""ifnotTRANSFORMERS_AVAILABLE:return"other",0.0try:# 编码输入文本text_embedding=self.sentence_model.encode(text)# 计算与每个意图示例的相似度intent_scores={}forintent,examplesinself.intent_patterns.items():ifnotexamples:continue# 编码所有示例example_embeddings=self.sentence_model.encode(examples)# 计算相似度similarities=np.dot(example_embeddings,text_embedding.T)max_similarity=np.max(similarities)# 标准化到0-1范围normalized_score=float((max_similarity+1)/2)intent_scores[intent]=normalized_score# 找到最佳匹配ifintent_scores:best_intent=max(intent_scores,key=intent_scores.get)best_score=intent_scores[best_intent]returnbest_intent,best_scorereturn"other",0.0exceptExceptionase:self.logger.error(f"相似度意图检测失败:{e}")return"other",0.0def_fuse_intent_results(self,rule_result:Tuple[str,float],similarity_result:Tuple[str,float])->Tuple[str,float,List[Dict]]:"""融合不同方法的结果"""rule_intent,rule_score=rule_result sim_intent,sim_score=similarity_result# 权重设置（可调整）rule_weight=0.6sim_weight=0.4# 计算加权得分scores={}scores[rule_intent]=rule_score*rule_weight scores[sim_intent]=scores.get(sim_intent,0)+sim_score*sim_weight# 找到最佳意图best_intent=max(scores,key=scores.get)best_score=scores[best_intent]# 生成备选意图alternatives=[{"intent":intent,"score":score}forintent,scoreinscores.items()ifintent!=best_intentandscore>0.1]alternatives.sort(key=lambdax:x["score"],reverse=True)returnbest_intent,best_score,alternativesdef_apply_context_rules(self,intent:str,context:Dict)->str:"""应用上下文规则调整意图"""# 示例规则：如果连续3次都是greeting，可能是有问题ifintent=="greeting"andcontext.get("greeting_count",0)>=3:return"technical_support"# 示例规则：如果刚刚完成支付相关对话，现在问时间可能是问发货时间ifintent=="schedule"andcontext.get("last_intent")=="payment_issue":return"shipping_inquiry"returnintentasyncdefanalyze_sentiment(self,text:str)->SentimentResult:""" 分析文本情感 Args: text: 待分析文本 Returns: SentimentResult对象 """try:ifnotTRANSFORMERS_AVAILABLE:# 回退到简单的情感分析returnself._simple_sentiment_analysis(text)# 使用transformers情感分析results=self.sentiment_analyzer(text)ifresults:result=results[0]label=result['label']score=result['score']# 转换标签iflabelin["POSITIVE","5 stars","4 stars"]:sentiment="positive"eliflabelin["NEGATIVE","1 star","2 stars"]:sentiment="negative"else:sentiment="neutral"# 简单情绪分析emotions=self._analyze_emotions(text)returnSentimentResult(sentiment=sentiment,score=score,confidence=score,emotions=emotions)returnSentimentResult(sentiment="neutral",score=0.5,confidence=0.0,emotions={})exceptExceptionase:self.logger.error(f"情感分析失败:{e}")returnSentimentResult(sentiment="neutral",score=0.5,confidence=0.0,emotions={})def_simple_sentiment_analysis(self,text:str)->SentimentResult:"""简单的情感分析（基于关键词）"""positive_words=["好","不错","满意","喜欢","棒","赞","感谢"]negative_words=["差","不好","不满意","讨厌","垃圾","投诉","生气"]positive_count=sum(1forwordinpositive_wordsifwordintext)negative_count=sum(1forwordinnegative_wordsifwordintext)total=positive_count+negative_countiftotal==0:returnSentimentResult(sentiment="neutral",score=0.5,confidence=0.0,emotions={})sentiment_score=positive_count/totalifsentiment_score>0.6:sentiment="positive"elifsentiment_score<0.4:sentiment="negative"else:sentiment="neutral"returnSentimentResult(sentiment=sentiment,score=sentiment_score,confidence=min(sentiment_score,1-sentiment_score)*2,emotions={})def_analyze_emotions(self,text:str)->Dict[str,float]:"""分析细分情绪"""emotion_keywords={"happy":["开心","高兴","快乐","愉快","兴奋"],"angry":["生气","愤怒","恼火","不爽","气愤"],"sad":["伤心","难过","悲伤","失望","沮丧"],"surprise":["惊讶","惊奇","意外","没想到"],"fear":["害怕","担心","恐惧","紧张"],"disgust":["恶心","讨厌","嫌弃","反感"]}emotions={}foremotion,keywordsinemotion_keywords.items():count=sum(1forkeywordinkeywordsifkeywordintext)ifcount>0:emotions[emotion]=count/len(keywords)returnemotionsasyncdefextract_entities(self,text:str)->Dict[str,Any]:""" 提取命名实体 Args: text: 待分析文本 Returns: 实体字典 """entities={"products":[],"numbers":[],"dates":[],"locations":[],"people":[],"organizations":[]}# 简单实体提取（实际应该使用NER模型）importre# 提取数字numbers=re.findall(r'\d+',text)entities["numbers"]=numbers# 提取产品名（简单实现）product_keywords=["产品","服务","软件","系统","应用"]forkeywordinproduct_keywords:ifkeywordintext:# 提取产品名周围的文本idx=text.find(keyword)start=max(0,idx-10)end=min(len(text),idx+10)entities["products"].append(text[start:end])# 提取日期date_patterns=[r'\d{4}年\d{1,2}月\d{1,2}日',r'\d{1,2}月\d{1,2}日',r'\d{4}-\d{2}-\d{2}',r'今天|明天|后天|昨天']forpatternindate_patterns:dates=re.findall(pattern,text)entities["dates"].extend(dates)returnentitiesasyncdefextract_keywords(self,text:str,language:str="zh")->List[str]:""" 提取关键词 Args: text: 待分析文本 language: 语言 Returns: 关键词列表 """iflanguage=="zh"andJIEBA_AVAILABLE:# 使用jieba提取中文关键词keywords=jieba.analyse.extract_tags(text,topK=10,withWeight=False,allowPOS=('n','nr','ns','nt','nz','v','a'))returnkeywordselse:# 简单分词（适用于英文）words=text.lower().split()# 移除停用词stop_words={"the","a","an","and","or","but","in","on","at","to","for"}keywords=[wordforwordinwordsifwordnotinstop_words]returnkeywords[:10]def_detect_language(self,text:str)->str:"""检测文本语言"""# 简单语言检测（基于字符范围）importre# 检查中文字符ifre.search(r'[\u4e00-\u9fff]',text):return"zh"# 检查英文字符ifre.search(r'[a-zA-Z]',text):return"en"# 默认返回中文return"zh"asyncdefget_text_embedding(self,text:str)->np.ndarray:""" 获取文本的向量嵌入 Args: text: 输入文本 Returns: 文本向量 """ifnotTRANSFORMERS_AVAILABLE:raiseRuntimeError("transformers库未安装")ifnotself.initialized:awaitself.initialize()embedding=self.sentence_model.encode(text)returnembeddingasyncdefbatch_analyze(self,texts:List[str])->List[NLPAnalysis]:""" 批量分析文本 Args: texts: 文本列表 Returns: 分析结果列表 """tasks=[self.analyze_text(text)fortextintexts]results=awaitasyncio.gather(*tasks)returnresults# 创建全局NLP处理器实例nlp_processor=NLPProcessor()

4.2 知识检索系统

创建src/ai_components/retrieval/engine.py：

""" 知识检索引擎 基于向量相似度从知识库中检索相关信息 """importasyncioimportloggingimportjsonfromtypingimportList,Dict,Optional,Any,Tuplefromdatetimeimportdatetimeimportnumpyasnpfromdataclassesimportdataclass# 导入向量数据库库try:importfaiss FAISS_AVAILABLE=TrueexceptImportError:FAISS_AVAILABLE=Falseprint("⚠️ faiss未安装，向量检索功能将受限")try:importchromadbfromchromadb.configimportSettings CHROMA_AVAILABLE=TrueexceptImportError:CHROMA_AVAILABLE=Falseprint("⚠️ chromadb未安装，向量数据库功能将受限")fromsrc.core.configimportsettingsfromsrc.ai_components.nlp.processorimportnlp_processor@dataclassclassSearchResult:"""搜索结果"""id:strcontent:strscore:floatmetadata:Dict[str,Any]source:str# knowledge_base, faq, document, etc.@dataclassclassRetrievalResponse:"""检索响应"""query:strresults:List[SearchResult]suggested_questions:List[str]processing_time_ms:floatclassKnowledgeRetrievalEngine:"""知识检索引擎"""def__init__(self):self.logger=logging.getLogger(__name__)self.initialized=False# 向量索引self.faiss_index=Noneself.id_to_content={}self.embedding_dim=384# all-MiniLM-L6-v2的维度# ChromaDB客户端self.chroma_client=Noneself.chroma_collection=None# 缓存self.cache={}self.cache_size=1000asyncdefinitialize(self):"""初始化检索引擎"""ifself.initialized:returnself.logger.info("初始化知识检索引擎...")try:# 初始化NLP处理器awaitnlp_processor.initialize()# 初始化向量数据库ifCHROMA_AVAILABLE:awaitself._initialize_chromadb()# 初始化FAISS索引ifFAISS_AVAILABLE:awaitself._initialize_faiss()# 从数据库加载知识库数据awaitself._load_knowledge_base()self.initialized=Trueself.logger.info("知识检索引擎初始化完成")exceptExceptionase:self.logger.error(f"知识检索引擎初始化失败:{e}")raiseasyncdef_initialize_chromadb(self):"""初始化ChromaDB"""self.logger.info("初始化ChromaDB...")# 创建ChromaDB客户端self.chroma_client=chromadb.Client(Settings(chroma_db_impl="duckdb+parquet",persist_directory="./data/chroma_db"))# 创建或获取集合collection_name="knowledge_base"try:self.chroma_collection=self.chroma_client.get_collection(collection_name)self.logger.info(f"加载现有集合:{collection_name}")except:self.chroma_collection=self.chroma_client.create_collection(name=collection_name,metadata={"description":"客服知识库"})self.logger.info(f"创建新集合:{collection_name}")asyncdef_initialize_faiss(self):"""初始化FAISS索引"""self.logger.info("初始化FAISS索引...")# 创建Flat索引（精确搜索，适合中小规模数据）self.faiss_index=faiss.IndexFlatIP(self.embedding_dim)# 如果有GPU，使用GPU加速ifsettings.USE_GPUandhasattr(faiss,'StandardGpuResources'):self.logger.info("启用FAISS GPU加速...")res=faiss.StandardGpuResources()self.faiss_index=faiss.index_cpu_to_gpu(res,0,self.faiss_index)asyncdef_load_knowledge_base(self):"""从数据库加载知识库"""self.logger.info("加载知识库数据...")# 这里应该从数据库加载知识库条目# 示例：模拟加载一些数据# 实际应该从数据库查询sample_knowledge=[{"id":"kb_001","content":"我们的工作时间是周一至周五9:00-18:00，周末休息。","category":"general","metadata":{"source":"faq","importance":5}},{"id":"kb_002","content":"产品退款需要在购买后30天内申请，退款将在7-14个工作日内处理。","category":"policy","metadata":{"source":"policy","importance":8}},{"id":"kb_003","content":"如果忘记密码，可以点击登录页面的'忘记密码'链接，通过邮箱重置。","category":"account","metadata":{"source":"faq","importance":7}},{"id":"kb_004","content":"技术支持电话：400-123-4567，工作时间随时可拨打。","category":"contact","metadata":{"source":"contact","importance":9}},{"id":"kb_005","content":"产品支持7天无理由退货，但需要商品完好、包装完整。","category":"policy","metadata":{"source":"policy","importance":6}}]# 添加到向量数据库awaitself.add_documents(sample_knowledge)self.logger.info(f"已加载{len(sample_knowledge)}条知识库数据")asyncdefadd_documents(self,documents:List[Dict[str,Any]]):""" 添加文档到知识库 Args: documents: 文档列表，每个文档包含id, content, metadata """ifnotdocuments:return# 生成文档向量contents=[doc["content"]fordocindocuments]embeddings=awaitself._generate_embeddings(contents)# 添加到FAISS索引ifFAISS_AVAILABLEandself.faiss_indexisnotNone:# 转换为numpy数组embeddings_np=np.array(embeddings).astype('float32')# 添加到索引self.faiss_index.add(embeddings_np)# 保存id到内容的映射fori,docinenumerate(documents):doc_id=doc["id"]self.id_to_content[doc_id]={"content":doc["content"],"metadata":doc.get("metadata",{}),"index":len(self.id_to_content)}# 添加到ChromaDBifCHROMA_AVAILABLEandself.chroma_collectionisnotNone:try:self.chroma_collection.add(embeddings=embeddings,documents=contents,metadatas=[doc.get("metadata",{})fordocindocuments],ids=[doc["id"]fordocindocuments])self.logger.info(f"已添加{len(documents)}个文档到ChromaDB")exceptExceptionase:self.logger.error(f"添加到ChromaDB失败:{e}")asyncdefsearch(self,query:str,top_k:int=5,threshold:float=0.5)->RetrievalResponse:""" 搜索相关知识 Args: query: 查询文本 top_k: 返回结果数量 threshold: 相似度阈值 Returns: RetrievalResponse对象 """start_time=datetime.now()ifnotself.initialized:awaitself.initialize()# 检查缓存cache_key=f"{query}_{top_k}_{threshold}"ifcache_keyinself.cache:self.logger.debug("从缓存返回搜索结果")returnself.cache[cache_key]# 分析查询query_analysis=awaitnlp_processor.analyze_text(query)query_embedding=awaitnlp_processor.get_text_embedding(query)# 执行搜索results=[]# 方法1：向量相似度搜索vector_results=awaitself._vector_search(query_embedding,query,top_k,threshold)results.extend(vector_results)# 方法2：关键词匹配搜索keyword_results=awaitself._keyword_search(query_analysis.keywords,top_k,threshold)results.extend(keyword_results)# 方法3：基于意图的搜索intent_results=awaitself._intent_based_search(query_analysis.intent_result.intent,top_k)results.extend(intent_results)# 去重和排序unique_results=self._deduplicate_and_sort(results)# 取前top_k个结果final_results=unique_results[:top_k]# 生成建议问题suggested_questions=awaitself._generate_suggested_questions(query,query_analysis,final_results)# 计算处理时间processing_time=(datetime.now()-start_time).total_seconds()*1000response=RetrievalResponse(query=query,results=final_results,suggested_questions=suggested_questions,processing_time_ms=processing_time)# 缓存结果self._add_to_cache(cache_key,response)returnresponseasyncdef_vector_search(self,query_embedding:np.ndarray,query:str,top_k:int,threshold:float)->List[SearchResult]:"""向量相似度搜索"""results=[]# 使用FAISS搜索ifFAISS_AVAILABLEandself.faiss_indexisnotNone:try:# 准备查询向量query_vector=np.array([query_embedding]).astype('float32')# 搜索distances,indices=self.faiss_index.search(query_vector,top_k*2)# 处理结果foriinrange(len(indices[0])):idx=indices[0][i]distance=distances[0][i]# 转换为相似度分数score=float((distance+1)/2)# 假设cosine相似度ifscore>=threshold:# 查找对应文档fordoc_id,doc_infoinself.id_to_content.items():ifdoc_info["index"]==idx:results.append(SearchResult(id=doc_id,content=doc_info["content"],score=score,metadata=doc_info["metadata"],source="knowledge_base"))breakself.logger.debug(f"FAISS搜索找到{len(results)}个结果")exceptExceptionase:self.logger.error(f"FAISS搜索失败:{e}")# 使用ChromaDB搜索ifCHROMA_AVAILABLEandself.chroma_collectionisnotNone:try:chroma_results=self.chroma_collection.query(query_embeddings=[query_embedding.tolist()],n_results=top_k,include=["documents","metadatas","distances"])ifchroma_resultsandchroma_results["documents"]:foriinrange(len(chroma_results["documents"][0])):doc=chroma_results["documents"][0][i]metadata=chroma_results["metadatas"][0][i]distance=chroma_results["distances"][0][i]# 转换为相似度分数score=float(1.0-distance)ifdistanceelse0.5ifscore>=threshold:doc_id=f"chroma_{i}"results.append(SearchResult(id=doc_id,content=doc,score=score,metadata=metadataor{},source="knowledge_base"))self.logger.debug(f"ChromaDB搜索找到{len(results)}个结果")exceptExceptionase:self.logger.error(f"ChromaDB搜索失败:{e}")returnresultsasyncdef_keyword_search(self,keywords:List[str],top_k:int,threshold:float)->List[SearchResult]:"""关键词搜索"""results=[]ifnotkeywords:returnresults# 简单实现：在内容中搜索关键词fordoc_id,doc_infoinself.id_to_content.items():content=doc_info["content"].lower()metadata=doc_info["metadata"]# 计算关键词匹配度match_count=0forkeywordinkeywords:ifkeyword.lower()incontent:match_count+=1ifmatch_count>0:score=match_count/len(keywords)ifscore>=threshold:results.append(SearchResult(id=doc_id,content=doc_info["content"],score=score,metadata=metadata,source="knowledge_base"))# 限制结果数量results.sort(key=lambdax:x.score,reverse=True)returnresults[:top_k]asyncdef_intent_based_search(self,intent:str,top_k:int)->List[SearchResult]:"""基于意图的搜索"""results=[]# 根据意图类别搜索相关文档intent_categories={"product_inquiry":["product","features","specification"],"technical_support":["technical","help","support","troubleshooting"],"payment_issue":["payment","refund","billing","invoice"],"complaint":["complaint","issue","problem","dissatisfied"],"account":["account","login","password","register"]}categories=intent_categories.get(intent,[])ifnotcategories:returnresultsfordoc_id,doc_infoinself.id_to_content.items():metadata=doc_info["metadata"]doc_categories=metadata.get("categories",[])ifisinstance(doc_categories,str):doc_categories=[doc_categories]# 检查类别匹配forcatincategories:ifcatindoc_categories:results.append(SearchResult(id=doc_id,content=doc_info["content"],score=0.7,# 中等置信度metadata=metadata,source="knowledge_base"))break# 限制结果数量returnresults[:top_k]def_deduplicate_and_sort(self,results:List[SearchResult])->List[SearchResult]:"""去重和排序"""seen_contents=set()unique_results=[]forresultinresults:# 简单去重：基于内容前50个字符content_prefix=result.content[:50]ifcontent_prefixnotinseen_contents:seen_contents.add(content_prefix)unique_results.append(result)# 按分数排序unique_results.sort(key=lambdax:x.score,reverse=True)returnunique_resultsasyncdef_generate_suggested_questions(self,query:str,query_analysis:Any,search_results:List[SearchResult])->List[str]:"""生成建议问题"""suggestions=[]# 基于意图的建议intent=query_analysis.intent_result.intent intent_suggestions={"product_inquiry":["这个产品有什么功能？","价格是多少？","有优惠活动吗？","怎么购买？"],"technical_support":["常见问题有哪些？","怎么联系技术支持？","系统要求是什么？","有用户手册吗？"],"payment_issue":["退款流程是怎样的？","支持哪些支付方式？","发票怎么开？","付款遇到问题怎么办？"],"account":["怎么注册账号？","忘记密码怎么办？","怎么修改个人信息？","账号安全如何保障？"]}ifintentinintent_suggestions:suggestions.extend(intent_suggestions[intent][:2])# 基于搜索结果的建议forresultinsearch_results[:2]:# 从内容中提取可能的问题content=result.content sentences=content.split('。')[:2]forsentenceinsentences:ifsentenceandlen(sentence)>10:suggestions.append(f"关于「{sentence[:20]}...」")# 去重unique_suggestions=list(dict.fromkeys(suggestions))returnunique_suggestions[:5]asyncdef_generate_embeddings(self,texts:List[str])->List[List[float]]:"""生成文本向量"""embeddings=[]fortextintexts:embedding=awaitnlp_processor.get_text_embedding(text)embeddings.append(embedding.tolist())returnembeddingsdef_add_to_cache(self,key:str,value:RetrievalResponse):"""添加到缓存"""iflen(self.cache)>=self.cache_size:# 移除最旧的缓存项oldest_key=next(iter(self.cache))delself.cache[oldest_key]self.cache[key]=valueasyncdefupdate_document(self,doc_id:str,content:str,metadata:Dict):"""更新文档"""# 从缓存中删除相关查询keys_to_remove=[kforkinself.cache.keys()ifdoc_idink]forkeyinkeys_to_remove:delself.cache[key]# 更新向量数据库（需要重新添加）document={"id":doc_id,"content":content,"metadata":metadata}# 注意：实际实现需要先从索引中删除旧文档，再添加新文档awaitself.add_documents([document])asyncdefdelete_document(self,doc_id:str):"""删除文档"""# 从缓存中删除相关查询keys_to_remove=[kforkinself.cache.keys()ifdoc_idink]forkeyinkeys_to_remove:delself.cache[key]# 从向量数据库中删除ifCHROMA_AVAILABLEandself.chroma_collectionisnotNone:try:self.chroma_collection.delete(ids=[doc_id])self.logger.info(f"已从ChromaDB删除文档:{doc_id}")exceptExceptionase:self.logger.error(f"从ChromaDB删除文档失败:{e}")# 创建全局检索引擎实例retrieval_engine=KnowledgeRetrievalEngine()

5. 对话管理系统

5.1 对话状态管理

创建src/services/conversation/manager.py：

""" 对话管理器 负责对话状态管理、上下文维护和流程控制 """importasyncioimportjsonimportloggingfromtypingimportDict,List,Optional,Any,Tuplefromdatetimeimportdatetime,timedeltafromenumimportEnumimportuuidfromsrc.core.configimportsettingsfromsrc.models.baseimportConversation,Messagefromsrc.ai_components.nlp.processorimportnlp_processor,NLPAnalysisfromsrc.ai_components.retrieval.engineimportretrieval_engine,RetrievalResponseclassConversationState(Enum):"""对话状态枚举"""ACTIVE="active"# 活跃状态PENDING="pending"# 等待人工TRANSFERRING="transferring"# 转接中CLOSED="closed"# 已关闭ESCALATED="escalated"# 已升级classHandlerType(Enum):"""处理器类型枚举"""AI="ai"# AI处理AGENT="agent"# 人工处理HYBRID="hybrid"# 混合处理classConversationContext:"""对话上下文"""def__init__(self,conversation_id:str,customer_id:str):self.conversation_id=conversation_id self.customer_id=customer_id self.messages:List[Dict]=[]self.state_history:List[Dict]=[]self.intent_history:List[Dict]=[]self.sentiment_history:List[Dict]=[]self.metadata:Dict[str,Any]={}self.created_at=datetime.now()self.updated_at=datetime.now()# 统计信息self.message_count=0self.ai_response_count=0self.agent_response_count=0self.transfer_count=0self.escalation_count=0# 时间相关self.last_message_time=Noneself.last_ai_response_time=Noneself.last_agent_response_time=None# 缓存self.cached_responses:Dict[str,Any]={}defadd_message(self,message:Dict):"""添加消息到上下文"""self.messages.append(message)self.message_count+=1self.last_message_time=datetime.now()self.updated_at=datetime.now()# 限制历史消息数量iflen(self.messages)>settings.MAX_CONVERSATION_HISTORY:self.messages=self.messages[-settings.MAX_CONVERSATION_HISTORY:]defadd_state_change(self,old_state:str,new_state:str,reason:str):"""添加状态变更记录"""self.state_history.append({"timestamp":datetime.now().isoformat(),"old_state":old_state,"new_state":new_state,"reason":reason})defadd_intent(self,intent:str,confidence:float,text:str):"""添注意图记录"""self.intent_history.append({"timestamp":datetime.now().isoformat(),"intent":intent,"confidence":confidence,"text":text})defadd_sentiment(self,sentiment:str,score:float):"""添加情感记录"""self.sentiment_history.append({"timestamp":datetime.now().isoformat(),"sentiment":sentiment,"score":score})defget_recent_messages(self,count:int=10)->List[Dict]:"""获取最近的消息"""returnself.messages[-count:]ifself.messageselse[]defget_conversation_summary(self)->Dict[str,Any]:"""获取对话摘要"""return{"conversation_id":self.conversation_id,"customer_id":self.customer_id,"message_count":self.message_count,"ai_response_count":self.ai_response_count,"agent_response_count":self.agent_response_count,"transfer_count":self.transfer_count,"escalation_count":self.escalation_count,"current_state":self.metadata.get("current_state","active"),"created_at":self.created_at.isoformat(),"updated_at":self.updated_at.isoformat(),"duration_seconds":(datetime.now()-self.created_at).total_seconds()}defis_timed_out(self)->bool:"""检查对话是否超时"""ifnotself.last_message_time:returnFalsetimeout_minutes=settings.SESSION_TIMEOUT_MINUTES timeout_delta=timedelta(minutes=timeout_minutes)returndatetime.now()-self.last_message_time>timeout_deltadefto_dict(self)->Dict[str,Any]:"""转换为字典"""return{"conversation_id":self.conversation_id,"customer_id":self.customer_id,"messages":self.messages,"state_history":self.state_history,"intent_history":self.intent_history,"sentiment_history":self.sentiment_history,"metadata":self.metadata,"statistics":{"message_count":self.message_count,"ai_response_count":self.ai_response_count,"agent_response_count":self.agent_response_count,"transfer_count":self.transfer_count,"escalation_count":self.escalation_count},"timestamps":{"created_at":self.created_at.isoformat(),"updated_at":self.updated_at.isoformat(),"last_message_time":self.last_message_time.isoformat()ifself.last_message_timeelseNone}}classConversationManager:"""对话管理器"""def__init__(self):self.logger=logging.getLogger(__name__)self.conversations:Dict[str,ConversationContext]={}self.customer_conversations:Dict[str,List[str]]={}# 配置self.max_conversations=10000self.cleanup_interval=300# 5分钟清理一次# 启动清理任务asyncio.create_task(self._cleanup_task())asyncdefcreate_conversation(self,customer_id:str,channel:str,metadata:Optional[Dict]=None)->ConversationContext:""" 创建新对话 Args: customer_id: 客户ID channel: 渠道 metadata: 元数据 Returns: 对话上下文 """conversation_id=str(uuid.uuid4())# 创建上下文context=ConversationContext(conversation_id,customer_id)context.metadata.update({"channel":channel,"current_state":ConversationState.ACTIVE.value,"created_channel":channel,"metadata":metadataor{}})# 保存到内存self.conversations[conversation_id]=context# 更新客户对话映射ifcustomer_idnotinself.customer_conversations:self.customer_conversations[customer_id]=[]self.customer_conversations[customer_id].append(conversation_id)# 限制每个客户的对话数量iflen(self.customer_conversations[customer_id])>10:# 移除最旧的对话old_conv_id=self.customer_conversations[customer_id].pop(0)ifold_conv_idinself.conversations:delself.conversations[old_conv_id]self.logger.info(f"创建新对话:{conversation_id}, 客户:{customer_id}, 渠道:{channel}")returncontextasyncdefprocess_message(self,conversation_id:str,message:Dict)->Dict[str,Any]:""" 处理消息 Args: conversation_id: 对话ID message: 消息数据 Returns: 处理结果 """start_time=datetime.now()# 获取对话上下文context=self.conversations.get(conversation_id)ifnotcontext:raiseValueError(f"对话不存在:{conversation_id}")# 更新最后消息时间context.last_message_time=datetime.now()# 添加消息到上下文context.add_message(message)# 分析消息text=message.get("content","")nlp_analysis=awaitnlp_processor.analyze_text(text)# 更新上下文记录context.add_intent(nlp_analysis.intent_result.intent,nlp_analysis.intent_result.confidence,text)context.add_sentiment(nlp_analysis.sentiment_result.sentiment,nlp_analysis.sentiment_result.score)# 确定处理器类型handler_type=awaitself._determine_handler_type(context,nlp_analysis,message)# 处理消息ifhandler_type==HandlerType.AI:response=awaitself._handle_with_ai(context,nlp_analysis,message)context.ai_response_count+=1context.last_ai_response_time=datetime.now()elifhandler_type==HandlerType.AGENT:response=awaitself._handle_with_agent(context,nlp_analysis,message)context.agent_response_count+=1context.last_agent_response_time=datetime.now()elifhandler_type==HandlerType.HYBRID:response=awaitself._handle_hybrid(context,nlp_analysis,message)else:response={"type":"error","content":"无法确定处理方式","handler":"unknown"}# 更新对话状态awaitself._update_conversation_state(context,response)# 计算处理时间processing_time=(datetime.now()-start_time).total_seconds()*1000# 构建响应result={"conversation_id":conversation_id,"message_id":message.get("id"),"handler_type":handler_type.value,"response":response,"analysis":{"intent":nlp_analysis.intent_result.intent,"confidence":nlp_analysis.intent_result.confidence,"sentiment":nlp_analysis.sentiment_result.sentiment,"sentiment_score":nlp_analysis.sentiment_result.score,"entities":nlp_analysis.entities,"keywords":nlp_analysis.keywords},"processing_time_ms":processing_time,"context_summary":context.get_conversation_summary()}returnresultasyncdef_determine_handler_type(self,context:ConversationContext,nlp_analysis:NLPAnalysis,message:Dict)->HandlerType:""" 确定处理器类型 Args: context: 对话上下文 nlp_analysis: NLP分析结果 message: 消息 Returns: 处理器类型 """current_state=context.metadata.get("current_state")# 如果已经在转接或人工状态，继续人工处理ifcurrent_statein[ConversationState.PENDING.value,ConversationState.TRANSFERRING.value,ConversationState.ESCALATED.value]:returnHandlerType.AGENT# 检查是否明确要求人工text=message.get("content","").lower()ifany(keywordintextforkeywordin["人工","真人","客服","转人工"]):returnHandlerType.AGENT# 检查情感：负面情感转人工ifnlp_analysis.sentiment_result.sentiment=="negative":sentiment_score=nlp_analysis.sentiment_result.scoreifsentiment_score>0.7:# 强烈负面returnHandlerType.AGENTelifsentiment_score>0.5:# 中度负面returnHandlerType.HYBRID# 检查意图置信度intent_confidence=nlp_analysis.intent_result.confidence intent_name=nlp_analysis.intent_result.intent# 获取意图配置（应该从数据库获取）intent_configs={"complaint":{"threshold":0.7,"handler":HandlerType.AGENT},"technical_support":{"threshold":0.6,"handler":HandlerType.HYBRID},"payment_issue":{"threshold":0.65,"handler":HandlerType.HYBRID},"product_inquiry":{"threshold":0.5,"handler":HandlerType.AI},"greeting":{"threshold":0.4,"handler":HandlerType.AI},"other":{"threshold":0.3,"handler":HandlerType.AI}}config=intent_configs.get(intent_name,intent_configs["other"])# 如果置信度低于阈值，考虑转人工ifintent_confidence<config["threshold"]:# 检查是否有历史对话可以提供上下文recent_messages=context.get_recent_messages(3)iflen(recent_messages)>=2:# 如果最近有成功交互，继续AI处理returnHandlerType.AIelse:# 新对话且置信度低，转人工或混合returnHandlerType.HYBRID# 检查对话长度：长对话可能需要人工介入ifcontext.message_count>15:returnHandlerType.HYBRID# 检查转移历史：频繁转移需要人工ifcontext.transfer_count>2:returnHandlerType.AGENT# 默认返回配置的处理器类型returnconfig["handler"]asyncdef_handle_with_ai(self,context:ConversationContext,nlp_analysis:NLPAnalysis,message:Dict)->Dict[str,Any]:"""使用AI处理消息"""try:text=message.get("content","")# 检索相关知识retrieval_result=awaitretrieval_engine.search(text)# 构建AI提示prompt=self._build_ai_prompt(context,text,nlp_analysis,retrieval_result)# 调用AI生成响应（这里简化，实际应该调用AI模型）ai_response=awaitself._generate_ai_response(prompt)# 构建响应response={"type":"ai_response","content":ai_response,"confidence":nlp_analysis.intent_result.confidence,"sources":[r.content[:100]forrinretrieval_result.results[:3]],"suggested_questions":retrieval_result.suggested_questions,"handler":"ai"}returnresponseexceptExceptionase:self.logger.error(f"AI处理失败:{e}")return{"type":"error","content":"抱歉，AI暂时无法处理您的请求，正在为您转接人工客服。","handler":"ai_fallback"}asyncdef_handle_with_agent(self,context:ConversationContext,nlp_analysis:NLPAnalysis,message:Dict)->Dict[str,Any]:"""使用人工处理消息"""try:# 获取可用的坐席available_agents=awaitself._find_available_agents(context,nlp_analysis)ifavailable_agents:# 分配坐席assigned_agent=awaitself._assign_agent(context,available_agents[0])response={"type":"agent_assigned","content":f"正在为您转接人工客服，坐席{assigned_agent['name']}将为您服务。","agent_id":assigned_agent["id"],"agent_name":assigned_agent["name"],"estimated_wait_time":30,# 预估等待时间（秒）"handler":"agent_routing"}else:# 没有可用坐席response={"type":"agent_queue","content":"当前所有客服坐席繁忙，您已被加入等待队列。请稍候，我们会尽快为您服务。","queue_position":awaitself._get_queue_position(context),"estimated_wait_time":120,# 预估等待时间（秒）"handler":"agent_queue"}returnresponseexceptExceptionase:self.logger.error(f"人工处理失败:{e}")return{"type":"error","content":"抱歉，人工客服暂时不可用，请稍后再试或留言。","handler":"agent_fallback"}asyncdef_handle_hybrid(self,context:ConversationContext,nlp_analysis:NLPAnalysis,message:Dict)->Dict[str,Any]:"""混合处理：AI提供建议，人工确认"""try:# 先获取AI响应text=message.get("content","")retrieval_result=awaitretrieval_engine.search(text)prompt=self._build_ai_prompt(context,text,nlp_analysis,retrieval_result)ai_response=awaitself._generate_ai_response(prompt)# 同时寻找人工坐席available_agents=awaitself._find_available_agents(context,nlp_analysis)response={"type":"hybrid_response","content":ai_response,"ai_suggestion":ai_response,"requires_agent_confirmation":True,"agent_available":len(available_agents)>0,"confidence":nlp_analysis.intent_result.confidence,"sources":[r.content[:100]forrinretrieval_result.results[:2]],"handler":"hybrid"}# 如果坐席可用，准备转接ifavailable_agents:response["agent_note"]="AI已提供建议，坐席可随时接管对话"returnresponseexceptExceptionase:self.logger.error(f"混合处理失败:{e}")returnawaitself._handle_with_ai(context,nlp_analysis,message)def_build_ai_prompt(self,context:ConversationContext,text:str,nlp_analysis:NLPAnalysis,retrieval_result:RetrievalResponse)->str:"""构建AI提示"""# 获取对话历史recent_messages=context.get_recent_messages(5)# 构建提示模板prompt=f"""你是一个专业的客服AI助手，正在与客户对话。 客户信息： - 对话ID:{context.conversation_id}- 消息总数:{context.message_count}- 当前情感:{nlp_analysis.sentiment_result.sentiment}(分数:{nlp_analysis.sentiment_result.score}) - 检测意图:{nlp_analysis.intent_result.intent}(置信度:{nlp_analysis.intent_result.confidence}) 对话历史（最近5条）： """formsginrecent_messages[-5:]:sender="客户"ifmsg.get("sender_type")=="customer"else"客服"prompt+=f"{sender}:{msg.get('content','')}\n"prompt+=f"\n当前客户消息：{text}\n\n"# 添加相关知识ifretrieval_result.results:prompt+="相关知识库信息：\n"fori,resultinenumerate(retrieval_result.results[:3],1):prompt+=f"{i}.{result.content[:200]}...\n"prompt+=""" 请根据以上信息，用友好、专业、有帮助的语气回复客户。注意： 1. 如果问题需要人工处理，请说明原因并建议转人工 2. 如果信息不足，可以询问更多细节 3. 保持回复简洁明了 4. 如果是中文对话，请用中文回复 你的回复："""returnpromptasyncdef_generate_ai_response(self,prompt:str)->str:"""生成AI响应（简化版本）"""# 这里应该调用实际的AI模型# 示例：使用简单的规则生成响应# 实际应该使用GPT或其他模型# 为了示例，返回一个简单的响应if"你好"inpromptor"嗨"inprompt:return"您好！我是AI客服助手，很高兴为您服务。请问有什么可以帮助您的吗？"elif"谢谢"inpromptor"感谢"inprompt:return"不客气！很高兴能帮助您。如果还有其他问题，请随时告诉我。"elif"再见"inpromptor"拜拜"inprompt:return"感谢您的咨询，再见！祝您有美好的一天！"else:return"我理解您的问题，正在为您查找相关信息。请稍等..."asyncdef_find_available_agents(self,context:ConversationContext,nlp_analysis:NLPAnalysis)->List[Dict]:"""查找可用坐席"""# 这里应该从数据库查询可用坐席# 示例：返回模拟数据skills_needed=[]ifnlp_analysis.intent_result.intent=="technical_support":skills_needed=["technical","support"]elifnlp_analysis.intent_result.intent=="payment_issue":skills_needed=["billing","finance"]# 模拟坐席数据mock_agents=[{"id":"agent_001","name":"张客服","skills":["technical","support","product"],"status":"online","current_chats":2,"max_chats":5},{"id":"agent_002","name":"李客服","skills":["billing","finance","account"],"status":"online","current_chats":1,"max_chats":5},{"id":"agent_003","name":"王客服","skills":["general","sales","product"],"status":"away","current_chats":0,"max_chats":5}]# 过滤可用坐席available_agents=[]foragentinmock_agents:if(agent["status"]=="online"andagent["current_chats"]<agent["max_chats"]):# 检查技能匹配ifskills_needed:ifany(skillinagent["skills"]forskillinskills_needed):available_agents.append(agent)else:available_agents.append(agent)returnavailable_agentsasyncdef_assign_agent(self,context:ConversationContext,agent:Dict)->Dict:"""分配坐席"""# 更新上下文context.metadata["assigned_agent_id"]=agent["id"]context.metadata["assigned_agent_name"]=agent["name"]context.transfer_count+=1# 这里应该更新数据库中的对话记录# 并通知坐席端returnagentasyncdef_get_queue_position(self,context:ConversationContext)->int:"""获取队列位置"""# 这里应该查询实际的队列系统return1asyncdef_update_conversation_state(self,context:ConversationContext,response:Dict):"""更新对话状态"""old_state=context.metadata.get("current_state",ConversationState.ACTIVE.value)new_state=old_state response_type=response.get("type","")ifresponse_type=="agent_assigned":new_state=ConversationState.TRANSFERRING.valueelifresponse_type=="agent_queue":new_state=ConversationState.PENDING.valueelif"escalat"inresponse_type.lower():new_state=ConversationState.ESCALATED.valueifnew_state!=old_state:context.metadata["current_state"]=new_state context.add_state_change(old_state,new_state,response_type)self.logger.info(f"对话{context.conversation_id}状态变更:{old_state}->{new_state}")asyncdefget_conversation(self,conversation_id:str)->Optional[ConversationContext]:"""获取对话上下文"""returnself.conversations.get(conversation_id)asyncdefclose_conversation(self,conversation_id:str,reason:str="client_closed"):"""关闭对话"""context=self.conversations.get(conversation_id)ifcontext:old_state=context.metadata.get("current_state",ConversationState.ACTIVE.value)context.metadata["current_state"]=ConversationState.CLOSED.value context.add_state_change(old_state,ConversationState.CLOSED.value,reason)context.metadata["closed_reason"]=reason context.metadata["closed_at"]=datetime.now().isoformat()self.logger.info(f"关闭对话:{conversation_id}, 原因:{reason}")asyncdef_cleanup_task(self):"""清理任务：定期清理超时对话"""whileTrue:try:awaitasyncio.sleep(self.cleanup_interval)current_time=datetime.now()expired_conversations=[]# 查找超时对话forconv_id,contextinself.conversations.items():ifcontext.is_timed_out():expired_conversations.append(conv_id)# 清理超时对话forconv_idinexpired_conversations:awaitself.close_conversation(conv_id,"timeout")delself.conversations[conv_id]ifexpired_conversations:self.logger.info(f"清理了{len(expired_conversations)}个超时对话")# 限制总对话数量iflen(self.conversations)>self.max_conversations:# 移除最旧的对话excess=len(self.conversations)-self.max_conversations old_ids=list(self.conversations.keys())[:excess]forconv_idinold_ids:awaitself.close_conversation(conv_id,"system_cleanup")delself.conversations[conv_id]self.logger.info(f"清理了{excess}个旧对话以控制内存")exceptExceptionase:self.logger.error(f"清理任务出错:{e}")# 创建全局对话管理器实例conversation_manager=ConversationManager()