[Datawheel学习]用Llama-index创建Agent、数据库对话Agent和RAG接入Agent

1.Llama-index创建Agent

1.0 背景知识

1. 什么是Llama-index?
   LlamaIndex（原名GPT Index）是一个专为大语言模型（LLMs）设计的数据框架，旨在帮助用户将外部数据与LLMs结合，实现更高效的数据检索和知识增强生成（RAG）。它通过构建索引和提供查询接口，使LLMs能够访问和利用私有或特定领域的数据，从而提升模型的准确性和实用性。
2. LlamaIndex 的核心功能
   数据连接器（Data Connectors） 支持从多种数据源（如本地文件、PDF、API、SQL数据库、Notion、Google文档等）摄取数据，并将其转换为统一的文档表示形式。
   索引结构（Index Structures） 将数据组织成可查询的索引形式，支持多种索引类型，包括：
   1）向量索引（Vector Store Index）：基于向量相似度检索数据。
   2）列表索引（List Index）：按顺序存储节点，支持关键字过滤。
   3）树形索引（Tree Index）：构建层次化结构，支持从根节点到叶节点的查询。
   4）关键字表索引（Keyword Table Index）：通过关键字映射快速检索节点。
   查询接口（Query Interface） 提供与大模型对话的接口，支持自然语言查询。通过检索索引和组合Prompt，使LLMs能够生成基于外部数据的准确回答。
   检索增强生成（RAG） LlamaIndex的核心应用场景是RAG，它通过以下步骤实现：
   1）索引阶段：将外部数据构建为知识库。
   2）查询阶段：从知识库中检索相关上下文，并将其与用户查询结合，生成增强后的响应。
3. LlamaIndex 的优势
   扩展性： 支持多种数据源和格式，扩展了LLMs的应用范围。
   灵活性： 允许用户自定义索引和查询逻辑，适应不同场景需求。
   实时性： 通过实时检索外部数据，确保模型提供最新信息。

4.Llamaindex的竞品
1） LangChain

• 特点：是一个多功能框架，支持构建复杂的 LLM 应用，包括聊天机器人、自动化工作流和多步骤任务处理。提供模块化设计，支持与多种数据源和工具集成，适合需要灵活性和复杂功能的场景。强调上下文记忆和任务编排能力，适合需要长时间交互和多步骤推理的应用。
• 适用场景：需要复杂交互和上下文保留的应用（如客户支持、聊天机器人）。需要与其他系统广泛集成的通用应用程序。

2） Flowise AI

• 特点：提供无代码（No-Code）开发界面，支持通过拖放组件快速构建 LLM 应用。与 LangChain 深度整合，支持其核心功能（如链式操作、数据增强等），但降低了开发门槛。
• 适用场景：适合非技术用户或快速原型开发。需要可视化工作流和低代码解决方案的场景。

3）AutoChain

• 特点：轻量级框架，专注于对话式智能代理的开发。强调简单性、自定义能力和自动化评估，适合快速构建和测试对话系统。
• 适用场景：需要快速构建对话代理的场景。适合初学者或需要高度定制化的对话系统。

4）Haystack

• 特点：专注于文档检索和问答系统，支持多种数据源和检索算法。提供强大的文档处理和检索功能，适合需要高效信息提取的应用。
• 适用场景：文档问答系统、知识库检索。需要高效处理非结构化数据的场景。

5）Weaviate

• 特点：是一个向量数据库，支持语义搜索和高效的数据检索。提供与 LLM 的集成能力，适合需要高性能向量检索的应用15。
• 适用场景：需要高效向量检索和语义搜索的场景。适合与 LLM 结合使用的知识库和推荐系统。

6）Pinecone

• 特点：专注于向量搜索和存储，支持大规模数据的高效检索。提供与 LLM 的集成能力，适合需要实时检索和上下文感知的应用。
• 适用场景：实时推荐系统、语义搜索。需要高性能向量检索的场景。

7）OpenAI Embeddings + FAISS

• 特点：结合 OpenAI 的嵌入模型和 FAISS 向量搜索库，提供高效的语义检索能力。适合需要自定义检索逻辑和高性能搜索的场景。
• 适用场景：需要高度定制化检索逻辑的应用。适合技术团队构建高性能搜索系统。

1.1 准备模型

import os
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()
# 从环境变量中读取api_key
api_key = os.getenv('ZISHU_API_KEY')
base_url = "http://43.200.7.56:8008/v1"
chat_model = "glm-4-flash"
emb_model = "embedding-3"

