1 设计思想与架构

1.1 设计思路

一个基于LlamaIndex框架、采用Agentic RAG架构的简单个性化推荐系统实现，适用于CPU和Windows环境。程序将整合中文语义检索增强，并使用本地部署的模型。

1.2 AI-DS借力架构

咨询提示词：给出CPU和Windows下采用LlamaIndex框架、基于Agentic RAG架构的简单个性化推荐的编码，整个程序用一个文件实现，注意各个依赖库的版本对应和中文的检索加强，不使用CONDA，LLM模型采用Ollama部署本地的DeepSeek-R1-1.5B，需要的嵌入模型己通过modelscope下载在本地D:\01ifmts\models\nlp_gte_sentence-embedding_chinese-base，详细说明各个实现过程和测调试运行。

2 开发及其模型准备

2.1 编程环境Python/Thonny

2.2 编辑环境Notepad++

2.3 Ollama/DeepSeek

2.4 准备NLP

2.5 依赖下载

3 功能编码实现

3.1 编码实现

# ==============================================================================
# Agentic RAG 个性化推荐系统
# 基于LlamaIndex框架 + Ollama(DeepSeek-R1-1.5B) + 本地GTE中文嵌入模型
# 单文件实现 | CPU & Windows | 中文检索增强
# ==============================================================================
"""
==============================================================================
[1] 环境准备 — 依赖安装 (pip, 不使用CONDA)
==============================================================================
在 Windows CMD 或 PowerShell 中依次执行：
    # 1. 创建并激活虚拟环境（推荐，非必须）
    python -m venv rag_env
    rag_env\Scripts\activate
    # 2. 核心依赖安装（注意版本对应）
    pip install llama-index-core==0.12.28
    pip install llama-index-llms-ollama==0.6.0
    pip install llama-index-embeddings-huggingface==0.5.1
    pip install llama-index-readers-file==0.4.3
    pip install sentence-transformers==3.4.1
    pip install torch==2.5.1+cpu -f https://download.pytorch.org/whl/torch_stable.html
    pip install pydantic==2.10.4
    pip install numpy==1.26.4
    pip install huggingface-hub==0.27.1
    pip install nest-asyncio==1.6.0
    pip install requests==2.32.3
    # 3. 确保 Ollama 已安装并运行 DeepSeek-R1-1.5B
    #    安装 Ollama: https://ollama.com/download
    #    拉取模型:    ollama pull deepseek-r1:1.5b
    #    启动服务:    ollama serve    (默认 http://localhost:11434)
    # 4. 确认嵌入模型路径
    #    D:\\01ifmts\\models\\nlp_gte_sentence-embedding_chinese-base
    #    该目录下应包含: config.json, model.safetensors, tokenizer.json 等文件
==============================================================================
[2] 版本兼容性说明
==============================================================================
| 包名                                | 版本       | 说明                    |
|------------------------------------|------------|--------------------------|
| llama-index-core                   | 0.12.28     | LlamaIndex 核心包        |
| llama-index-llms-ollama            | 0.6.0       | Ollama LLM 集成         |
| llama-index-embeddings-huggingface | 0.5.1       | HuggingFace 嵌入集成     |
| llama-index-readers-file           | 0.4.3       | 文件读取器               |
| sentence-transformers              | 3.4.1       | SentenceTransformer 底层   |
| torch                              | 2.5.1+cpu   | CPU版PyTorch           |
| pydantic                           | 2.10.4      | 数据模型框架             |
| numpy                              | 1.26.4     | 数值计算                 |
| huggingface-hub                    | 0.27.1     | 模型下载/缓存             |
| nest-asyncio                       | 1.6.0      | 异步事件循环兼容(Jupyter等) |
| requests                           | 2.32.3     | HTTP请求(Ollama诊断用)   |
核心兼容链:
  llama-index-core 0.12.x
    ├─ llama-index-llms-ollama 0.5.x~0.6.x
    ├─ llama-index-embeddings-huggingface 0.4.x~0.5.x
    └─ llama-index-readers-file 0.3.x~0.4.x
  sentence-transformers 3.x → torch 2.x
  HuggingFaceEmbedding 底层调用 sentence-transformers.SentenceTransformer
  AgentWorkflow (0.12+) → FunctionAgent + asyncio
==============================================================================
[3] 运行方式
==============================================================================
    python agentic_rag_recommend.py
==============================================================================
"""
# ==============================================================================
# [4] 导入与配置
# ==============================================================================
import os, sys, json, time, asyncio, logging
from pathlib import Path
from typing import List, Dict, Optional
# ---- 日志配置 ----
logging.basicConfig( level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s - %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)] )
logger = logging.getLogger("AgenticRAG")
# ---- 关键路径配置（根据你的实际环境修改） ----
# Ollama 服务地址
OLLAMA_BASE_URL = "http://localhost:11434"
# DeepSeek-R1-1.5B 在 Ollama 中的模型名称
OLLAMA_MODEL_NAME = "deepseek-r1:1.5b"
# 本地GTE中文嵌入模型路径（ModelScope下载的模型目录）
# 注意：Windows路径中的反斜杠需要用原始字符串或双反斜杠
EMBEDDING_MODEL_PATH = r"D:\01ifmts\models\nlp_gte_sentence-embedding_chinese-base"
# ==============================================================================
# [5] LlamaIndex 核心导入
# ==============================================================================
from llama_index.core import ( Settings, VectorStoreIndex, Document, )
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.tools import QueryEngineTool
from llama_index.core.agent.workflow import FunctionAgent, AgentWorkflow
# ==============================================================================
# [6] LLM 配置 — Ollama + DeepSeek-R1-1.5B
# ==============================================================================
from llama_index.llms.ollama import Ollama
def setup_llm():
    """
    配置本地 Ollama LLM。
    DeepSeek-R1-1.5B 是蒸馏推理模型：
      - 参数量: 1.5B
      - 上下文窗口: 约 8192 tokens (实际使用 4096 以保证稳定性)
      - 支持思维链推理 (Chain-of-Thought)
    request_timeout 设为 300s，因为 CPU 推理速度较慢，
    DeepSeek-R1 的思维链可能产生大量 token。
    """
    llm = Ollama( model=OLLAMA_MODEL_NAME, base_url=OLLAMA_BASE_URL,
        temperature=0.7,           # 适中的创造性
        context_window=4096,      # 保守的上下文窗口，1.5B模型稳定运行
        request_timeout=300.0,     # CPU推理较慢，需要较长超时
        is_function_calling_model=True,  # 启用函数调用以支持Agent工具选择
        additional_kwargs={ "num_predict": 1024, }, )  # 限制最大生成token数
    # 全局设置 LLM
    Settings.llm = llm
    logger.info(f"LLM 配置完成: Ollama + {OLLAMA_MODEL_NAME}")
    return llm
