- 你好, 北京时间现在几点？
- 您好，建议您联网获取时效性较强的信息；如果还有其他问题需要帮助，请随时告诉我！

官方文档: https://docs.llamaindex.ai/
LlamaIndex 是一个基于LLM的应用程序的数据框架，该应用程序受益于上下文增强，是典型的RAG系统
LlamaIndex 提供了必要的抽象，可以更轻松地摄取、构建和访问私有或特定领域的数据，以便将这些数据安全可靠地注入 LLM 中，以实现更准确的文本生成

from llama_index import GPTVectorStoreIndex

# 初始化 Llama-Index
index = GPTVectorStoreIndex()

# 添加单个文档
doc_text = "这是文档内容。"
index.insert(doc_text)

# 添加多个文档
doc_texts = ["文档1内容。", "文档2内容。"]
for text in doc_texts:
    index.insert(text)
    
# 构建索引
index.build()


# 执行查询
query = "查询内容"
response = index.query(query)
print(response)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Ollama + DeepSeek-R1-1.5B的RAG检索脚本
加载本地语料库，通过Ollama调用DeepSeek模型生成回答
"""

# 导入核心模块
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings, Document

# 导入Ollama的LLM和嵌入模型
from llama_index.llms.ollama import Ollama
from llama_index.embeddings.ollama import OllamaEmbedding

# 配置Ollama连接(DeepSeek-R1-1.5B模型和嵌入模型)
Settings.llm = Ollama(
    model="deepseek-r1:1.5b",
    base_url="http://localhost:11434",
    temperature=0.1
)
  
# 加载 nomic-embed-text 向量分析
# 记得运行 sudo -u ollama ollama pull nomic-embed-text
Settings.embed_model = OllamaEmbedding(
    model_name="nomic-embed-text",
    base_url="http://localhost:11434"
)
 
# 手动加载语料的备选方式(无需文件加载)
# 这里后续会从数据库当中加载提取内容
documents = [
    Document(text="DeepSeek-R1-1.5B是轻量级大模型，适合边缘部署", metadata={"title": "DeepSeek特性"}),
    Document(text="Ollama支持一键运行GGUF格式的大模型", metadata={"title": "Ollama功能"})
]

# 构建向量索引
index = VectorStoreIndex.from_documents(documents)

# 创建查询引擎并测试
query_engine = index.as_query_engine()
response = query_engine.query("DeepSeek-R1-1.5B的特点是什么？")

# 输出结果
print("查询结果：\n", response)

查询结果：
 DeepSeek-R1-1.5B的特点包括：

1. **轻量级大模型**：该模型设计为轻量化，适用于资源受限的环境，如边缘计算。

2. **适合边缘部署**：能够高效运行在边缘设备上，提升实时处理能力。

3. **Ollama功能支持**：通过一键运行GGUF格式的大模型，简化了部署流程。

# 其他略

# 引入时间库
import datetime


# 获取时间并格式化输出
current_time = datetime.datetime.now()
formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
documents = [
    Document(text=f"当前时间为{formatted_time}", metadata={"title": "当前时间"})
]

# 其他略

# 创建查询引擎并测试
query_engine = index.as_query_engine()
response = query_engine.query("当前时间")

# 输出结果
print("查询结果：\n", response)

------- 导入语料库之前 ------------
- 当前时间
- 您好，建议您联网获取时效性较强的信息；如果还有其他问题需要帮助，请随时告诉我！

------- 导入语料库之后 ------------
- 当前时间
- 2025-11-26 11:38:56

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from fastapi import FastAPI
import uvicorn
import datetime
from llama_index.core import VectorStoreIndex, Settings, Document
from llama_index.llms.ollama import Ollama
from llama_index.embeddings.ollama import OllamaEmbedding

# 初始化FastAPI
app = FastAPI(title="DeepSeek-R1 RAG API", version="1.0")

# 配置Ollama
Settings.llm = Ollama(model="deepseek-r1:1.5b", base_url="http://localhost:11434", temperature=0.1)
Settings.embed_model = OllamaEmbedding(model_name="nomic-embed-text", base_url="http://localhost:11434")

# 生成语料库（可替换为数据库加载）
def get_corpus():
    current_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    return [
        Document(text=f"当前时间为{current_time}", metadata={"title": "当前时间"}),
        Document(text="DeepSeek-R1-1.5B是轻量级大模型，适合边缘部署", metadata={"title": "DeepSeek特性"}),
        Document(text="Ollama支持一键运行GGUF格式的大模型", metadata={"title": "Ollama功能"})
    ]

# 构建全局索引（避免每次查询重新构建）
index = VectorStoreIndex.from_documents(get_corpus())
query_engine = index.as_query_engine()

# 定义API接口
@app.get("/rag/query")
def rag_query(question: str):
    """RAG问答接口，参数：question（查询问题）"""
    response = query_engine.query(question)
    return {
        "question": question,
        "answer": str(response),
        "corpus_source": [doc.text for doc in response.source_nodes]  # 返回匹配的语料
    }

if __name__ == "__main__":
    # 启动API服务(监听所有IP，端口8000)
    uvicorn.run(app, host="0.0.0.0", port=8000)

from fastapi import FastAPI, Body, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field, validator
from typing import List, Optional, Dict, Any
# 仅保留低版本兼容的核心导入
from llama_index.core import VectorStoreIndex, Document, Settings
from llama_index.llms.ollama import Ollama
from llama_index.embeddings.ollama import OllamaEmbedding
import os

# 初始化FastAPI应用
app = FastAPI(title="Ollama-RAG 动态语料服务", version="1.0")

# -------------------------- 全局配置与索引管理 --------------------------
# 配置Ollama LLM和嵌入模型（本地Ollama需启动并拉取对应模型）
Settings.llm = Ollama(model="deepseek-r1:1.5b", base_url="http://localhost:11434", temperature=0.1)
Settings.embed_model = OllamaEmbedding(model_name="nomic-embed-text", base_url="http://localhost:11434")

# 全局索引缓存：保存最新的语料库索引（低版本默认内存存储）
GLOBAL_RAG_INDEX = None

# -------------------------- Pydantic模型：定义Document请求结构 --------------------------
class RequestDocument(BaseModel):
    """外部传入的Document对象结构，贴合llama-index的Document原生字段"""
    text: str = Field(..., description="文档的核心文本内容，不能为空")
    doc_id: Optional[str] = Field(None, description="文档唯一标识，不传入则自动生成")
    metadata: Optional[Dict[str, Any]] = Field({}, description="文档元数据，如作者、来源、时间等")

    @validator("text")
    def text_not_empty(cls, v):
        """校验text字段不能为空"""
        if not v.strip():
            raise ValueError("文档text字段不能为空或仅包含空白字符")
        return v

# -------------------------- 核心工具函数 --------------------------
def build_index_from_documents(documents: List[Document]) -> VectorStoreIndex:
    """
    从llama-index的Document列表构建向量索引（低版本默认内存存储）
    :param documents: Document对象列表
    :return: 构建后的VectorStoreIndex
    """
    if not documents:
        raise ValueError("语料库不能为空，无法构建索引")
    # 低版本核心：直接构建索引，无需手动指定StorageContext和向量存储
    index = VectorStoreIndex.from_documents(documents)
    return index

def convert_request_docs_to_llama_docs(request_docs: List[RequestDocument]) -> List[Document]:
    """
    将外部传入的RequestDocument转换为llama-index的原生Document对象
    :param request_docs: 外部请求的Document列表
    :return: llama-index的Document列表
    """
    llama_docs = []
    for req_doc in request_docs:
        # 自动生成doc_id（如果外部未传入）
        doc_id = req_doc.doc_id or f"doc_{os.urandom(6).hex()}"  # 6位16进制随机数，更唯一
        # 构建llama-index的Document对象（低版本兼容）
        llama_doc = Document(
            text=req_doc.text,
            doc_id=doc_id,
            metadata=req_doc.metadata or {}
        )
        llama_docs.append(llama_doc)
    return llama_docs

# -------------------------- API接口定义 --------------------------
@app.post("/rag/corpus", summary="提交Document格式的语料库并构建最新索引", tags=["语料库管理"])
async def submit_corpus(
    documents: List[RequestDocument] = Body(..., description="Document格式的语料库JSON数组，每个元素包含text、可选doc_id和metadata")
):
    """
    接收外部传入的Document格式JSON数组，转换为llama-index原生Document后构建最新RAG索引
    每次提交会覆盖原有索引，如需增量添加可修改build_index_from_documents逻辑
    """
    global GLOBAL_RAG_INDEX
    try:
        # 1. 转换外部请求的Document为llama-index原生对象
        llama_docs = convert_request_docs_to_llama_docs(documents)
        
        # 2. 构建最新索引
        GLOBAL_RAG_INDEX = build_index_from_documents(llama_docs)
        
        # 3. 统计语料库信息
        doc_ids = [doc.doc_id for doc in llama_docs]
        metadata_keys = [list(doc.metadata.keys()) for doc in llama_docs]
        
        return JSONResponse(content={
            "status": "success",
            "message": f"语料库提交成功，已构建最新索引（共{len(llama_docs)}个文档）",
            "corpus_stats": {
                "total_documents": len(llama_docs),
                "document_ids": doc_ids,
                "metadata_sample": metadata_keys[:3]  # 仅返回前3个文档的元数据键，避免数据过大
            }
        })
    
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"语料库提交失败：{str(e)}")

@app.post("/rag/query", summary="基于最新语料库执行RAG查询", tags=["RAG查询"])
async def rag_query(
    query: str = Body(..., description="用户查询问题"),
    top_k: int = Body(3, description="检索最相似的文档数量，默认3"),
    response_mode: str = Body("compact", description="回答模式：compact/simple/tree等"),
    with_metadata: bool = Body(True, description="是否返回源文档的元数据，默认True")
):
    """
    基于最新提交的Document语料库执行RAG查询，返回检索增强后的回答及源文档信息
    """
    global GLOBAL_RAG_INDEX
    try:
        # 校验索引是否存在
        if GLOBAL_RAG_INDEX is None:
            raise HTTPException(status_code=400, detail="暂无最新语料库索引，请先通过/rag/corpus提交Document格式的语料库")
        
        # 构建查询引擎（低版本兼容）
        query_engine = GLOBAL_RAG_INDEX.as_query_engine(
            similarity_top_k=top_k,
            response_mode=response_mode
        )
        
        # 执行查询
        response = query_engine.query(query)
        
        # 提取检索的源文档信息（包含元数据）
        source_nodes = []
        for node in response.source_nodes:
            source_info = {
                "doc_id": node.node.ref_doc_id if hasattr(node.node, 'ref_doc_id') else node.node.doc_id,
                "similarity_score": round(node.score, 4) if hasattr(node, 'score') else 0.0,  # 兼容极低版本无score的情况
                "content_snippet": node.node.text[:200] + "..." if len(node.node.text) > 200 else node.node.text
            }
            # 可选返回元数据
            if with_metadata and hasattr(node.node, 'metadata'):
                source_info["metadata"] = node.node.metadata
            source_nodes.append(source_info)
        
        return JSONResponse(content={
            "status": "success",
            "query": query,
            "answer": str(response),
            "retrieval_result": {
                "top_k": top_k,
                "matched_documents": source_nodes,
                "total_matched": len(source_nodes)
            }
        })
    
    except HTTPException as e:
        raise e
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"RAG查询失败：{str(e)}")

@app.get("/rag/status", summary="查看RAG服务状态", tags=["服务监控"])
async def rag_status():
    """
    查看最新语料库索引的状态及文档数量
    """
    global GLOBAL_RAG_INDEX
    if GLOBAL_RAG_INDEX is None:
        return JSONResponse(content={
            "status": "uninitialized",
            "message": "尚未提交Document语料库，索引未构建"
        })
    else:
        # 低版本兼容的文档数量统计
        doc_count = len(GLOBAL_RAG_INDEX.docstore.docs) if hasattr(GLOBAL_RAG_INDEX.docstore, 'docs') else 0
        sample_doc_id = list(GLOBAL_RAG_INDEX.docstore.docs.keys())[:1] if doc_count > 0 else []
        return JSONResponse(content={
            "status": "initialized",
            "message": "Document语料库索引已构建",
            "index_stats": {
                "total_documents": doc_count,
                "sample_document_id": sample_doc_id
            }
        })

# 启动服务
if __name__ == "__main__":
    import uvicorn
    # 监听所有网络接口，端口8000
    uvicorn.run(app, host="0.0.0.0", port=8000)