1.2自定义LLM类

from openai import OpenAI
from pydantic import Field  # 导入Field，用于Pydantic模型中定义字段的元数据
from llama_index.core.llms import (
    CustomLLM,
    CompletionResponse,
    LLMMetadata,
)
from llama_index.core.embeddings import BaseEmbedding
from llama_index.core.llms.callbacks import llm_completion_callback
from typing import List, Any, Generator
# 定义OurLLM类，继承自CustomLLM基类
class OurLLM(CustomLLM):
    api_key: str = Field(default=api_key)
    base_url: str = Field(default=base_url)
    model_name: str = Field(default=chat_model)
    client: OpenAI = Field(default=None, exclude=True)  # 显式声明 client 字段

    def __init__(self, api_key: str, base_url: str, model_name: str = chat_model, **data: Any):
        super().__init__(**data)
        self.api_key = api_key
        self.base_url = base_url
        self.model_name = model_name
        self.client = OpenAI(api_key=self.api_key, base_url=self.base_url)  # 使用传入的api_key和base_url初始化 client 实例

    @property
    def metadata(self) -> LLMMetadata:
        """Get LLM metadata."""
        return LLMMetadata(
            model_name=self.model_name,
        )

    @llm_completion_callback()
    def complete(self, prompt: str, **kwargs: Any) -> CompletionResponse:
        response = self.client.chat.completions.create(model=self.model_name, messages=[{"role": "user", "content": prompt}])
        if hasattr(response, 'choices') and len(response.choices) > 0:
            response_text = response.choices[0].message.content
            return CompletionResponse(text=response_text)
        else:
            raise Exception(f"Unexpected response format: {response}")

    @llm_completion_callback()
    def stream_complete(
        self, prompt: str, **kwargs: Any
    ) -> Generator[CompletionResponse, None, None]:
        response = self.client.chat.completions.create(
            model=self.model_name,
            messages=[{"role": "user", "content": prompt}],
            stream=True
        )

        try:
            for chunk in response:
                chunk_message = chunk.choices[0].delta
                if not chunk_message.content:
                    continue
                content = chunk_message.content
                yield CompletionResponse(text=content, delta=content)

        except Exception as e:
            raise Exception(f"Unexpected response format: {e}")

llm = OurLLM(api_key=api_key, base_url=base_url, model_name=chat_model)

1.3测试大模型是否可用

response = llm.stream_complete("你是谁？")
for chunk in response:
    print(chunk, end="", flush=True)

1.4测试结果

1.5 编写工具函数并引入

这里重新定义了加法和乘法，在提问后要求大模型调用工具函数完成任务而不是直接回答。这里注意：大模型会根据函数的注释来判断使用哪个函数来完成任务。所以，注释一定要写清楚函数的功能和返回值。

import sys
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))

from llama_index.core.agent import ReActAgent
from llama_index.core.tools import FunctionTool


def multiply(a: float, b: float) -> float:
    """Multiply two numbers and returns the product"""
    return a * b


def add(a: float, b: float) -> float:
    """Add two numbers and returns the sum"""
    return a + b


def main():

    multiply_tool = FunctionTool.from_defaults(fn=multiply)
    add_tool = FunctionTool.from_defaults(fn=add)

    # 创建ReActAgent实例
    agent = ReActAgent.from_tools([multiply_tool, add_tool], llm=llm, verbose=True)

    response = agent.chat("20+（2*4）等于多少？使用工具计算每一步")

    print(response)


if __name__ == "__main__":
    main()

1.6 结果展示

1.7 写一个查询天气的工具函数，实现天气查询agent

def get_weather(city: str) -> int:
    """
    Gets the weather temperature of a specified city.

    Args:
    city (str): The name or abbreviation of the city.

    Returns:
    int: The temperature of the city. Returns 20 for 'NY' (New York),
         30 for 'BJ' (Beijing), and -1 for unknown cities.
    """

    # Convert the input city to uppercase to handle case-insensitive comparisons
    city = city.upper()

    # Check if the city is New York ('NY')
    if city == "NY":
        return 20  # Return 20°C for New York

    # Check if the city is Beijing ('BJ')
    elif city == "BJ":
        return 30  # Return 30°C for Beijing

    # If the city is neither 'NY' nor 'BJ', return -1 to indicate unknown city
    else:
        return -1

weather_tool = FunctionTool.from_defaults(fn=get_weather)