# ==============================================================================
# [7] 嵌入模型配置 — 本地 GTE 中文嵌入模型
# ==============================================================================
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
def setup_embedding():
    """
    配置本地 GTE 中文嵌入模型。
    模型来源: 阿里达摩院 ModelScope -> iic/nlp_gte_sentence-embedding_chinese-base
    关键参数说明:
      - model_name: 本地路径，HuggingFaceEmbedding 会自动识别为路径并加载
      - device: "cpu"，因为目标环境无GPU
      - max_length: 128，该模型默认最大文本长度
      - normalize: True，归一化嵌入向量以提高检索精度
      - trust_remote_code: True，本地模型可能包含自定义代码
    中文检索增强策略:
      1. GTE 模型原生支持中文语义编码
      2. 使用中文友好的 SentenceSplitter（按标点分句）
      3. chunk_size 设为 300（中文约150~200字），保证语义完整性
      4. 查询时使用中文查询模板增强检索效果
    """
    # 验证模型路径是否存在
    model_path = Path(EMBEDDING_MODEL_PATH)
    if not model_path.exists():
        logger.error( f"嵌入模型路径不存在: {EMBEDDING_MODEL_PATH}\n"
            f"请确认模型已下载到该目录，或修改 EMBEDDING_MODEL_PATH 变量" )
        sys.exit(1)
    # 检查关键模型文件
    required_files = ["config.json"]
    missing_files = [f for f in required_files if not (model_path / f).exists()]
    if missing_files:
        logger.warning(f"模型目录缺少文件: {missing_files}，尝试继续加载...")
    embed_model = HuggingFaceEmbedding(
        model_name=EMBEDDING_MODEL_PATH,  # 直接指向本地目录
        device="cpu",                      # CPU 模式
        max_length=128,                    # GTE-base 默认最大长度
        normalize=True,                   # 归一化向量
        trust_remote_code=True,           # 允许本地自定义代码
        embed_batch_size=8,               # 小批量避免内存溢出
        show_progress_bar=True,           # 显示嵌入进度条
    )
    # 全局设置嵌入模型
    Settings.embed_model = embed_model
    # 验证嵌入模型
    test_embedding = embed_model.get_text_embedding("测试中文嵌入")
    logger.info( f"嵌入模型配置完成: GTE-base 中文\n"
        f"  向量维度: {len(test_embedding)}\n"
        f"  模型路径: {EMBEDDING_MODEL_PATH}" )
    return embed_model
