[Datawheel]用Llama-index创建Agent、数据库对话Agent和RAG接入Agent
1.Llama-index创建Agent1.0 配景知识
[*]什么是Llama-index?
LlamaIndex（原名GPT Index）是一个专为大语言模型（LLMs）计划的数据框架，旨在资助用户将外部数据与LLMs联合，实现更高效的数据检索和知识增强生成（RAG）。它通过构建索引和提供查询接口，使LLMs能够访问和使用私有或特定领域的数据，从而提升模型的正确性和实用性。
[*]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）查询阶段：从知识库中检索相关上下文，并将其与用户查询联合，生成增强后的相应。
[*]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.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:
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.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测试效果
https://i-blog.csdnimg.cn/direct/4b958eee36d54d03b217a24a0a6d71e6.png
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(, llm=llm, verbose=True)
response = agent.chat("20+（2*4）等于多少？使用工具计算每一步")
print(response)
if __name__ == "__main__":
main()
1.6 效果展示
https://i-blog.csdnimg.cn/direct/e0ff556337fa40e499b56c862c65fee2.png
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(, llm=llm, verbose=True)
response = agent.chat("纽约天气怎么样?")
这里写了一个伪函数，固然也可以去调用天气查询的api。
https://i-blog.csdnimg.cn/direct/8d6cb152ed614e1fa48e09656b126028.png
https://i-blog.csdnimg.cn/direct/b3a1b178e54645ebb946c547393ec21a.png
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.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:
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.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(, verbose=True)
# 通过agent给出指令
response = agent.chat("请从数据库表中获取`专利部`和`商标部`的人数，并将这两个部门的人数相加！")
2.4 效果
https://i-blog.csdnimg.cn/direct/3d0be4a212354c248448dc05c034776a.png
2.5 明白
这里对于初学者比较难明白embedding模型起到了什么作用，目前推测是在NLSQLTableQueryEngine类中用于为文本生成向量表现，支持语义明白和检索任务。别的，QueryEngineTool 是 LlamaIndex 框架中的一个工具类，它的作用是将一个 查询引擎（Query Engine）封装成一个 工具（Tool），以便在更高条理的组件（如 ReActAgent 或其他署理）中使用。通过这种方式，查询引擎可以被集成到更复杂的任务流程中，例如多步骤推理、任务分解或工具调用。
这里另有个问题：为什么这里一定要用到embedding模型来明白自然语言llm做不到吗？
虽然 LLM 能够直接明白自然语言，但在某些场景下，Embedding 模型仍然是必要的。以下是 Embedding 模型的重要作用：
[*]向量化表现
[*]Embedding 模型将文本转换为固定长度的向量表现，这些向量可以用于盘算文本之间的相似度。
[*]例如，在文档检索、语义搜索等任务中，Embedding 模型可以快速找到与查询文本最相关的文档。
[*]高效检索
[*]在大规模数据集中，直接使用 LLM 进行检索可能会非常低效。Embedding 模型可以将文本转换为向量后，使用向量检索技能（如余弦相似度、近似近来邻搜索）快速找到相关效果。
[*]例如，如果你有一个包含数百万条记录的数据库，使用 Embedding 模型进行向量检索会比直接使用 LLM 更高效。
[*]语义明白
[*]Embedding 模型能够捕捉文本的语义信息，使得语义相似的文本在向量空间中距离较近。
[*]例如，在问答系统中，Embedding 模型可以用于找到与用户问题最相关的答案。
[*]任务分离
[*]在某些任务中，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 =
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_synthesizersimport 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="")
https://i-blog.csdnimg.cn/direct/66614bccfaa14d26baabe14c0a666bf6.png## 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)
输出：
https://i-blog.csdnimg.cn/direct/166af1d046b24530a10bb6e15b4063c4.png
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(, llm=llm, verbose=True)
测试一下
# 测试用例
query = "阿里巴巴2024年的ESG报告主要讲了哪些内容？"
response = agent.chat(f"请帮我搜索以下内容：{query}")
print(response)
效果
https://i-blog.csdnimg.cn/direct/21a07b7e9a4c455582ef134b10428681.png
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]