agent = ReActAgent.from_tools([multiply_tool, add_tool, weather_tool], llm=llm, verbose=True)

response = agent.chat("纽约天气怎么样?")

这里写了一个伪函数，当然也可以去调用天气查询的api。

2.数据库对话Agent

2.1 创建数据库并写入数据

import sqlite3
# 创建数据库
sqllite_path = 'llmdb.db'
con = sqlite3.connect(sqllite_path)

# 创建表
sql = """
CREATE TABLE `section_stats` (
  `部门` varchar(100) DEFAULT NULL,
  `人数` int(11) DEFAULT NULL
);
"""
c = con.cursor()
cursor = c.execute(sql)
c.close()
con.close()

2.2 构建llm和embedding模型

这里稍微有点麻烦，我这里使用的是本地部署的llm，然后embeeding模型调用的是智谱的api，感觉是在我的场景下比较简单的方式了

2.2.1 llm构建

import os
from dotenv import load_dotenv

from openai import OpenAI
from pydantic import Field  # 导入Field，用于Pydantic模型中定义字段的元数据
from llama_index.core.llms import (
    CustomLLM,
    CompletionResponse,
    LLMMetadata,
)
from llama_index.core.embeddings import BaseEmbedding
from llama_index.core.llms.callbacks import llm_completion_callback
from typing import List, Any, Generator

# 加载环境变量
load_dotenv()
# 从环境变量中读取api_key
api_key = os.getenv('ZISHU_API_KEY')
base_url = os.getenv('base_url')
chat_model = os.getenv('chat_model')
emb_model = "embedding-2"

## 自定义对话模型
# 导入必要的库和模块
from openai import OpenAI
from pydantic import Field  # 导入Field，用于Pydantic模型中定义字段的元数据
from typing import Optional, List, Mapping, Any, Generator
import os

from llama_index.core import SimpleDirectoryReader, SummaryIndex
from llama_index.core.callbacks import CallbackManager
from llama_index.core.llms import (
    CustomLLM,
    CompletionResponse,
    CompletionResponseGen,
    LLMMetadata,
)
from llama_index.core.llms.callbacks import llm_completion_callback
from llama_index.core import Settings

# 定义OurLLM类，继承自CustomLLM基类
class OurLLM(CustomLLM):
    api_key: str = Field(default=api_key)
    base_url: str = Field(default=base_url)
    model_name: str = Field(default=chat_model)
    client: OpenAI = Field(default=None, exclude=True)  # 显式声明 client 字段

    def __init__(self, api_key: str, base_url: str, model_name: str = chat_model, **data: Any):
        super().__init__(**data)
        self.api_key = api_key
        self.base_url = base_url
        self.model_name = model_name
        self.client = OpenAI(api_key=self.api_key, base_url=self.base_url)  # 使用传入的api_key和base_url初始化 client 实例

    @property
    def metadata(self) -> LLMMetadata:
        """Get LLM metadata."""
        return LLMMetadata(
            model_name=self.model_name,
        )

    @llm_completion_callback()
    def complete(self, prompt: str, **kwargs: Any) -> CompletionResponse:
        response = self.client.chat.completions.create(model=self.model_name, messages=[{"role": "user", "content": prompt}])
        if hasattr(response, 'choices') and len(response.choices) > 0:
            response_text = response.choices[0].message.content
            return CompletionResponse(text=response_text)
        else:
            raise Exception(f"Unexpected response format: {response}")

    @llm_completion_callback()
    def stream_complete(
        self, prompt: str, **kwargs: Any
    ) -> Generator[CompletionResponse, None, None]:
        response = self.client.chat.completions.create(
            model=self.model_name,
            messages=[{"role": "user", "content": prompt}],
            stream=True
        )

        try:
            for chunk in response:
                chunk_message = chunk.choices[0].delta
                if not chunk_message.content:
                    continue
                content = chunk_message.content
                yield CompletionResponse(text=content, delta=content)

        except Exception as e:
            raise Exception(f"Unexpected response format: {e}")


llm = OurLLM(api_key=api_key, base_url=base_url, model_name=chat_model)
# 测试对话模型
# response = llm.complete("你是谁？")
# print(response)

2.2.2构建embedding模型

from llama_index.embeddings.zhipuai import ZhipuAIEmbedding
embedding = ZhipuAIEmbedding(
    api_key = 'your_api_key',
    model = emb_model,
)