# ==============================================================================
# [8] 中文增强的文档分割器
# ==============================================================================
def setup_node_parser():
    """
    配置中文友好的文档分割器。
    中文分割策略:
      - chunk_size: 300 字符（约 150~200 中文字），
        小 chunk 更精确匹配，提升检索召回率
      - chunk_overlap: 50 字符，保证跨 chunk 的语义连贯
      - separator: 保留默认空格（不影响中文切分）
      - paragraph_separator: 段落分隔符
      - secondary_chunking_regex: 加入中文标点的正则，精准断句
    LlamaIndex 的 SentenceSplitter 会按以下优先级分割:
      1. 段落分隔符 (\\n\\n)
      2. 句子级分隔符 (。！？；)
      3. 分词级
      4. 字符级
    """
    node_parser = SentenceSplitter( chunk_size=300, # 中文短 chunk，提升检索精度
        chunk_overlap=50,           # 适当重叠避免语义断裂
        separator=" ",              # 基础分隔符（保留默认）
        paragraph_separator="\n\n", # 段落分隔符
        # 中文标点加入正则，精准断句
        secondary_chunking_regex=r'[^,.;；。？！……，]+[,.;；。？！……，]?|[,.;；。？！……，]', )
    Settings.node_parser = node_parser
    logger.info("文档分割器配置完成: 中文增强 (chunk_size=300)")
    return node_parser
