llama-index做一个简单查询工作流（rag）

目次
  一、导入本地大模型、embedding
  二、workflow
  三、导入数据
  四、工作流可视化
  五、输入工具，输出结果
  一、导入本地大模型、embedding

 代码如下（示例）：导入的千问的0.5b大模型
或者使用api接口

1. from llama_index.llms.zhipuai import ZhipuAI
2. llms = ZhipuAI(model="glm-4")#api要设置

复制代码

1. from llama_index.embeddings.huggingface import HuggingFaceEmbedding
2. from llama_index.llms.huggingface import HuggingFaceLLM
3. from llama_index.core import Settings
4. llm=HuggingFaceLLM(
5.     model_name="LLM_model/Qwen2.5-0.5B-Instruct",
6.     tokenizer_name="LLM_model/Qwen2.5-0.5B-Instruct",
7.     context_window=30000,
8.     max_new_tokens=2000,
9.     generate_kwargs={"temperature": 0.7, "top_k": 50, "top_p": 0.95},
10.     device_map="auto")

复制代码

  # 设置embedding模型，后面进行向量库储存的时候需要用到
# 用魔搭下载，更快，改成本身的路径
  modelscope download --model AI-ModelScope/bge-large-zh-v1.5
  --local_dir LLM/LLM_model/embeddding-bge-large-zh-v1.5

1. from llama_index.embeddings.huggingface import HuggingFaceEmbedding
2. from llama_index.core import Settings
3. Settings.embed_model =  HuggingFaceEmbedding(
4.     model_name="LLM_model/embeddding-bge-large-zh-v1.5",trust_remote_code=True)

复制代码

二、workflow

工作流应该是这样：
制定计划、进行查询、汇聚答案，决定是否再继续查询
故我们可以界说两个变乱

1. #事件
2. from llama_index.core.workflow import Event
3. class QueryPlanItemResult(Event):
4.     """The result of a query plan item"""
6.     query: str
7.     result: str
9. class ExecutedPlanEvent(Event):
10.     """The result of a query plan"""
12.     result: str

复制代码

同时因为我们要调用工具，所以我们要规范模型的输出，这个会输入到结构化输出的函数内里，规范大模型的输出，原理应该雷同这里https://blog.csdn.net/2303_77229879/article/details/143441989?spm=1001.2014.3001.5501

1. from pydantic import BaseModel, Field
2. from llama_index.core.workflow import Event
3. class QueryPlanItem(Event):
4.     """A single step in an execution plan for a RAG system."""
5.     name: str = Field(description="The name of the tool to use.")
6.     query: str = Field(description="A natural language search query for a RAG system.")
8. class QueryPlan(BaseModel):
9.     """A plan for a RAG system. After running the plan, we should have either enough information to answer the user's original query, or enough information to form a new query plan."""
10.     items: list[QueryPlanItem] = Field(
11.         description="A list of the QueryPlanItem objects in the plan."
12.     )
13. '''通过astructured传入模型即可得到结构化输出
14. query_plan = await llm.astructured_predict(
15.                 QueryPlan,
16.                 self.planning_prompt,
17.                 context=context_str,
18.                 query=query,
19.             )'''

复制代码

现在我们来界说工作流，一共有如下几个函数

• 制定计划，或者竣事工作流
  
• 执行函数
  
• 汇聚结果
  

