llamaindex笔记

配置大模型api提问

from llama_index.llms.deepseek import DeepSeek

MODEL="deepseek-ai/DeepSeek-V3"
API_KEY=''
BASE_URL='https://api.siliconflow.cn/v1'

llm=DeepSeek(api_base=BASE_URL,model=MODEL,api_key=API_KEY)

response=llm.complete("河北三河是否可以化为北京呢？")
print(response)

消息列表对话

from llama_index.llms.deepseek import DeepSeek
from llama_index.core.llms import ChatMessage

MODEL="deepseek-ai/DeepSeek-V3"
API_KEY=''
BASE_URL='https://api.siliconflow.cn/v1'

llm=DeepSeek(api_base=BASE_URL,model=MODEL,api_key=API_KEY)

message=[
        ChatMessage(
            role='system',content='你作为一个易经专家，对我给出的八字进行测算'
            ),
        ChatMessage(
            role='user',content='1983年腊月初一 19:30出生'
            )
        ]

resp=llm.chat(message)

print(resp)

消息对话流式输出

from llama_index.llms.deepseek import DeepSeek
from llama_index.core.llms import ChatMessage

MODEL="deepseek-ai/DeepSeek-V3"
API_KEY=''
BASE_URL='https://api.siliconflow.cn/v1'

llm=DeepSeek(api_base=BASE_URL,model=MODEL,api_key=API_KEY)

message=[
        ChatMessage(
            role='system',content='你作为一个易经专家，对我给出的八字进行测算'
            ),
        ChatMessage(
            role='user',content='1982年腊月初一 19:22'
            )
        ]

resp=llm.stream_chat(message)

for r in resp:
    print(r.delta,end='')

完整RAG demo

import asyncio
import os
from llama_index.llms.deepseek import DeepSeek
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.agent import AgentRunner
from llama_index.core.tools import FunctionTool
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.core.settings import Settings
from typing import List

# 设置tokenizers并行处理
os.environ["TOKENIZERS_PARALLELISM"] = "false"

# API配置
API_KEY = ""
MODEL = "deepseek-ai/DeepSeek-V3"
BASE_URL = "https://api.siliconflow.cn/v1"

# 设置embedding模型路径
MODEL_PATH = "./model_cache/models--shibing624--text2vec-base-chinese/snapshots/183bb99aa7af74355fb58d16edf8c13ae7c5433e"

# 设置embedding模型
embed_model = HuggingFaceEmbedding(
    model_name=MODEL_PATH
)

# 初始化 LLM
llm = DeepSeek(api_base=BASE_URL, model=MODEL, api_key=API_KEY)

# 设置全局配置
Settings.embed_model = embed_model
Settings.llm = llm

# 加载文档
documents = SimpleDirectoryReader("documents").load_data()
# 从文档创建向量索引
index = VectorStoreIndex.from_documents(
    documents,
    embed_model=embed_model,
    llm=llm
)
# 创建查询引擎
query_engine = index.as_query_engine(
    llm=llm
)

def search_documents(query: str) -> str:
    """
    使用 query_engine 搜索文档。
    Args:
        query (str): 搜索查询语句
    Returns:
        str: 搜索结果
    """
    response = query_engine.query(query)  # 使用同步查询
    return str(response)

# 创建工具
search_tool = FunctionTool.from_defaults(fn=search_documents)

# 定义 Agent
agent = AgentRunner.from_llm(
    llm=llm,
    tools=[search_tool],
    system_prompt="""
    你是一个专业的本地知识库助理。你的任务是：

    1.  **接收用户的问题**：仔细理解用户提出的问题。
    2.  **在本地知识库中搜索相关信息**：使用你掌握的工具，在本地知识库中搜索与问题相关的信息。
    3.  **总结搜索结果**：对搜索到的信息进行总结和提炼，去除冗余信息，保留关键内容。
    4.  **清晰、简洁地回答问题**：使用清晰、简洁的语言回答用户的问题。
    5.  **明确指出信息来源**：在回答的最后，务必明确指出搜索内容来自哪个文档，以便用户追溯信息来源。
    6.  **处理未找到信息的情况**：如果在本地知识库中没有找到与问题相关的信息，请明确告知用户“未找到相关内容”。

    请严格按照以上步骤执行任务，确保回答的准确性、完整性和可追溯性。
    """,
)