# ==============================================================================
# [9] 示例知识库数据 — 个性化推荐场景
# ==============================================================================
def create_sample_documents() -> List[Document]:
    """
    创建示例知识库文档，模拟个性化推荐系统的数据源。
    这里模拟三个不同领域的知识库:
      1. 电影推荐库 — 适合娱乐类查询
      2. 书籍推荐库 — 适合阅读类查询
      3. 美食推荐库 — 适合餐饮类查询
    Agentic RAG 的核心思想:
      Agent 根据用户查询，自主决定检索哪个知识库（工具选择），
      而不是像传统 RAG 把所有数据混在一起检索。
    每个文档的 metadata 包含:
      - category: 领域分类
      - tags: 标签（用于后续个性化过滤）
      - audience: 适合的受众
    """
    documents = []
    # ---- 电影推荐数据 ----
    movies_data = [
        {   "title": "星际穿越",
            "content": """《星际穿越》是克里斯托弗·诺兰执导的科幻电影。影片讲述了一组宇航员穿越虫洞，为人类寻找新家园的故事。电影融合了量子物理、时间膨胀等硬科幻元素，视觉效果震撼，配乐由汉斯·季默创作。适合喜欢科幻、物理和宇宙探索的观众。影片情感线围绕父女关系展开，非常感人。推荐指数：五颗星""",
            "tags": "科幻,冒险,亲情,诺兰", "audience": "科幻爱好者,物理迷",
        },
        {   "title": "千与千寻",
            "content": """《千与千寻》是宫崎骏执导的动画电影，荣获奥斯卡最佳动画长片奖。讲述少女千寻误入神灵世界，在汤屋工作的奇幻经历。影片充满日本神道教文化元素，画面精美细腻，配乐久石让操刀。
主题涉及成长、勇气、环保和身份认同。适合全年龄段观众。推荐指数：五颗星""",
            "tags": "动画,奇幻,成长,宫崎骏", "audience": "动画爱好者,家庭观众",
        },
        {   "title": "肖申克的救赎", "content": """《肖申克的救赎》被誉为影史最伟大电影之一。讲述银行家安迪被冤入狱，在肖申克监狱中用智慧和毅力重获自由的故事。影片主题是希望与自由，台词希望是好事也许是最好的事深入人心。节奏沉稳，演技精湛，结局令人热泪盈眶。推荐指数：五颗星""",
            "tags": "剧情,自由,希望,经典", "audience": "剧情片爱好者",
        },
        {   "title": "流浪地球2",
            "content": """《流浪地球2》是中国科幻电影里程碑之作。讲述了太阳即将毁灭，人类开启流浪地球计划寻找新家园的故事。视觉效果达到好莱坞水准，刘培强、图恒宇双线叙事。影片融合了AI伦理、数字生命、人类命运等深刻主题。适合中国科幻迷和喜欢宏大叙事的观众。推荐指数：四颗半星""",
            "tags": "科幻,中国电影,灾难,AI", "audience": "中国科幻迷",
        },
        {   "title": "盗梦空间",
            "content": """《盗梦空间》是诺兰又一力作，探索梦境与现实的边界。莱昂纳多饰演盗梦者Cobb，在多层梦境中执行植入想法的任务。影片结构精巧，梦层设计令人叹为观止，结尾陀螺是否停转成为永恒悬念。适合喜欢烧脑、逻辑推理和哲学思考的观众。推荐指数：五颗星""",
            "tags": "科幻,悬疑,梦境,诺兰", "audience": "烧脑片爱好者",
        },
    ]
    for m in movies_data:
        documents.append(Document(
            text=m["content"],
            metadata={ "category": "电影", "title": m["title"], "tags": m["tags"], "audience": m["audience"], },
            excluded_llm_metadata_keys=["tags", "audience"],
            metadata_template="{title}\n类别: {category}",
            text_template=" metadata: {metadata_str} \n 内容: {content}", ))
    # ---- 书籍推荐数据 ----
    books_data = [
        {   "title": "三体",
            "content": """《三体》是刘慈欣创作的硬科幻小说，中国科幻文学里程碑。讲述地球文明与三体文明的首次接触和博弈，涵盖宇宙社会学、黑暗森林法则等概念。三部曲包括《三体》《黑暗森林》《死神永生》，每一部格局递进。适合喜欢硬科幻、宇宙探索和哲学思考的读者。获雨果奖最佳长篇小说。
推荐指数：五颗星""",
            "tags": "科幻,中国文学,宇宙,哲学", "audience": "科幻迷,理工科读者",
        },
        {   "title": "活着",
            "content": """《活着》是余华的代表作，讲述了农民福贵坎坷的一生。从地主少爷到贫农，经历了内战、大跃进、文革等历史动荡，亲人一个个离去，但他依然坚强地活着。文字简洁有力，情感冲击极强，是中国当代文学经典。适合喜欢现实主义文学和深度思考的读者。推荐指数：五颗星""",
            "tags": "文学,现实主义,中国,生命", "audience": "文学爱好者",
        },
        {   "title": "小王子",
            "content": """《小王子》是法国作家圣埃克苏佩里的经典童话。讲述小王子从B612星球出发，游历各星球的奇妙旅程。表面是童话，实则是对成人世界的深刻反思——真正重要的东西，用眼睛是看不见的。适合所有年龄段的读者，尤其适合在迷茫时重读。推荐指数：五颗星""",
            "tags": "童话,哲学,法国,成长", "audience": "所有读者",
        },
        {   "title": "人类简史",
            "content": """《人类简史》是以色列学者尤瓦尔·赫拉利的代表作。从认知革命、农业革命到科学革命，重新审视人类7万年发展史。提出了虚构故事推动协作、小麦驯化了人类等颠覆性观点。语言生动，跨学科视角独特。适合对历史、人类学和社会学感兴趣的读者。推荐指数：四颗半星""",
            "tags": "历史,人类学,社科", "audience": "社科爱好者",
        },
    ]
    for b in books_data:
        documents.append(Document( text=b["content"],
            metadata={ "category": "书籍", "title": b["title"], "tags": b["tags"], "audience": b["audience"], },
            excluded_llm_metadata_keys=["tags", "audience"],
            metadata_template="{title}\n类别: {category}",
            text_template=" metadata: {metadata_str} \n 内容: {content}", ))
    # ---- 美食推荐数据 ----
    food_data = [
        {   "title": "四川火锅",
            "content": """四川火锅是中国最具代表性的美食之一。以麻辣鲜香著称，锅底由牛油、花椒、辣椒等数十种香料熬制。毛肚、鹅肠、黄喉是必点涮菜，蘸料以蒜泥香油为主。成都火锅偏麻，重庆火锅偏辣。适合喜欢重口味和热闹氛围的食客。冬天吃火锅是最幸福的体验。推荐指数：五颗星""",
            "tags": "川菜,辣,聚餐,冬天", "audience": "辣味爱好者",
        },
        {   "title": "广式早茶",
            "content": """广式早茶是广东饮食文化的精髓，又称饮茶。虾饺、烧卖、肠粉、叉烧包、凤爪是经典茶点，一盅两件是传统搭配。茶楼氛围悠闲，适合慢慢品味。广州人叹早茶是一种生活态度。适合喜欢精致小食、体验岭南文化的食客。推荐指数：五颗星""",
            "tags": "粤菜,点心,广东,休闲", "audience": "精致美食爱好者",
        },
        {   "title": "兰州牛肉面",
            "content": """兰州牛肉面是西北面食代表，讲究一清二白三红四绿五黄——汤清、萝卜白、辣油红、蒜苗绿、面条黄。手工拉面是灵魂，面条分毛细、二细、韭叶等多种粗细。清晨一碗热面，是兰州人的早餐标配。适合喜欢面食和朴实美味的食客。推荐指数：四颗半星""",
            "tags": "面食,西北,早餐,牛肉", "audience": "面食爱好者",
        },
        {   "title": "日式寿司",
            "content": """日式寿司是日本料理的代表，讲究食材新鲜和匠人技艺。握寿司是最经典形式，醋饭搭配新鲜鱼生，芥末提味。金枪鱼、三文鱼、鲷鱼是常见选择，高级寿司店追求时令鱼材。吃寿司讲究顺序：白身鱼、光物、赤身、卷物。适合喜欢海鲜和日式精致料理的食客。推荐指数：五颗星""",
            "tags": "日料,海鲜,精致,匠人", "audience": "日料爱好者",
        },
    ]
    for f in food_data:
        documents.append(Document( text=f["content"],
            metadata={ "category": "美食", "title": f["title"], "tags": f["tags"], "audience": f["audience"], },
            excluded_llm_metadata_keys=["tags", "audience"],
            metadata_template="{title}\n类别: {category}",
            text_template=" metadata: {metadata_str} \n 内容: {content}", ))
    logger.info(f"知识库文档创建完成: 共 {len(documents)} 篇")
    return documents
