向量数据库实战-利用PostgreSQL+pgvector构建AI 智能应用：AI-Agent集成-智能对话系统

第四章:AI Agent集成 - 从FAQ到智能对话本章目标: 构建基于RAG架构的智能客服Agent学习时间: 90分钟难度等级: ⭐⭐⭐⭐ 高级本章导读上一章我们实现了FAQ搜索,但用户真正想要的是"聊天式"的智能回答,而不是冷冰冰的FAQ列表。这一章,我们把pgvector和大语言模型结合,构建真正的AI Agent!你将学到:✅ 什么是RAG(检索增强生成)架构✅ 如何把向量搜索和GPT结合✅ 构建智能客服Agent(完整代码)✅ 文档问答系统(支持PDF上传)✅ Prompt工程最佳实践4.1 什么是RAG?(检索增强生成)传统AI的痛点直接用ChatGPT有三个严重问题:知识过时: 只知道训练数据,不了解你的业务幻觉问题: 会胡编乱造不存在的信息无法引用: 不知道答案来源,无法验证举例:用户: 你们的收费标准是什么? ChatGPT: (幻觉)我们提供免费版和专业版,专业版每月$19.99... ❌ 完全是编的!RAG解决方案RAG = Retrieval-Augmented Generation工作流程:Retrieval(检索): 从向量库找相关知识Augmented(增强): 把知识喂给AIGeneration(生成): AI基于知识回答同样的问题,用RAG:1. 从向量库检索到: "按月订阅制,基础版99元/月,专业版599元/月..." 2. 构建Prompt: "基于以下知识回答: [检索的FAQ] 用户问题: 你们的收费标准?" 3. GPT生成: "我们提供两种套餐:基础版99元/月,专业版599元/月..." ✅ 准确且有依据!RAG架构图用户提问向量化查询pgvector检索TOP-K文档构建Prompt知识+问题LLM生成回答返回答案+引用4.2 案例1:智能客服Agent项目结构code/04_customer_service_agent/ ├── agent.py # 智能客服主程序(完整代码) ├── requirements.txt # 依赖 ├── .env.example # 环境配置模板 └── README.md # 使用说明完整代码实现requirements.txt:psycopg2-binary==2.9.9 openai==1.12.0 python-dotenv==1.0.1agent.py(完整代码,280行):""" 智能客服Agent 功能: 基于RAG架构的智能问答系统 架构: pgvector检索 + GPT生成 """importpsycopg2fromopenaiimportOpenAIimportosfromdotenvimportload_dotenvfromtypingimportList,Dict,Optionalimportjson# 加载环境变量load_dotenv()# 数据库配置DB_CONFIG={'host':os.getenv('DB_HOST','localhost'),'port':os.getenv('DB_PORT','5432'),'database':os.getenv('DB_NAME','vectordb'),'user':os.getenv('DB_USER','postgres'),'password':os.getenv('DB_PASSWORD','mysecret123')}classCustomerServiceAgent:"""智能客服Agent类"""def__init__(self,api_key:Optional[str]=None):""" 初始化Agent Args: api_key: OpenAI API Key(可选,默认从环境变量读取) """# 初始化OpenAI客户端self.llm_client=OpenAI(api_key=api_keyoros.getenv('OPENAI_API_KEY'))# 数据库连接self.db_conn=Noneself.connect_database()print("✅ 智能客服Agent初始化成功!")defconnect_database(self):"""连接数据库"""try:self.db_conn=psycopg2.connect(**DB_CONFIG)print("✓ 数据库连接成功")exceptExceptionase:print(f"❌ 数据库连接失败:{e}")raisedef_get_embedding(self,text:str)-List[float]:""" 获取文本向量 Args: text: 要向量化的文本 Returns: 1536维向量 """response=self.llm_client.embeddings.create(model="text-embedding-3-small",input=text)returnresponse.data[0].embeddingdefretrieve_knowledge(self,query:str,top_k:int=3,category_filter:Optional[str]=None)-List[Dict]:""" 从知识库检索相关FAQ Args: query: 用户查询 top_k: 返回结果数量 category_filter: 分类过滤(可选) Returns: 相关FAQ列表 """cur=self.db_conn.cursor()try:# 1. 向量化查询query_embedding=self._get_embedding(query)# 2. 构建SQLifcategory_filter:sql=""" SELECT id, category, question, answer, embedding = %s::vector AS distance FROM faq_knowledge WHERE category = %s ORDER BY distance LIMIT %s """params=(query_embedding,category_filter,top_k)else:sql=""" SELECT id, category, question, answer, embedding = %s::vector AS distance FROM faq_knowledge ORDER BY distance LIMIT %s """params=(query_embedding,top_k)# 3. 执行查询cur.execute(sql,params)results=cur.fetchall()# 4. 格式化结果knowledge_list=[]forrowinresults:/