1. class QueryPlanningWorkflow(Workflow):
2.     #让模型进行规划根据上下文内容和用户输入
3.     planning_prompt = PromptTemplate(
4.         "Think step by step. Given an initial query, as well as information about the indexes you can query, return a plan for a RAG system.\n"
5.         "The plan should be a list of QueryPlanItem objects, where each object contains a query.\n"
6.         "The result of executing an entire plan should provide a result that is a substantial answer to the initial query, "
7.         "or enough information to form a new query plan.\n"
8.         "Sources you can query: {context}\n"
9.         "Initial query: {query}\n"
10.         "Plan:"
11.     )
12.     #根据目前结果，结束还是举行执行任务
13.     decision_prompt = PromptTemplate(
14.         "Given the following information, return a final response that satisfies the original query, or return 'PLAN' if you need to continue planning.\n"
15.         "Original query: {query}\n"
16.         "Current results: {results}\n"
17.     )
19.     @step
20.     async def planning_step(
21.         self, ctx: Context, ev: StartEvent | ExecutedPlanEvent
22.     ) -> QueryPlanItem | StopEvent:
23.         if isinstance(ev, StartEvent):
24.             # stratevent的时候才会进行Initially, we need to plan
25.             query = ev.get("query")
26.             tools = ev.get("tools")
27.             await ctx.set("tools", {t.metadata.name: t for t in tools})
28.             await ctx.set("original_query", query)
29.             # 将工具的信息作为上下文导入，使得模型看见工具
30.             context_str = "\n".join(
31.                 [
32.                     f"{i+1}. {tool.metadata.name}: {tool.metadata.description}"
33.                     for i, tool in enumerate(tools)
34.                 ]
35.             )
36.             await ctx.set("context", context_str)
37.             #planning_prompt导入，要求返回列表，列表里面是使用的工具和查询的自然语言
38.             #这里有需要也可以吧大模型设置成类变量，我这里直接改成上面定义的全局变量了
39.             query_plan = await llm.astructured_predict(
40.                 QueryPlan,
41.                 self.planning_prompt,
42.                 context=context_str,
43.                 query=query,
44.             )
45.             ctx.write_event_to_stream(
46.                 Event(msg=f"Planning step: {query_plan}")
47.             )
48.             num_items = len(query_plan.items)
49.             await ctx.set("num_items", num_items)
50.             #返回事件
51.             for item in query_plan.items:
52.                 ctx.send_event(item)#将事件发送到工作流中的特定步骤。如果 step 为 None，则事件将发送给所有接收者，由它们决定是否丢弃不需要的事件。
53.         else:
54.             #ExecutedPlanEvent时会执行
55.             query = await ctx.get("original_query")
56.             current_results_str = ev.result
57.             ##根据目前结果，结束还是举行执行任务
58.             decision = await llm.apredict(
59.                 self.decision_prompt,
60.                 query=query,
61.                 results=current_results_str,
62.             )
64.             # 简单的匹配，如果里面有plan就继续规划，否则就结束，疑惑：如果模型输出不需要plan，怎么结束？
65.             if "PLAN" in decision:
66.                 context_str = await ctx.get("context")
67.                 query_plan = await llm.astructured_predict(
68.                     QueryPlan,
69.                     self.planning_prompt,
70.                     context=context_str,
71.                     query=query,
72.                 )
73.                 ctx.write_event_to_stream(
74.                     Event(msg=f"Re-Planning step: {query_plan}")
75.                 )
76.                 num_items = len(query_plan.items)
77.                 await ctx.set("num_items", num_items)
78.                 for item in query_plan.items:
79.                     ctx.send_event(item)
80.             else:
81.                 #如果没有plan就结束
82.                 return StopEvent(result=decision)
84.     @step(num_workers=4)
85.     async def execute_item(
86.         self, ctx: Context, ev: QueryPlanItem
87.     ) -> QueryPlanItemResult:
88.         tools = await ctx.get("tools")
89.         tool = tools[ev.name]
91.         ctx.write_event_to_stream(
92.             Event(
93.                 msg=f"Querying tool {tool.metadata.name} with query: {ev.query}"
94.             )
95.         )
96.         result = await tool.acall(ev.query)#传入的工具定义的方法，直接把语句放到查询引擎中，然后返回结果
97.         ctx.write_event_to_stream(
98.             Event(msg=f"Tool {tool.metadata.name} returned: {result}")
99.         )
100.         return QueryPlanItemResult(query=ev.query, result=str(result))
102.     @step
103.     async def aggregate_results(
104.         self, ctx: Context, ev: QueryPlanItemResult
105.     ) -> ExecutedPlanEvent:
106.         num_items = await ctx.get("num_items")
107.         # 收集全部的事件才会结束，否则会一直等待
108.         results = ctx.collect_events(ev, [QueryPlanItemResult] * num_items)
109.         if results is None:
110.             return
111.         #汇总结果
112.         aggregated_result = "\n------\n".join(
113.             [
114.                 f"{i+1}. {result.query}: {result.result}"
115.                 for i, result in enumerate(results)
116.             ]
117.         )
118.         return ExecutedPlanEvent(result=aggregated_result)