# ==============================================================================
# [10] 构建多领域向量索引 — Agentic RAG 的基础
# ==============================================================================
def build_category_indexes(documents: List[Document]) -> Dict[str, VectorStoreIndex]:
    """
    按领域分别构建 VectorStoreIndex。
    Agentic RAG 的核心架构:
      1. 不是把所有文档塞进一个大索引，而是按领域建立独立索引
      2. 每个索引封装为 QueryEngineTool，成为 Agent 可调用的"工具"
      3. Agent 根据用户查询语义，自主选择最合适的工具（索引）
    为什么分开索引优于合并索引:
      - 检索精度: "推荐一部科幻电影" 不会返回科幻书籍的结果
      - 可扩展性: 新增领域只需加索引，不影响已有索引
      - 个性化: 不同领域可以用不同检索策略
    中文检索增强要点:
      - GTE 中文嵌入模型天然支持中文语义相似度计算
      - chunk_size=300 保证中文短文本的精确匹配
      - VectorStoreIndex 使用余弦相似度（归一化向量）
    """
    category_indexes = {}
    # 按类别分组文档
    categories = {}
    for doc in documents:
        cat = doc.metadata.get("category", "未分类")
        if cat not in categories:
            categories[cat] = []
        categories[cat].append(doc)
    # 为每个类别构建独立的向量索引
    for category, cat_docs in categories.items():
        logger.info(f"正在构建 [{category}] 类别向量索引 ({len(cat_docs)} 篇文档)...")
        start_time = time.time()
        index = VectorStoreIndex.from_documents( cat_docs, show_progress=True, )
        build_time = time.time() - start_time
        logger.info(f"  [{category}] 索引构建完成，耗时 {build_time:.1f}s")
        category_indexes[category] = index
    return category_indexes
# ==============================================================================
# [11] 构建 Agentic RAG 查询引擎
# ==============================================================================
def build_agentic_query_engine(category_indexes: Dict[str, VectorStoreIndex]):
    """
    构建 Agentic RAG 的核心: 将多个索引封装为工具，由 Agent 自主路由。
    架构层次:
      用户查询
         │
         ▼
      AgentWorkflow (FunctionAgent)
      (DeepSeek-R1-1.5B 作为推理引擎)
      分析用户意图 → 选择最合适的工具
         │
         ├──────────┼──────────┐
         ▼          ▼          ▼
      电影索引    书籍索引    美食索引
      工具        工具        工具
    关键组件:
      1. QueryEngineTool: 将每个 VectorStoreIndex 的查询引擎包装为工具
      2. FunctionAgent: 使用 LLM 的函数调用能力选择工具
      3. AgentWorkflow: 编排 Agent 的推理和工具调用流程
      4. 工具描述: 中文描述帮助 LLM 理解每个工具的用途
    与传统 RAG 的区别:
      传统 RAG: 用户查询 → 单一索引检索 → 生成回答
      Agentic RAG: 用户查询 → Agent推理 → 选择索引 → 定向检索 → 生成回答
    注意: 新版 LlamaIndex (0.12+) 使用 AgentWorkflow + FunctionAgent
    替代了旧版 FunctionCallingAgent.from_tools() 接口。
    """
    tools = []
    # ---- 为每个领域构建查询引擎工具 ----
    tool_descriptions = { "电影": "电影推荐工具。当用户询问电影、影片、导演、演员、观影推荐等影视娱乐相关问题时使用此工具。可以检索到电影的详细介绍、推荐理由和评分。", "书籍": "书籍推荐工具。当用户询问书籍、小说、阅读推荐、文学、作者等阅读相关问题时使用此工具。可以检索到书籍的详细介绍、推荐理由和评分。", "美食": "美食推荐工具。当用户询问美食、餐厅、菜系、烹饪、餐饮推荐等饮食相关问题时使用此工具。可以检索到美食的详细介绍、口味特点和推荐指数。", }
    for category, index in category_indexes.items():
        # 创建查询引擎
        # similarity_top_k: 返回最相似的 k 个结果
        # 对于1.5B小模型，k=2避免信息过载
        query_engine = index.as_query_engine( similarity_top_k=2, )
        # 将查询引擎封装为工具
        tool = QueryEngineTool.from_defaults( query_engine=query_engine,
            name=f"{category}_recommend_tool",
            description=tool_descriptions.get(category, f"{category}领域推荐工具"), )
        tools.append(tool)
        logger.info(f"  工具已创建: {category}_recommend_tool")
    # ---- 创建 FunctionAgent + AgentWorkflow ----
    # 这是 Agentic RAG 的"大脑"
    # Agent 会:
    #   1. 分析用户查询的语义
    #   2. 决定调用哪个工具（或多个工具）
    #   3. 综合工具返回的结果
    #   4. 生成最终的个性化推荐回答
    # 系统提示词：指导 Agent 的行为
    system_prompt = ( "你是一个个性化推荐助手。根据用户的请求和偏好，"
        "选择最合适的工具检索知识库，然后给出个性化推荐。"
        "请用中文回答，推荐内容要具体，说明推荐理由。" )
    # 创建 FunctionAgent（新版 API）
    agent = FunctionAgent( name="RecommendAgent",
        description="个性化推荐智能体，能根据用户请求从电影、书籍、美食等领域给出推荐",
        tools=tools, system_prompt=system_prompt, llm=Settings.llm, verbose=True, )
    # 包装为 AgentWorkflow（统一入口）
    workflow = AgentWorkflow( agents=[agent], verbose=True, )
    logger.info("Agentic RAG 查询引擎构建完成")
    return workflow