2.3 导入Llama-index相关的库，并配置对话模型和嵌入模型,构建数据库对话agent

from llama_index.core.agent import ReActAgent  
from llama_index.core.tools import FunctionTool  
from llama_index.core import SimpleDirectoryReader, VectorStoreIndex, Settings  
from llama_index.core.tools import QueryEngineTool   
from llama_index.core import SQLDatabase  
from llama_index.core.query_engine import NLSQLTableQueryEngine  
from sqlalchemy import create_engine, select  


# 配置默认大模型  
Settings.llm = llm
Settings.embed_model = embedding

## 创建数据库查询引擎  
engine = create_engine("sqlite:///llmdb.db")  
# prepare data  
sql_database = SQLDatabase(engine, include_tables=["section_stats"])  
query_engine = NLSQLTableQueryEngine(  
    sql_database=sql_database,   
    tables=["section_stats"],   
    llm=Settings.llm  
)

# 创建工具函数  
def multiply(a: float, b: float) -> float:  
    """将两个数字相乘并返回乘积。"""  
    return a * b  

multiply_tool = FunctionTool.from_defaults(fn=multiply)  

def add(a: float, b: float) -> float:  
    """将两个数字相加并返回它们的和。"""  
    return a + b

add_tool = FunctionTool.from_defaults(fn=add)

# 把数据库查询引擎封装到工具函数对象中  
staff_tool = QueryEngineTool.from_defaults(
    query_engine,
    name="section_staff",
    description="查询部门的人数。"  
)

# 构建ReActAgent
agent = ReActAgent.from_tools([multiply_tool, add_tool, staff_tool], verbose=True)  
# 通过agent给出指令
response = agent.chat("请从数据库表中获取`专利部`和`商标部`的人数，并将这两个部门的人数相加！")  

2.4 结果

2.5 理解

这里对于初学者比较难理解embedding模型起到了什么作用，目前推测是在NLSQLTableQueryEngine类中用于为文本生成向量表示，支持语义理解和检索任务。此外，QueryEngineTool 是 LlamaIndex 框架中的一个工具类，它的作用是将一个 查询引擎（Query Engine）封装成一个 工具（Tool），以便在更高层次的组件（如 ReActAgent 或其他代理）中使用。通过这种方式，查询引擎可以被集成到更复杂的任务流程中，例如多步骤推理、任务分解或工具调用。

这里还有个问题：为什么这里一定要用到embedding模型来理解自然语言llm做不到吗？
虽然 LLM 能够直接理解自然语言，但在某些场景下，Embedding 模型仍然是必要的。以下是 Embedding 模型的主要作用：

1. 向量化表示

• Embedding 模型将文本转换为固定长度的向量表示，这些向量可以用于计算文本之间的相似度。
• 例如，在文档检索、语义搜索等任务中，Embedding 模型可以快速找到与查询文本最相关的文档。

2. 高效检索

• 在大规模数据集中，直接使用 LLM 进行检索可能会非常低效。Embedding 模型可以将文本转换为向量后，使用向量检索技术（如余弦相似度、近似最近邻搜索）快速找到相关结果。
• 例如，如果你有一个包含数百万条记录的数据库，使用 Embedding 模型进行向量检索会比直接使用 LLM 更高效。

3. 语义理解

• Embedding 模型能够捕捉文本的语义信息，使得语义相似的文本在向量空间中距离较近。
• 例如，在问答系统中，Embedding 模型可以用于找到与用户问题最相关的答案。

4. 任务分离

• 在某些任务中，LLM 和 Embedding 模型可以分工合作：
• LLM 负责复杂的推理和任务分解。
• Embedding 模型负责高效的检索和语义匹配。

从代码来看，Embedding 模型并不是必需的，因为：使用的是 NLSQLTableQueryEngine，它直接依赖 LLM 将自然语言查询转换为 SQL 查询，而不需要 Embedding 模型。Embedding 模型的主要优势在于高效的向量化表示和检索能力，适用于大规模数据集或需要语义匹配的场景。如果后续扩展代码（如引入文档检索或语义搜索功能），可以考虑使用 Embedding 模型来增强系统的能力。

3.RAG接入Agent

接下来尝试把RAG当作Agent可以调用的一个工具。引入llm和embedding模型的过程和上面一样不赘述。

3.1 构建索引

# 从指定文件读取，输入为List
from llama_index.core import SimpleDirectoryReader,Document
documents = SimpleDirectoryReader(input_files=['../docs/问答手册.txt']).load_data()