复制代码

使用glm-4可能出现的问题，我调用api接口一直返回的格式不精确，可能是 planning_prompt的的第三行只说返回一个query，这里改成a query and a name应该就能返回精确的格式了
按理来说prompt的问题的话qwen也会有同样的错误，但是虽然也有发生，重复两次也总是能执行
三、导入数据

使用llamaparse解析PDF文件，有一个文件夹，内里都是pdf，这里可以任意放几篇论文进去，然后可以通过循环界说不同的引擎，这里我只放了一个，所以名称就没修改，只有一个

1. # 在这里，我们用于加载和解析文档，调用LlamaParseapi解析文件
2. from llama_parse import LlamaParse
3. api_key=''#去llama-cloud注册有免费额度的
4. parser = LlamaParse(api_key=api_key,fast_mode=True)
6. from llama_index.core import (
7.     VectorStoreIndex,
8.     StorageContext,
9.     load_index_from_storage,
10. )
11. from llama_index.core.tools import QueryEngineTool
13. folder = "./data_pdf/"
14. files = os.listdir(folder)
16. query_engine_tools = []
17. for file in files:
18.     #取名字第一个字符
19.     number = file.split(" - ")[0]#根据论文名字的分隔符确定，防止文章太长
20.     index_persist_path = f"./storage/paper-{number}/"
21.     #创建储存路径，如果存在就加载，不存在就创建
22.     if os.path.exists(index_persist_path):
23.         storage_context = StorageContext.from_defaults(
24.             persist_dir=index_persist_path
25.         )
26.         index = load_index_from_storage(storage_context)
27.     else:
28.         documents = await parser.aload_data(folder + file)
29.         index = VectorStoreIndex.from_documents(documents)
30.         index.storage_context.persist(index_persist_path)
31.     #这里需要输入llm，不然会自动调用openai的大模型
32.     engine = index.as_query_engine(llm=llm)
33.     query_engine_tools.append(
34.         QueryEngineTool.from_defaults(
35.             engine,
36.             name=f"TimeSeriesAnalysis",
37.             description=f"This is a paper on time series analysis, which includes information on time series forecasting and attention mechanisms."
38.         )
39.     )

复制代码

如果tool名称太长或者，描述不精确都有可能会报错，返回格式不精确这种，好比pydanitc的报错或者ctx获取不到items(basemodel那个子类界说的名称）
四、工作流可视化

1. from llama_index.utils.workflow import draw_all_possible_flows
2. draw_all_possible_flows(
3.     QueryPlanningWorkflow, filename="twwww.html"
4. )

复制代码

五、输入工具，输出结果

这里是jupterbook内里操作，如果要在py内里运行需要界说一个main函数

1. #测试工作流
2. workflow = QueryPlanningWorkflow(verbose=False, timeout=120)
3. # run the workflow
4. handler = workflow.run(
5.     query="什么是时间序列预测",
6.     tools=query_engine_tools,
7. )
8. #流式处理
9. async for event in handler.stream_events():
10.     if hasattr(event, "msg"):
11.         print(event.msg)
12. result = await handler

复制代码

1.qwen0.5b输出
进行了多次尝试，输出结果都不是很抱负哈，prompt内里又是英语又是中文的，有点为难这个0.5b的大模型了，第一次答的还算沾点边，第二次不知道为啥干翻译去了，然后这个任务的名字他也常常堕落，返回错误的格式，就得不停重试和调解prompt

2.glm-4输出结果
glm这里输出就显着稳固多了，照旧需要调解prompt，而且这里没有加入重试的尝试，一般来说加上可以减少报错，这个prompt也可以调解调解，改成官方做functioncall的情势。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。