1.Llama-index创建Agent
1.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[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 模型的重要作用:
- 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 = [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)
复制代码 效果
免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!更多信息从访问主页:qidao123.com:ToB企服之家,中国第一个企服评测及商务社交产业平台。 |