# 构建节点
from llama_index.core.node_parser import SentenceSplitter
transformations = [SentenceSplitter(chunk_size = 512)]

from llama_index.core.ingestion.pipeline import run_transformations
nodes = run_transformations(documents, transformations=transformations)

# 构建索引
from llama_index.vector_stores.faiss import FaissVectorStore
import faiss
from llama_index.core import StorageContext, VectorStoreIndex

emb = embedding.get_text_embedding("你好呀呀")
vector_store = FaissVectorStore(faiss_index=faiss.IndexFlatL2(len(emb)))
storage_context = StorageContext.from_defaults(vector_store=vector_store)

index = VectorStoreIndex(
    nodes = nodes,
    storage_context=storage_context,
    embed_model = embedding,
)

3.2 构建问答引擎

# 构建检索器
from llama_index.core.retrievers import VectorIndexRetriever
# 想要自定义参数，可以构造参数字典
kwargs = {'similarity_top_k': 5, 'index': index, 'dimensions': len(emb)} # 必要参数
retriever = VectorIndexRetriever(**kwargs)

# 构建合成器
from llama_index.core.response_synthesizers  import get_response_synthesizer
response_synthesizer = get_response_synthesizer(llm=llm, streaming=True)

# 构建问答引擎
from llama_index.core.query_engine import RetrieverQueryEngine
engine = RetrieverQueryEngine(
      retriever=retriever,
      response_synthesizer=response_synthesizer,
        )

尝试直接使用RAG回答

# 提问
question = "What are the applications of Agent AI systems ?"
response = engine.query(question)
for text in response.response_gen:
    print(text, end="")

## 3.3 配置问答工具并创建Agent
把这个RAG当作一个工具给Agent调用，让它去思考

# 配置查询工具
from llama_index.core.tools import QueryEngineTool
from llama_index.core.tools import ToolMetadata
query_engine_tools = [
    QueryEngineTool(
        query_engine=engine,
        metadata=ToolMetadata(
            name="RAG工具",
            description=(
                "用于在原文中检索相关信息"
            ),
        ),
    ),
]

# 创建ReAct Agent
from llama_index.core.agent import ReActAgent
agent = ReActAgent.from_tools(query_engine_tools, llm=llm, verbose=True)
# 让Agent完成任务
response = agent.chat("What are the applications of Agent AI systems ?")
print(response)

输出：

4.构建search agent

这里使用了Bocha Web Search API，大家可以直接去官网注册然后申请api，注意这个是收费的。
完事开始构建search agent

from llama_index.core.tools import FunctionTool
import requests
# 需要先把BOCHA_API_KEY填写到.env文件中去。
BOCHA_API_KEY = os.getenv('BOCHA_API_KEY')

# 定义Bocha Web Search工具
def bocha_web_search_tool(query: str, count: int = 8) -> str:
    """
    使用Bocha Web Search API进行联网搜索，返回搜索结果的字符串。
    
    参数:
    - query: 搜索关键词
    - count: 返回的搜索结果数量

    返回:
    - 搜索结果的字符串形式
    """
    url = 'https://api.bochaai.com/v1/web-search'
    headers = {
        'Authorization': f'Bearer {BOCHA_API_KEY}',  # 请替换为你的API密钥
        'Content-Type': 'application/json'
    }
    data = {
        "query": query,
        "freshness": "noLimit", # 搜索的时间范围，例如 "oneDay", "oneWeek", "oneMonth", "oneYear", "noLimit"
        "summary": True, # 是否返回长文本摘要总结
        "count": count
    }

    response = requests.post(url, headers=headers, json=data)

    if response.status_code == 200:
        # 返回给大模型的格式化的搜索结果文本
        # 可以自己对博查的搜索结果进行自定义处理
        return str(response.json())
    else:
        raise Exception(f"API请求失败，状态码: {response.status_code}, 错误信息: {response.text}")

search_tool = FunctionTool.from_defaults(fn=bocha_web_search_tool)
from llama_index.core.agent import ReActAgent
agent = ReActAgent.from_tools([search_tool], llm=llm, verbose=True)

测试一下

# 测试用例
query = "阿里巴巴2024年的ESG报告主要讲了哪些内容？"
response = agent.chat(f"请帮我搜索以下内容：{query}")
print(response)

结果

原文地址：https://blog.csdn.net/qq_41776136/article/details/145191851

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！