def main():
    """主函数，用于运行智能体"""
    response = agent.chat("太虚大师的人间佛教说的什么")  # 使用同步chat方法
    print(response)

if __name__ == "__main__":
    """如果作为主程序运行，则执行以下代码"""
    main()

智体工具

这个非常好用，补充了相关的关键词词典，弥补了数据上的不足；

import asyncio
import os
from llama_index.llms.deepseek import DeepSeek
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.agent import AgentRunner
from llama_index.core.tools import FunctionTool
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.core.settings import Settings
from typing import List

# 设置tokenizers并行处理
os.environ["TOKENIZERS_PARALLELISM"] = "false"

# API配置
API_KEY = os.environ.get("DEEPSEEK_API_KEY", "")
MODEL = "deepseek-ai/DeepSeek-V3"
BASE_URL = "https://api.siliconflow.cn/v1"

# 设置embedding模型路径
MODEL_PATH = "./model_cache/models--shibing624--text2vec-base-chinese/snapshots/183bb99aa7af74355fb58d16edf8c13ae7c5433e"

# 设置embedding模型
embed_model = HuggingFaceEmbedding(
    model_name=MODEL_PATH
)

# 初始化 LLM
llm = DeepSeek(api_base=BASE_URL, model=MODEL, api_key=API_KEY)

# 设置全局配置
Settings.embed_model = embed_model
Settings.llm = llm

# 加载文档
documents = SimpleDirectoryReader("documents").load_data()
# 从文档创建向量索引
index = VectorStoreIndex.from_documents(
    documents,
    embed_model=embed_model,
    llm=llm
)
# 创建查询引擎
query_engine = index.as_query_engine(
    llm=llm
)

def search_documents(query: str) -> str:
    """
    使用 query_engine 搜索文档。
    Args:
        query (str): 搜索查询语句
    Returns:
        str: 搜索结果
    """
    response = query_engine.query(query)  # 使用同步查询
    return str(response)

def get_person_details(person_name: str) -> str:
    """
    根据人物姓名搜索并返回人物的详细信息。
    Args:
        person_name (str): 人物姓名
    Returns:
        str: 人物详细信息，如果未找到则返回 "未找到该人物的信息"
    """
    # 模拟从数据库或API获取人物信息
    person_info = {
        "太虚大师": "太虚大师是中国近现代佛教的重要人物，他提倡人间佛教，主张佛教应该关注社会现实，服务于人民。",
        "观世音菩萨": "观世音菩萨是佛教中的一位重要菩萨，代表慈悲和智慧，被广泛信仰。",
        # 更多人物信息...
    }
    if person_name in person_info:
        return person_info[person_name]
    else:
        return "未找到该人物的信息"

# 创建工具
search_tool = FunctionTool.from_defaults(fn=search_documents)
person_details_tool = FunctionTool.from_defaults(fn=get_person_details)

# 定义 Agent
agent = AgentRunner.from_llm(
    llm=llm,
    tools=[search_tool, person_details_tool],
    system_prompt="""
    你是一个专业的本地知识库助理。你的任务是：

    1.  **接收用户的问题**：仔细理解用户提出的问题。
    2.  **如果用户的问题是关于某个人的详细信息，请首先使用 `get_person_details` 工具获取该人物的详细信息。**
    3.  **在本地知识库中搜索相关信息**：使用你掌握的工具，在本地知识库中搜索与问题相关的信息。
    4.  **总结搜索结果**：对搜索到的信息进行总结和提炼，去除冗余信息，保留关键内容。
    5.  **清晰、简洁地回答问题**：使用清晰、简洁的语言回答用户的问题。
    6.  **明确指出信息来源**：在回答的最后，务必明确指出搜索内容来自哪个文档，以便用户追溯信息来源。
    7.  **处理未找到信息的情况**：如果在本地知识库中没有找到与问题相关的信息，请明确告知用户“未找到相关内容”。

    请严格按照以上步骤执行任务，确保回答的准确性、完整性和可追溯性。
    """,
)

def main():
    """主函数，用于运行智能体"""
    response = agent.chat("请详细介绍太虚大师，包括他的人间佛教思想和主要贡献。")  # 使用同步chat方法
    print(response)

if __name__ == "__main__":
    """如果作为主程序运行，则执行以下代码"""
    main()