# ==============================================================================
# [12] 个性化推荐函数
# ==============================================================================
def personalized_recommend(workflow, query: str, user_profile: Optional[Dict] = None) -> str:
    """
    执行个性化推荐查询。
    个性化策略:
      1. 将用户画像信息注入查询上下文
      2. Agent 根据查询语义选择最合适的知识领域
      3. 检索结果融入用户偏好进行推理
    user_profile 示例:
      {   "name": "小明",
          "age": 25,
          "interests": ["科幻", "动画"],
          "dietary": "不吃辣",
      }
    注意: DeepSeek-R1-1.5B 作为推理模型，会先进行思维链推理，
    再输出最终答案。
    新版 LlamaIndex 的 AgentWorkflow.run() 是异步方法，
    需要用 asyncio.run() 来同步调用。
    """
    # 构建增强查询
    if user_profile:
        profile_str = json.dumps(user_profile, ensure_ascii=False)
        enhanced_query = ( f"用户画像: {profile_str}\n"
            f"用户请求: {query}\n"
            f"请根据用户画像和请求，从知识库中检索并给出个性化推荐。" )
    else:
        enhanced_query = query
    logger.info(f"增强查询: {enhanced_query[:80]}...")
    start_time = time.time()
    # AgentWorkflow.run() 是异步方法，需要用 asyncio.run() 同步调用
    try:
        loop = asyncio.get_event_loop()
        if loop.is_running():
            # 如果已有事件循环运行中（如 Jupyter），使用 nest_asyncio
            import nest_asyncio
            nest_asyncio.apply()
            response = asyncio.run(workflow.run(user_msg=enhanced_query))
        else:
            response = asyncio.run(workflow.run(user_msg=enhanced_query))
    except RuntimeError:
        # 没有事件循环，直接创建
        response = asyncio.run(workflow.run(user_msg=enhanced_query))
    elapsed = time.time() - start_time
    logger.info(f"查询完成，耗时 {elapsed:.1f}s")
    return str(response)
# ==============================================================================
# [13] 连接测试与诊断
# ==============================================================================
def test_ollama_connection():
    """
    测试 Ollama 服务是否可用，模型是否已加载。
    诊断步骤:
      1. 尝试连接 Ollama API
      2. 检查 deepseek-r1:1.5b 模型是否存在
      3. 发送简单测试请求验证推理能力
    """
    import requests
    logger.info("=" * 60)
    logger.info("Ollama 连接测试")
    logger.info("=" * 60)
    # 测试1: Ollama 服务是否运行
    try:
        resp = requests.get(f"{OLLAMA_BASE_URL}/api/tags", timeout=10)
        if resp.status_code == 200:
            models = resp.json().get("models", [])
            model_names = [m.get("name", "") for m in models]
            logger.info(f"  Ollama 服务运行中，已安装模型: {model_names}")
            # 测试2: 检查目标模型
            target_found = any(OLLAMA_MODEL_NAME in n for n in model_names)
            if target_found:
                logger.info(f" OK 模型 {OLLAMA_MODEL_NAME} 已安装")
            else:
                logger.error( f" FAIL 模型 {OLLAMA_MODEL_NAME} 未找到!\n"
                    f" 请执行: ollama pull {OLLAMA_MODEL_NAME}" )
                return False
        else:
            logger.error(f" Ollama 返回异常状态码: {resp.status_code}")
            return False
    except requests.exceptions.ConnectionError:
        logger.error( f" FAIL无法连接Ollama 服务 ({OLLAMA_BASE_URL})\n"
            f" 请确认: 1) Ollama 已安装  2) 执行ollama serve 启动服务" )
        return False
    except Exception as e:
        logger.error(f"  FAIL Ollama 连接异常: {e}")
        return False
    # 测试3: 简单推理测试
    try:
        from llama_index.llms.ollama import Ollama
        test_llm = Ollama( model=OLLAMA_MODEL_NAME,
            base_url=OLLAMA_BASE_URL, request_timeout=60.0, )
        test_resp = test_llm.complete("你好，请用一句话介绍自己。")
        logger.info(f"  OK 推理测试成功: {str(test_resp)[:80]}...")
    except Exception as e:
        logger.error(f"  FAIL 推理测试失败: {e}")
        return False
    return True
