当前位置: 首页 > news >正文

21 - 大模型智能体进阶指南 (5):电影助手的知识革命:从静态存储到实时进化 —— 检索增强与自主学习的协同机制

目录

1、整体架构:大模型智能体的 “五脏六腑”

2、大模型智能体的 “大脑”:决策与调度机制

2.1 意图识别与任务路由

2.2 闭环学习:从 “未知” 到 “已知”

3、检索增强生成(RAG):让智能体 “说真话”

3.1 知识的 “数字化”:向量嵌入(CustomEmbeddings)

3.2 知识的 “仓库”:Chroma 向量数据库(load_knowledge_base)

3.3 知识的 “检索”:从仓库到答案(query_movie_info)

4、电影助手的核心功能:从查询到推荐

4.1 精准查询

4.2 自主学习

4.3 多维度对比

4.4 个性化推荐

4.5 网络搜索

4.6 数学计算

5、技术亮点:为何这个智能体 “更聪明”

6、技术选型的深层考量

6.1 为何选择 Chroma 而非 Milvus/Faiss?

6.2 Ollama vs 原生 API 的取舍

6.3 正则提取 vs 大模型提取的边界

7、完整代码

8、实验结果


1、整体架构:大模型智能体的 “五脏六腑”

实现了一个具备检索增强能力的大模型智能体(LLM Agent),核心目标是通过 “本地知识库 + 网络搜索 + 自主学习” 的协同,提供精准的电影信息服务。其架构可分为三层,形成 “感知 - 决策 - 执行 - 学习” 的闭环:

层级核心组件功能描述
基础层向量嵌入模型(CustomEmbeddings)、Chroma 向量数据库将电影知识转化为向量并存储,支撑高效检索
功能层检索系统(query_movie_info)、自我学习系统(SelfLearningSystem)、网络搜索(search_with_serpapi实现知识检索、动态学习、外部信息获取
应用层决策模块(decide_processing)、交互接口(main解析用户需求,调度功能模块,生成自然语言响应

2、大模型智能体的 “大脑”:决策与调度机制

智能体的核心能力在于理解用户意图并自主选择工具,这一过程由decide_processing函数实现,体现了 “智能体” 的自主性。

2.1 意图识别与任务路由

智能体通过关键词匹配和规则引擎解析用户输入类型,优先级如下:

  • 电影查询(最高优先级):若输入包含电影名(如 “《肖申克的救赎》的导演”),先检索本地知识库;未命中则触发网络搜索,并自动学习新信息。
    # 核心逻辑:本地查询→网络搜索→自动学习
    movie_name = extract_movie_name(user_input)
    if movie_name:movie_ans = query_movie_info(...)  # 查本地知识库if movie_ans: return movie_ans# 本地无结果,调用网络搜索search_result, search_info = search_with_serpapi(...)learn_messages = learner.auto_learn_from_search(...)  # 学习新信息
    
  • 推荐请求:若输入含 “推荐”“类似” 等关键词(如 “推荐科幻电影”),提取类型、参考电影等条件,调用推荐模块。
  • 复杂推理:若输入涉及多电影对比(如 “《肖申克的救赎》和《泰坦尼克号》哪个早上映”),调用推理模块分析时间 / 类型等维度。
  • 其他任务:数学计算、手动知识学习(如 “《盗梦空间》的导演是诺兰”)等。

2.2 闭环学习:从 “未知” 到 “已知”

智能体通过 “查询 - 缺失 - 搜索 - 学习 - 复用” 的闭环持续进化:

  • 当本地知识库无结果时,自动触发网络搜索;
  • 从搜索结果中提取结构化信息(导演、主演等),通过SelfLearningSystem存入本地知识库;
  • 下次查询同一电影时,直接返回本地结果,无需重复搜索。

3、检索增强生成(RAG):让智能体 “说真话”

检索增强(RAG)是解决大模型 “幻觉” 和 “知识过时” 的核心技术,通过 “知识检索→增强生成” 流程,确保回答基于真实数据。

3.1 知识的 “数字化”:向量嵌入(CustomEmbeddings

为让计算机理解电影知识的语义,代码使用sentence-transformers/all-MiniLM-L6-v2模型将文本转化为 384 维向量:

class CustomEmbeddings(Embeddings):def embed_query(self, text: str) -> List[float]:# 文本编码为向量(取CLS token的输出)inputs = self.tokenizer(text, return_tensors="pt").to(self.device)with torch.no_grad():outputs = self.model(** inputs)embeddings = outputs.last_hidden_state[:, 0, :].cpu().numpy()return (embeddings / np.linalg.norm(embeddings, axis=1, keepdims=True))[0].tolist()
  • 向量特性:语义相似的文本(如 “导演” 和 “执导”)对应的向量距离更近,支撑相似性检索。

3.2 知识的 “仓库”:Chroma 向量数据库(load_knowledge_base

Chroma 负责存储电影知识的向量,支持高效检索。知识库初始化时加载两类数据:

  • 内置知识:如《肖申克的救赎》《泰坦尼克号》的导演、主演等基础信息;
  • 用户学习的知识:从learning_data.json加载的动态学习内容。
def load_knowledge_base(embedding_model):all_movie_info = ["《肖申克的救赎》|导演|弗兰克·德拉邦特",  # 内置知识# ... 其他内置信息]# 加载用户已学习的知识(避免重复)if os.path.exists(LEARNING_PATH):with open(LEARNING_PATH, 'r') as f:learning_data = json.load(f)for movie, info in learning_data.items():for k, v in info.items():entry = f"《{movie}》|{k}|{v}"if entry not in all_movie_info:all_movie_info.append(entry)# 初始化Chroma数据库db = Chroma.from_documents([Document(page_content=info) for info in all_movie_info],embedding_model,persist_directory="./movie_db")return db

3.3 知识的 “检索”:从仓库到答案(query_movie_info

检索流程分为三步,确保精准获取所需知识:

  1. 查询优化:通过大模型将模糊查询(如 “肖申克的救赎导演”)转化为标准化格式(如 “《肖申克的救赎》的导演”),提升检索精度。
  2. 相似性检索:先通过电影名筛选候选知识,再用向量相似性排序,取最相关结果。
    # 筛选该电影的相关知识
    all_docs = db.get()
    movie_docs = [doc for doc in all_docs['documents'] if doc.startswith(normalized_name)]
    
  3. 信息整合:从检索结果中提取导演、主演等属性,生成结构化回答。

4、电影助手的核心功能:从查询到推荐

基于上述架构,代码实现了六大核心功能,覆盖电影相关场景:

4.1 精准查询

支持电影名 + 属性(导演 / 主演 / 类型 / 上映时间)的查询,如 “《泰坦尼克号》的主演”,返回 “莱昂纳多・迪卡普里奥、凯特・温斯莱特”。

4.2 自主学习

  • 手动学习:用户输入陈述句(如 “《盗梦空间》的导演是诺兰”),智能体验证后存入知识库。
  • 自动学习:从网络搜索结果中提取信息(如导演名),自动存入本地,无需人工干预。

4.3 多维度对比

支持两部电影的时间(如 “哪部早上映”)和类型(如 “共同类型”)对比,例如:

# 时间对比逻辑
y1, y2 = get_year(movies[0]), get_year(movies[1])
if y1 and y2:return f"《{movies[0]}》({y1}年)比《{movies[1]}》({y2}年)早{y2-y1}年"

4.4 个性化推荐

根据类型、参考电影等条件推荐影片,如 “推荐类似《肖申克的救赎》的犯罪片”,返回《监狱风云》《绿里奇迹》等。

4.5 网络搜索

当本地无结果时,通过 SERPAPI 调用谷歌搜索,获取实时信息(如最新上映电影)。

4.6 数学计算

支持简单运算(如 “1994+1997”),满足用户在时间对比等场景中的计算需求。

5、技术亮点:为何这个智能体 “更聪明”

  1. 检索增强的鲁棒性:通过 “关键词筛选 + 向量检索” 的混合模式,解决纯向量检索的 “语义漂移” 问题,确保知识召回率。
  2. 自主进化能力:自我学习系统支持动态更新知识,避免传统工具 “知识固定” 的局限。
  3. 轻量高效:使用轻量级嵌入模型(all-MiniLM-L6-v2)和本地部署的大模型(Ollama-mistral),平衡性能与成本。
  4. 工程化细节
    • 知识去重:避免重复存储相同信息;
    • 错误处理:网络搜索失败时返回友好提示;
    • 兼容性:支持 CPU/GPU 自动切换,适配不同硬件环境。

6、技术选型的深层考量

6.1 为何选择 Chroma 而非 Milvus/Faiss?

  • 轻量性:Chroma 的 “零配置” 特性适合中小规模知识库(10 万条以内),无需复杂的集群部署;
  • 与 LangChain 的兼容性:原生支持Document格式,减少数据转换成本;
  • 自动持久化:新版本默认支持数据自动保存,简化工程实现(对应代码中移除persist()调用)。

6.2 Ollama vs 原生 API 的取舍

  • 选择 Ollama 的核心原因是本地部署能力,避免对外部 API 的依赖,适合离线场景;
  • 权衡:牺牲部分性能(相比云端大模型)换取隐私性(用户数据不流出本地)和部署灵活性。

6.3 正则提取 vs 大模型提取的边界

  • 简单规则(如提取电影名、导演关键词)用正则实现,优势是速度快(微秒级)、无模型依赖;
  • 复杂场景(如从长文本中提取多实体关系)则调用大模型,通过提示词工程(Prompt Engineering)提升精度。
  • 平衡点:通过提取复杂度评分动态选择工具,评分 = 规则匹配度 + 文本长度 / 100,超过阈值则调用大模型。

本文为大模型智能体的工程实现提供了典型范式,尤其在垂直领域(如电影、医疗、法律)的知识服务场景中具有很高的参考价值。

7、完整代码

"""
文件名: llm_agent_final.py
作者: 墨尘
日期: 2025/8/5
项目名: llm_finetune
备注: 消除Chroma警告,确保本地查询功能正常
"""
import os
import torch
import numpy as np
import re
import json
import requests
import pandas as pd
from typing import List, Optional, Dict, Tuple
from transformers import AutoModel, AutoTokenizer
from langchain_community.vectorstores import Chroma
from langchain.embeddings.base import Embeddings
from langchain.schema import Document
from langchain.prompts import ChatPromptTemplate
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain_community.llms import Ollama
import warnings# 忽略Chroma弃用警告
warnings.filterwarnings("ignore", category=UserWarning, message="Since Chroma 0.4.x the manual persistence method is no longer supported as docs are automatically persisted.")# 路径配置
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
MODEL_PATH = os.path.join(BASE_DIR, "all-MiniLM-L6-v2")
MEMORY_PATH = os.path.join(BASE_DIR, "conversation_memory.json")
LEARNING_PATH = os.path.join(BASE_DIR, "learning_data.json")
enable_web_search = True# 确保模型目录存在,不存在则从HuggingFace下载
if not os.path.exists(MODEL_PATH):from transformers import AutoModel, AutoTokenizermodel = AutoModel.from_pretrained("sentence-transformers/all-MiniLM-L6-v2")tokenizer = AutoTokenizer.from_pretrained("sentence-transformers/all-MiniLM-L6-v2")model.save_pretrained(MODEL_PATH)tokenizer.save_pretrained(MODEL_PATH)# 常见电影名库
COMMON_MOVIES = {"肖申克的救赎", "泰坦尼克号", "盗梦空间", "蝙蝠侠", "黑暗骑士","星际穿越", "黑客帝国", "阿甘正传", "飞屋环游记", "寻梦环游记"
}# 电影类型列表
MOVIE_GENRES = ["剧情", "喜剧", "动作", "爱情", "悬疑", "恐怖", "纪录片","动画", "冒险", "犯罪", "惊悚", "战争", "奇幻", "历史", "传记", "科幻"]# 1. 自定义嵌入模型
class CustomEmbeddings(Embeddings):def __init__(self, model_path: str):self.device = "cuda" if torch.cuda.is_available() else "cpu"self.tokenizer = AutoTokenizer.from_pretrained(model_path)self.model = AutoModel.from_pretrained(model_path).to(self.device)self.model.eval()def embed_query(self, text: str) -> List[float]:inputs = self.tokenizer(text, padding=True, truncation=True, return_tensors="pt").to(self.device)with torch.no_grad():outputs = self.model(** inputs)embeddings = outputs.last_hidden_state[:, 0, :].cpu().numpy()return (embeddings / np.linalg.norm(embeddings, axis=1, keepdims=True))[0].tolist()def embed_documents(self, texts: List[str]) -> List[List[float]]:return [self.embed_query(text) for text in texts]# 2. 数学计算模块
def is_math_expression(input_str: str) -> bool:math_keywords = {"加", "减", "乘", "除", "+", "-", "*", "/", "×", "÷", "=", "比"}return any(keyword in input_str for keyword in math_keywords)def calculate_math_expression(expr: str) -> str:try:clean_expr = re.sub(r'[^\d+\-*/().^%]', '', expr).replace("^", "**").replace("×", "*").replace("÷", "/")result = eval(clean_expr, {"__builtins__": None}, {})return f"计算结果:{result}"except Exception as e:return f"计算错误:{str(e)}"# 3. 网络搜索模块
def search_with_serpapi(query: str, api_key: str, movie_name: Optional[str] = None) -> Tuple[str, Dict[str, str]]:global enable_web_searchif not enable_web_search:return "网络搜索已关闭,输入'search on'开启", {}if not api_key:return "未设置SERPAPI_API_KEY,无法搜索", {}try:if movie_name:query = f"{movie_name} 电影 导演 主演 类型 上映时间 中国大陆"params = {"engine": "google", "q": query, "api_key": api_key, "num": 5}response = requests.get("https://serpapi.com/search", params=params)data = response.json()if "error" in data:return f"搜索错误:{data['error']}", {}results = []for i, item in enumerate(data.get("organic_results", [])[:5]):title = item.get("title", "")snippet = item.get("snippet", "")results.append(f"【结果{i+1}】{title}\n{snippet}")full_result = "\n\n".join(results)extracted_info = {}if movie_name:extracted_info = extract_movie_info(full_result, movie_name)return f"搜索结果:\n{full_result}", extracted_infoexcept Exception as e:return f"搜索失败:{str(e)}", {}# 4. 电影信息提取函数
def extract_movie_name(question: str) -> Optional[str]:"""提取带《》和不带《》的电影名"""# 优先提取《》包裹的电影名match = re.search(r"《(.+?)》", question)if match:return match.group(1).strip()# 从常见电影名库中匹配(处理无《》的情况)for movie in COMMON_MOVIES:if re.search(rf"\b{re.escape(movie)}\b", question, re.IGNORECASE):return movie# 尝试通过"的"字结构提取(如"XXX的导演")pattern = r"(.+?)的(导演|主演|上映时间|类型|什么时候|谁)"match = re.search(pattern, question)if match:candidate = match.group(1).strip()if len(candidate) >= 2:return candidatereturn Nonedef extract_movie_info(search_result: str, movie_name: str) -> Dict[str, str]:"""精准提取导演、主演、类型、上映时间,清理冗余信息"""info = {}search_lower = search_result.lower()movie_lower = movie_name.lower()# 提取导演director_patterns = [rf"{re.escape(movie_name)}.*?(?:导演|执导)[::]\s*([^,,。;;\n]+)",rf"(?:导演|执导)[::]\s*([^,,。;;\n]+).*?{re.escape(movie_name)}",rf"{re.escape(movie_name)}.*?由([^,,。;;\n]+?)[编剧, ]*导演"]for pattern in director_patterns:match = re.search(pattern, search_result, re.IGNORECASE)if match:# 清理导演名字中的冗余内容director_name = match.group(1).strip()director_name = re.sub(r"[编剧,,、].*", "", director_name)  # 移除"编剧"等多余内容director_name = re.sub(r"执导", "", director_name).strip()info["导演"] = director_namebreak# 提取主演actor_patterns = [rf"{re.escape(movie_name)}.*?(?:主演|领衔主演)[::]\s*([^,,。;;\n]+)",rf"(?:主演|领衔主演)[::]\s*([^,,。;;\n]+).*?{re.escape(movie_name)}"]for pattern in actor_patterns:match = re.search(pattern, search_result, re.IGNORECASE)if match:info["主演"] = match.group(1).strip()break# 提取上映时间year_match = re.search(rf"{re.escape(movie_name)}.*?(\d{{4}})年", search_result)if not year_match:year_match = re.search(rf"(\d{{4}})年.*?{re.escape(movie_name)}", search_result)if year_match:info["上映时间"] = year_match.group(1) + "年"# 提取类型genre_text = ""for genre in MOVIE_GENRES:if re.search(rf"{genre}", search_result, re.IGNORECASE):genre_text += genre + "、"if genre_text:info["类型"] = genre_text.rstrip("、")return infodef has_multiple_movies(question: str) -> bool:return len(re.findall(r"《(.+?)》", question)) >= 2 or \sum(1 for movie in COMMON_MOVIES if movie in question) >= 2def is_question(input_str: str) -> bool:question_marks = {"?", "?"}question_words = {"谁", "什么", "何时", "哪里", "为什么", "怎么", "吗", "呢"}if re.search(r"是\s*[^\??]+$", input_str):return Falsereturn any(mark in input_str for mark in question_marks) or any(word in input_str for word in question_words)def is_valid_assertion(input_str: str) -> bool:pattern = r"(《.+?》|[\u4e00-\u9fa5a-zA-Z0-9 ]+)的(导演|主演|上映时间|类型)是(.+?)([。.,;;]?)$"match = re.search(pattern, input_str)if not match:return Falsevalue = match.group(3).strip()return not any(word in value for word in {"谁", "什么", "吗", "?"})def extract_assertion_info(input_str: str) -> Optional[Tuple[str, str, str]]:pattern = r"(《(.+?)》|([\u4e00-\u9fa5a-zA-Z0-9 ]+))的(导演|主演|上映时间|类型)是(.+?)([。.,;;]?)$"match = re.search(pattern, input_str)if match:movie = match.group(2) if match.group(2) else match.group(3).strip()attr = match.group(4).strip()value = match.group(5).strip()return movie, attr, valuereturn None# 5. 检索增强相关函数
def optimize_movie_query(raw_query: str, llm) -> str:if not raw_query.strip():return raw_queryprompt_template = """请将以下查询优化为更适合检索电影信息的形式,规则:1. 若包含电影名,用《》包裹(如"肖申克的救赎"→"《肖申克的救赎》");2. 明确查询维度(导演/主演/上映时间/类型);3. 保持中文,不添加额外解释。原始查询:{raw_query}"""prompt = ChatPromptTemplate.from_template(prompt_template)runner = prompt | llmresult = runner.invoke({"raw_query": raw_query})return result.content if hasattr(result, 'content') else raw_querydef init_compression_retriever(db: Chroma, llm) -> ContextualCompressionRetriever:base_retriever = db.as_retriever(search_type="similarity",search_kwargs={"k": 3})compressor = LLMChainExtractor.from_llm(llm)return ContextualCompressionRetriever(base_compressor=compressor,base_retriever=base_retriever)def query_movie_info(db: Chroma, compression_retriever, raw_query: str, llm, full_info: bool = False) -> Optional[str]:"""改进检索逻辑,确保能找到已学习的信息"""optimized_query = optimize_movie_query(raw_query, llm)print(f"优化后查询:{optimized_query}")movie_name = extract_movie_name(optimized_query)if not movie_name:return Nonenormalized_name = f"《{movie_name}》"# 直接检查知识库中是否存在该电影的文档all_docs = db.get()movie_docs = [doc for doc in all_docs['documents'] if doc.startswith(normalized_name)]if not movie_docs:# 尝试通过检索器查找compressed_docs = compression_retriever.invoke(optimized_query)if not compressed_docs:return Nonemovie_docs = [doc.page_content for doc in compressed_docs if doc.page_content.startswith(normalized_name)]# 提取信息info = {}for doc in movie_docs:parts = doc.split('|')if len(parts) >= 3 and parts[0] == normalized_name:attr, value = parts[1], parts[2]info[attr] = value# 调试信息:显示找到的信息print(f"从知识库中找到的信息:{info}")if full_info or not any(key in optimized_query for key in ["导演", "主演", "上映时间", "类型"]):missing = [k for k in ["导演", "主演", "上映时间", "类型"] if k not in info]missing_msg = f"\n提示:未找到{', '.join(missing)}信息" if missing else ""return (f"{normalized_name}的详细信息:\n"f"- 导演:{info.get('导演','未知')}\n"f"- 主演:{info.get('主演','未知')}\n"f"- 上映时间:{info.get('上映时间','未知')}\n"f"- 类型:{info.get('类型','未知')}" + missing_msg)for attr in ["导演", "主演", "上映时间", "类型"]:if attr in optimized_query and attr in info:return f"{normalized_name}的{attr}是{info[attr]}"return None# 6. 知识库加载函数
def load_knowledge_base(embedding_model: CustomEmbeddings) -> Chroma:"""确保知识库正确加载已学习的数据"""all_movie_info = ["《肖申克的救赎》|导演|弗兰克·德拉邦特","《肖申克的救赎》|主演|蒂姆·罗宾斯、摩根·弗里曼","《肖申克的救赎》|上映时间|1994年","《肖申克的救赎》|类型|剧情、犯罪","《泰坦尼克号》|导演|詹姆斯·卡梅隆","《泰坦尼克号》|主演|莱昂纳多·迪卡普里奥、凯特·温斯莱特","《泰坦尼克号》|上映时间|1997年","《泰坦尼克号》|类型|爱情、灾难"]# 加载学习数据learned_count = 0if os.path.exists(LEARNING_PATH):try:with open(LEARNING_PATH, 'r', encoding='utf-8') as f:learning_data = json.load(f)for movie, info in learning_data.items():for k, v in info.items():clean_v = re.sub(r'[\n\r]', '', v)entry = f"《{movie}》|{k}|{clean_v}"if entry not in all_movie_info:  # 避免重复添加all_movie_info.append(entry)learned_count += 1print(f"已从{LEARNING_PATH}加载{learned_count}条学习数据")except Exception as e:print(f"加载学习数据失败:{e}")documents = [Document(page_content=info) for info in all_movie_info]# 确保正确初始化或加载知识库persist_dir = os.path.join(BASE_DIR, "movie_db")if not os.path.exists(persist_dir):os.makedirs(persist_dir)try:# 尝试加载现有知识库db = Chroma(persist_directory=persist_dir,embedding_function=embedding_model)# 添加新文档(如果有)db.add_documents(documents)print(f"知识库加载成功,包含{len(all_movie_info)}条数据")return dbexcept Exception as e:print(f"加载现有知识库失败,创建新知识库:{e}")# 创建新的知识库db = Chroma.from_documents(documents,embedding_model,persist_directory=persist_dir)return db# 7. 对话记忆系统
class ConversationMemory:def __init__(self, path):self.path = pathself.memory = self._load()def _load(self):if os.path.exists(self.path):try:with open(self.path, 'r', encoding='utf-8') as f:return json.load(f)except:return []return []def save(self):with open(self.path, 'w', encoding='utf-8') as f:json.dump(self.memory, f, ensure_ascii=False, indent=2)def add(self, user, resp):self.memory.append({"user_input": user,"response": resp,"time": str(pd.Timestamp.now())})self.save()# 8. 自我学习系统
class SelfLearningSystem:def __init__(self, path, db):self.path = pathself.data = self._load()self.db = dbself.db_persist_dir = os.path.join(BASE_DIR, "movie_db")def _load(self):if os.path.exists(self.path):try:with open(self.path, 'r', encoding='utf-8') as f:return json.load(f)except:return {}return {}def save(self):with open(self.path, 'w', encoding='utf-8') as f:json.dump(self.data, f, ensure_ascii=False, indent=2)def learn(self, movie, attr, value, confirm=True) -> str:"""确保学习的信息被正确保存到知识库"""clean_value = re.sub(r'[\n\r]', '', value).strip()if not clean_value or len(clean_value) < 2:return f"无效的{attr}信息:{value}"if movie not in self.data:self.data[movie] = {}if self.data[movie].get(attr) == clean_value:return f"已掌握《{movie}》的{attr}是{clean_value}"# 确认学习if confirm:user_confirm = input(f"是否学习《{movie}》的{attr}是{clean_value}?(y/n):").strip().lower()if user_confirm not in ["y", "yes", ""]:return "已取消学习"# 更新数据self.data[movie][attr] = clean_valueself.save()# 更新知识库doc_content = f"《{movie}》|{attr}|{clean_value}"# 先删除旧版本(如果存在)all_docs = self.db.get()for i, doc in enumerate(all_docs['documents']):if doc.startswith(f"《{movie}》|{attr}|"):self.db.delete(ids=[all_docs['ids'][i]])break# 添加新文档self.db.add_documents([Document(page_content=doc_content)])return f"已学习:《{movie}》的{attr}是{clean_value}"def auto_learn_from_search(self, movie_name: str, search_info: Dict[str, str]) -> List[str]:messages = []if not search_info:return ["未从搜索结果中提取到可学习的信息,建议手动补充:《电影名》的X是Y"]for attr, value in search_info.items():if value and len(value) > 1:msg = self.learn(movie_name, attr, value, confirm=False)messages.append(msg)missing = [k for k in ["导演", "主演", "上映时间", "类型"] if k not in search_info]if missing:messages.append(f"提示:未提取到{', '.join(missing)}信息,可手动输入补充")return messages# 9. 复杂推理模块
def complex_reasoning(user_input: str, db: Chroma, compression_retriever, llm) -> Optional[str]:if has_multiple_movies(user_input) and ("早" in user_input or "晚" in user_input or "差" in user_input):movies = []for m in re.findall(r"《(.+?)》", user_input):movies.append(m)if len(movies) < 2:for movie in COMMON_MOVIES:if movie in user_input and movie not in movies:movies.append(movie)movies = list(set(movies))[:2]if len(movies) < 2:return "请明确两部电影名称"def get_year(movie):res = query_movie_info(db, compression_retriever, f"{movie}的上映时间", llm)if res:year = re.search(r"\d{4}", res)return int(year.group()) if year else Nonereturn Noney1, y2 = get_year(movies[0]), get_year(movies[1])if y1 and y2:diff = abs(y1 - y2)if "差" in user_input:return f"《{movies[0]}》({y1}年)与《{movies[1]}》({y2}年)相差{diff}年"return (f"《{movies[0]}》({y1}年)比《{movies[1]}》({y2}年)更{'' if y1<y2 else '不'}早,"f"相差{diff}年")return "未找到足够的上映时间信息进行对比"if has_multiple_movies(user_input) and "类型" in user_input:movies = []for m in re.findall(r"《(.+?)》", user_input):movies.append(m)if len(movies) < 2:for movie in COMMON_MOVIES:if movie in user_input and movie not in movies:movies.append(movie)movies = list(set(movies))[:2]if len(movies) < 2:return "请明确两部电影名称"def get_genre(movie):res = query_movie_info(db, compression_retriever, f"{movie}的类型", llm)if res:genre_match = re.search(r"类型是(.+)", res)return genre_match.group(1).strip() if genre_match else Nonereturn Noneg1, g2 = get_genre(movies[0]), get_genre(movies[1])if g1 and g2:common = set(g1.split('、')) & set(g2.split('、'))if common:return (f"《{movies[0]}》的类型是{g1},《{movies[1]}》的类型是{g2},"f"它们的共同类型是:{','.join(common)}")else:return (f"《{movies[0]}》的类型是{g1},《{movies[1]}》的类型是{g2},"f"它们没有共同的电影类型")return "未找到足够的电影类型信息进行对比"return None# 10. 电影推荐模块
def is_recommendation_request(input_str: str) -> bool:recommend_keywords = {"推荐", "推荐一部", "推荐一些", "类似", "像", "好看的", "喜欢"}return any(keyword in input_str for keyword in recommend_keywords)def extract_recommendation_criteria(input_str: str) -> Dict[str, str]:criteria = {}for genre in MOVIE_GENRES:if genre in input_str:criteria["类型"] = genrebreakref_movie = extract_movie_name(input_str)if ref_movie:criteria["参考电影"] = ref_movieif "导演" in input_str:director_match = re.search(r"导演[::]\s*([^,,。;;、]+)", input_str)if director_match:criteria["导演"] = director_match.group(1).strip()if "演员" in input_str or "主演" in input_str:actor_match = re.search(r"(演员|主演)[::]\s*([^,,。;;、]+)", input_str)if actor_match:criteria["演员"] = actor_match.group(2).strip()return criteriadef recommend_movies(criteria: Dict[str, str], api_key: str) -> Tuple[str, List[str]]:if "参考电影" in criteria:query = f"类似《{criteria['参考电影']}》的电影 推荐"elif "类型" in criteria:query = f"{criteria['类型']}类型高分电影 推荐"elif "导演" in criteria:query = f"{criteria['导演']} 导演的电影 推荐"elif "演员" in criteria:query = f"{criteria['演员']} 主演的电影 推荐"else:query = "高分电影推荐 豆瓣top250"search_result, _ = search_with_serpapi(query, api_key)recommended_movies = re.findall(r"《(.+?)》", search_result)[:5]return search_result, recommended_moviesdef format_recommendation_response(criteria: Dict[str, str], search_result: str, movies: List[str]) -> str:if not movies:return f"根据您的需求为您推荐:\n{search_result}"response = "为您推荐以下电影:\n"for i, movie in enumerate(movies, 1):response += f"{i}. 《{movie}》\n"response += "\n详细信息:\n" + search_resultreturn response# 11. 决策模块
def decide_processing(user_input: str, db: Chroma, compression_retriever, llm, api_key: str,memory: ConversationMemory, learner: SelfLearningSystem) -> str:movie_name = extract_movie_name(user_input)if movie_name:full_info = any([movie_name in user_input and "的" not in user_input,f"《{movie_name}》" in user_input and "的" not in user_input])# 先尝试本地查询movie_ans = query_movie_info(db, compression_retriever, user_input, llm, full_info=full_info)if movie_ans:return movie_ans# 本地查询失败,进行网络搜索search_query = f"{movie_name} 电影 导演 主演 类型 上映时间"search_result, search_info = search_with_serpapi(search_query, api_key, movie_name)learn_messages = learner.auto_learn_from_search(movie_name, search_info)return (f"本地未找到《{movie_name}》的信息,已为您搜索并尝试自动学习:\n"f"{search_result}\n\n"f"学习结果:\n" + "\n".join(learn_messages) + "\n\n"f"您可以再次输入《{movie_name}》获取更新后信息,或手动补充信息")if is_recommendation_request(user_input):criteria = extract_recommendation_criteria(user_input)search_result, movies = recommend_movies(criteria, api_key)return format_recommendation_response(criteria, search_result, movies)reasoning = complex_reasoning(user_input, db, compression_retriever, llm)if reasoning:return reasoningif is_math_expression(user_input):return calculate_math_expression(user_input)if is_valid_assertion(user_input):info = extract_assertion_info(user_input)if info:movie, attr, value = inforeturn learner.learn(movie, attr, value, confirm=True)if is_question(user_input):search_result, _ = search_with_serpapi(user_input, api_key)return f"查询结果:\n{search_result}"return "我不太明白您的意思。您可以尝试:\n" \"- 输入电影名查询信息(如'肖申克的救赎'或《肖申克的救赎》)\n" \"- 输入电影名+维度(如'泰坦尼克号的主演')\n" \"- 其他支持的功能(计算、推荐等)"# 主程序
def main():print("=== === === === === === === === === === === === === === === === === 电影助手 === === === === === === === === === === === === === === === === ======")print("功能说明:")print("1. 电影查询:输入电影名(如'肖申克的救赎'或《肖申克的救赎》)获取信息")print("2. 知识学习:输入陈述句添加知识(如'盗梦空间的导演是诺兰')")print("3. 电影对比:支持上映时间对比和类型对比")print("4. 数学计算:直接输入数学表达式")print("5. 网络控制:输入'search on'/'search off'控制网络搜索")print("6. 电影推荐:输入推荐请求(如'推荐科幻电影')")print("7. 退出程序:输入'exit'或'退出'")print("=== === === === === === === === === === === === === === === === === 电影助手 === === === === === === === === === === === === === === === === ======")# 初始化嵌入模型和知识库embedding_model = CustomEmbeddings(MODEL_PATH)db = load_knowledge_base(embedding_model)# 初始化LLM和压缩检索器print("加载大语言模型...")llm = Ollama(model="mistral")  # 需要先通过Ollama部署mistral模型compression_retriever = init_compression_retriever(db, llm)DEFAULT_SERPAPI_KEY = ""print("用户请自行注册获取SERPAPI_API_KEY")api_key = input("\n输入SERPAPI_API_KEY(回车使用默认值):").strip()if not api_key:api_key = DEFAULT_SERPAPI_KEYprint(f"已使用默认SERPAPI_API_KEY:{api_key[:4]}****(部分隐藏)")memory = ConversationMemory(MEMORY_PATH)learner = SelfLearningSystem(LEARNING_PATH, db)print(f"已加载知识库,包含{len(learner.data)}条学习数据")while True:user_input = input("\n请输入:").strip()if user_input.lower() in ["exit", "退出"]:print("再见!")breakif user_input.lower() in ["search on", "search off"]:global enable_web_searchenable_web_search = (user_input.lower() == "search on")print(f"网络搜索已{'' if enable_web_search else '关闭'}开启")continueresponse = decide_processing(user_input, db, compression_retriever, llm, api_key, memory, learner)print(response)memory.add(user_input, response)if __name__ == "__main__":main()

8、实验结果

http://www.dtcms.com/a/315844.html

相关文章:

  • 防水防尘防摔性能很好的智能三防手机,还有22000mAh大电池
  • 云平台监控-Zabbix企业级高级应用
  • Cisco 3750X交换机更新到IOS 15.2后无法启动 提示:Boot process failed...
  • 水库雨水情测报和大坝安全监测系统解决方案
  • 极验 G-star 人才特训营:为业务安全领域培养下一代新兴力量
  • 单链表应用实践
  • Python包安全工程实践:构建安全可靠的Python生态系统
  • AI - RAG知识库-进阶(一)
  • Effective C++ 条款24:若所有参数皆需类型转换,请为此采用 non-member 函数
  • 数据结构基础:链表(2)——双向链表、循环链表、内核链表
  • TensorFlow深度学习实战(28)——扩散模型(Diffusion Model)
  • 算法训练之哈希表
  • Yarn Application 日志总结
  • 美化一下达梦grant授权说明
  • 蓝桥杯----DS1302实时时钟
  • 私有云盘新体验:FileRise在cpolar的加持下如何让数据管理更自由?
  • 对话访谈|盘古信息×易景科技:宜宾OEM+ODM 标杆,如何规划数字化转型?
  • MySQL Redo Log浅析
  • 无刷电机控制 - STM32F405+CubeMX+HAL库+SimpleFOC06,速度闭环控制(没电流环)
  • 人工智能领域、图欧科技、IMYAI智能助手2025年7月更新月报
  • SOLIDWORKS 买断许可和订阅许可的资金流影响分析-代理商硕迪科技
  • 江协科技STM32学习笔记1
  • Augmodo AI:零售门店智能货架管理平台
  • 复制网页文字到Word、WPS文字?选中后直接拖放
  • MousePlus鼠标右键增强工具v5.5.25,支持鼠标轮盘功能
  • mac前端环境安装
  • HTTP 与 HTTPS 的区别深度解析:从原理到实践
  • 实战教程 node js 实现上传xls文件批量导入到数据库 解析导入
  • 微服务—Gateway
  • 分发饼干(贪心算法)