def test_embedding_model():
    """
    测试嵌入模型是否可用。
    诊断步骤:
      1. 加载模型
      2. 计算测试文本的嵌入向量
      3. 验证向量维度和相似度计算
    """
    logger.info("=" * 60)
    logger.info("嵌入模型测试")
    logger.info("=" * 60)
    try:
        embed_model = HuggingFaceEmbedding( model_name=EMBEDDING_MODEL_PATH,
            device="cpu", max_length=128, normalize=True, trust_remote_code=True, )
        # 测试1: 基础嵌入
        vec = embed_model.get_text_embedding("今天天气真好")
        logger.info(f"  OK 嵌入计算成功, 向量维度: {len(vec)}")
        # 测试2: 中文语义相似度
        vec1 = embed_model.get_text_embedding("我喜欢看科幻电影")
        vec2 = embed_model.get_text_embedding("推荐一部科幻片")
        vec3 = embed_model.get_text_embedding("今天中午吃什么")
        import numpy as np
        sim_12 = np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2) + 1e-8)
        sim_13 = np.dot(vec1, vec3) / (np.linalg.norm(vec1) * np.linalg.norm(vec3) + 1e-8)
        logger.info(f"  语义相似度测试:")
        logger.info(f"    '我喜欢看科幻电影' vs '推荐一部科幻片': {sim_12:.4f} (应较高)")
        logger.info(f"    '我喜欢看科幻电影' vs '今天中午吃什么': {sim_13:.4f} (应较低)")
        if sim_12 > sim_13:
            logger.info("  OK 中文语义区分度正常")
        else:
            logger.warning("  WARN 语义区分度异常，可能影响检索效果")
        return True
    except Exception as e:
        logger.error(f"  FAIL 嵌入模型测试失败: {e}")
        return False
# ==============================================================================
# [14] 主程序 — 完整运行流程
# ==============================================================================
def main():
    """
    完整运行流程:
      Step 1: 连接诊断
      Step 2: 配置 LLM + 嵌入模型 + 分割器
      Step 3: 加载/创建知识库文档
      Step 4: 按领域构建向量索引
      Step 5: 构建 Agentic RAG 查询引擎
      Step 6: 执行个性化推荐查询
    """
    print("\n" + "=" * 70)
    print("  Agentic RAG 个性化推荐系统")
    print("  LlamaIndex + Ollama(DeepSeek-R1-1.5B) + GTE中文嵌入")
    print("=" * 70 + "\n")
    # ──────────────────────────────────────────────────────
    # Step 1: 连接诊断
    # ──────────────────────────────────────────────────────
    print("Step 1: 环境诊断\n")
    ollama_ok = test_ollama_connection()
    embed_ok = test_embedding_model()
    if not ollama_ok:
        print("\nOllama 连接失败，请检查后重试。")
        print("   常见修复方法:")
        print("   1. 安装 Ollama: https://ollama.com/download")
        print("   2. 拉取模型: ollama pull deepseek-r1:1.5b")
        print("   3. 启动服务: ollama serve")
        return
    if not embed_ok:
        print("\n嵌入模型加载失败，请检查模型路径。")
        print(f"   当前路径: {EMBEDDING_MODEL_PATH}")
        print("   请确认该目录下包含 config.json, model.safetensors 等文件")
        return
    print("\n所有组件诊断通过\n")
    # ──────────────────────────────────────────────────────
    # Step 2: 配置全局模型
    # ──────────────────────────────────────────────────────
    print("Step 2: 配置模型\n")
    llm = setup_llm()
    embed_model = setup_embedding()
    node_parser = setup_node_parser()
    # ──────────────────────────────────────────────────────
    # Step 3: 创建知识库
    # ──────────────────────────────────────────────────────
    print("\nStep 3: 构建知识库\n")
    documents = create_sample_documents()
    print(f"  知识库包含 {len(documents)} 篇文档")
    # ──────────────────────────────────────────────────────
    # Step 4: 按领域构建向量索引
    # ──────────────────────────────────────────────────────
    print("\nStep 4: 构建向量索引\n")
    category_indexes = build_category_indexes(documents)
    # ──────────────────────────────────────────────────────
    # Step 5: 构建 Agentic RAG 引擎
    # ──────────────────────────────────────────────────────
    print("\nStep 5: 构建 Agentic RAG 查询引擎\n")
    workflow = build_agentic_query_engine(category_indexes)
    # ──────────────────────────────────────────────────────
    # Step 6: 个性化推荐查询测试
    # ──────────────────────────────────────────────────────
    print("\nStep 6: 个性化推荐测试\n")
    print("=" * 60)
    # 测试1: 电影推荐（带用户画像）
    print("\n--- 测试1: 带用户画像的电影推荐 ---")
    user_profile_1 = { "name": "小明", "age": 25, "interests": ["科幻", "烧脑"], "favorite_director": "诺兰", }
    query_1 = "推荐一部适合我的电影"
    print(f"用户画像: {json.dumps(user_profile_1, ensure_ascii=False)}")
    print(f"查询: {query_1}")
    print("Agent 推理过程:")
    response_1 = personalized_recommend(workflow, query_1, user_profile_1)
    print(f"\n推荐结果:\n{response_1}")
    # 测试2: 书籍推荐
    print("\n\n--- 测试2: 书籍推荐 ---")
    query_2 = "推荐一本好看的科幻小说"
    print(f"查询: {query_2}")
    print("Agent 推理过程:")
    response_2 = personalized_recommend(workflow, query_2)
    print(f"\n推荐结果:\n{response_2}")
    # 测试3: 美食推荐（带饮食偏好）
    print("\n\n--- 测试3: 带饮食偏好的美食推荐 ---")
    user_profile_3 = { "name": "小红", "taste": "喜欢精致料理，不太能吃辣", "occasion": "周末朋友聚餐", }
    query_3 = "推荐一个适合聚餐的美食"
    print(f"用户画像: {json.dumps(user_profile_3, ensure_ascii=False)}")
    print(f"查询: {query_3}")
    print("Agent 推理过程:")
    response_3 = personalized_recommend(workflow, query_3, user_profile_3)
    print(f"\n推荐结果:\n{response_3}")
    # 测试4: 跨领域查询（测试 Agent 的工具选择能力）
    print("\n\n--- 测试4: 跨领域模糊查询 ---")
    query_4 = "周末放松有什么好推荐的？"
    print(f"查询: {query_4}")
    print("Agent 推理过程: (测试Agent是否能选择多个领域工具)")
    response_4 = personalized_recommend(workflow, query_4)
    print(f"\n推荐结果:\n{response_4}")
    # ──────────────────────────────────────────────────────
    # 交互式查询模式
    # ──────────────────────────────────────────────────────
    print("\n\n" + "=" * 60)
    print("  进入交互式推荐模式 (输入 'quit' 退出)")
    print("=" * 60)
    while True:
        try:
            user_input = input("\n请输入推荐请求: ").strip()
            if user_input.lower() in ("quit", "exit", "q", "退出"):
                print("再见！")
                break
            if not user_input:
                continue
            # 可选: 输入用户画像
            profile_input = input("输入用户画像 (JSON格式，直接回车跳过): ").strip()
            profile = None
            if profile_input:
                try:
                    profile = json.loads(profile_input)
                except json.JSONDecodeError:
                    print("  (JSON解析失败，使用空画像)")
            print("\nAgent 推理过程:")
            response = personalized_recommend(workflow, user_input, profile)
            print(f"\n推荐结果:\n{response}")
        except KeyboardInterrupt:
            print("\n再见！")
            break
        except Exception as e:
            logger.error(f"查询出错: {e}")
            print(f"出错: {e}")
# ==============================================================================
# [15] 入口
# ==============================================================================
if __name__ == "__main__":
    main()

3.2 过程展现

3.2.1 命令行操作

3.2.2 IDE操作

4 部署试运行

4.1 基本操作

1 安装python运行环境

2 解压缩软件运行包

3 Ollama安装与模型部署

4 嵌入模型准备

4.2 试运行与测试

5 丰富完善增强

7.1 方案与架构

龙虾机器人MuleRun：给出CPU和Windows下采用LlamaIndex框架、基于Agentic RAG架构的简单个性化推荐的编码，整个程序用一个文件实现，注意各个依赖库的版本对应和中文的检索加强，不使用CONDA，LLM模型采用Ollama部署本地的DeepSeek-R1-8B，需要的嵌入模型己通过modelscope下载在本地D:\01ifmts\models\nlp_gte_sentence-embedding_chinese-base，详细说明各个实现过程和测调试运行。注意编码精简与优化，增强预装案例外的个性化推荐适应能力。

7.2 编码运行