在本案例中，我们将展示如何用AWEL创建一个RAG(Retrieval Augmented Generation) 程序。

首先创建一个名称为first_rag_with_awel.py的python文件。

本案例提供的教程中，我们通过URL来获取知识，并存储到向量数据库当当中。

首先，需要创建环境并安装对应的依赖。 这里需要注意，dbgpt 包SDK的安装，不要安装在当前运行dbgpt服务的虚拟环境下。最好起个新的虚拟环境，如dbgpt_use

  1. conda create -n dbgpt_use python=3.10
  3. conda activate dbgpt_use
  5. pip install dbgpt

准备Embedding模型

将知识处理为向量，我们需要用到向量模型。 DB-GPT支持许多的向量模型，比如代理模型OpenAI Embedding，本地模型text2vec，当然也支持集群模式部署好之后，通过API的方式来使用。下面是几种方式的使用案例。

三种不同的向量模型使用方式

OpenAI Embedding API

  1. from dbgpt.rag.embedding import DefaultEmbeddingFactory
  2. embeddings = DefaultEmbeddingFactory.openai()

本地向量模型

  1. from dbgpt.rag.embedding import DefaultEmbeddingFactory
  2. embeddings = DefaultEmbeddingFactory.default("/data/models/text2vec-large-chinese")

集群部署下API调用

  1. from dbgpt.rag.embedding import DefaultEmbeddingFactory
  3. embeddings = DefaultEmbeddingFactory.remote(
  4.     api_url="http://localhost:8100/api/v1/embeddings",
  5.     api_key="{your_api_key}",
  6.     model_name="text2vec"
  7. )

在本案例中，我们使用本地向量模型来完成相关的操作。

获取知识并存储在向量数据库中

如下的DAG程序可以通过URL获取数据，并将其存储在向量数据库中。

  1. from dbgpt.rag.embedding import DefaultEmbeddingFactory
  3. embeddings = DefaultEmbeddingFactory.default("DB-GPT/models/text2vec-large-chinese")
  5. import asyncio
  6. import shutil
  7. from dbgpt.core.awel import DAG
  8. from dbgpt.rag import ChunkParameters
  10. from dbgpt.rag.knowledge import KnowledgeType
  11. from dbgpt.rag.operators import EmbeddingAssemblerOperator, KnowledgeOperator
  12. from dbgpt.storage.vector_store.chroma_store import ChromaVectorConfig
  13. from dbgpt.storage.vector_store.connector import VectorStoreConnector
  15. # Delete old vector store directory(/tmp/awel_rag_test_vector_store)
  16. shutil.rmtree("/tmp/awel_rag_test_vector_store", ignore_errors=True)
  18. # 2.init vector_connector
  19. vector_connector = VectorStoreConnector.from_default(
  20.     "Chroma",
  21.     vector_store_config=ChromaVectorConfig(
  22.         name="test_vstore",
  23.         persist_path="/tmp/awel_rag_test_vector_store",
  24.     ),
  25.     embedding_fn=embeddings
  26. )
  28. # 3. write awel assembler_task
  29. with DAG("load_knowledge_dag") as knowledge_dag:
  30.     # Load knowledge from URL
  31.     knowledge_task = KnowledgeOperator(knowledge_type=KnowledgeType.URL.name)
  32.     assembler_task = EmbeddingAssemblerOperator(
  33.         vector_store_connector=vector_connector,
  34.         chunk_parameters=ChunkParameters(chunk_strategy="CHUNK_BY_SIZE")
  35.     )
  36.     knowledge_task >> assembler_task
  38. chunks = asyncio.run(assembler_task.call("https://docs.dbgpt.site/docs/latest/awel/"))
  39. print(f"Chunk length: {chunks}")

从向量库中检索知识

你可以检索相关的知识，从向量数据库中。

  2. from dbgpt.core.awel import MapOperator
  3. from dbgpt.rag.operators import EmbeddingRetrieverOperator
  5. with DAG("retriever_dag") as retriever_dag:
  6.     retriever_task = EmbeddingRetrieverOperator(
  7.         top_k=3,
  8.         vector_store_connector=vector_connector,
  9.     )
  10.     content_task = MapOperator(lambda cks: "\n".join(c.content for c in cks))
  11.     retriever_task >> content_task
  13. chunks = asyncio.run(content_task.call("What is the AWEL?"))
  14. print(chunks)

准备LLM

可以有几种方式来调用LLM归纳总结，这里我们使用OpenAI的API

  1. pip install openai
  3. export OPENAI_API_KEY=sk-xx
  4. export OPENAI_API_BASE=https://xx:80/v1

使用代理模型Client来连接到OpenAI API

  1. from dbgpt.model.proxy import OpenAILLMClient
  2. llm_client = OpenAILLMClient()

创建RAG程序

最后我们创建一个rag程序，来进行智能问答。

  2. from dbgpt.core.awel import InputOperator, JoinOperator, InputSource
  3. from dbgpt.core.operators import PromptBuilderOperator, RequestBuilderOperator
  4. from dbgpt.model.operators import LLMOperator
  6. prompt = """Based on the known information below, provide users with professional and concise answers to their questions. 
  7. If the answer cannot be obtained from the provided content, please say: 
  8. "The information provided in the knowledge base is not sufficient to answer this question.". 
  9. It is forbidden to make up information randomly. When answering, it is best to summarize according to points 1.2.3.
  10.           known information: 
  11.           {context}
  12.           question:
  13.           {question}
  14. """
  16. with DAG("llm_rag_dag") as rag_dag:
  17.     input_task = InputOperator(input_source=InputSource.from_callable())
  18.     retriever_task = EmbeddingRetrieverOperator(
  19.         top_k=3,
  20.         vector_store_connector=vector_connector,
  21.     )
  22.     content_task = MapOperator(lambda cks: "\n".join(c.content for c in cks))
  24.     merge_task = JoinOperator(lambda context, question: {"context": context, "question": question})
  26.     prompt_task = PromptBuilderOperator(prompt)
  27.     # The model is gpt-3.5-turbo, you can replace it with other models.
  28.     req_build_task = RequestBuilderOperator(model="gpt-3.5-turbo")
  29.     llm_task = LLMOperator(llm_client=llm_client)
  30.     result_task = MapOperator(lambda r: r.text)
  32.     input_task >> retriever_task >> content_task >> merge_task
  33.     input_task >> merge_task
  35.     merge_task >> prompt_task >> req_build_task >> llm_task >> result_task
  37. print(asyncio.run(result_task.call("What is the AWEL?")))

输出如下： 

  1. AWEL stands for Agentic Workflow Expression Language. It is a set of intelligent agent workflow expression language specially designed for large model application development. AWEL provides functionality and flexibility for development of business logic for LLMs applications without the need to focus on cumbersome model and environment details. It adopts a layered API design, consisting of the operator layer, AgentFrame layer, and DSL layer. The operator layer includes basic operation atoms in the LLM application development process such as retrieval, vectorization, model interaction, and prompt processing.

恭喜你，到这里你已经完整的运行完了通过dbgpt sdk来开发一个RAG程序的所有过程。

完整代码

整个程序完整代码如下：

  1. import asyncio
  2. import shutil
  3. from dbgpt.core.awel import DAG, MapOperator, InputOperator, JoinOperator, InputSource
  4. from dbgpt.core.operators import PromptBuilderOperator, RequestBuilderOperator
  5. from dbgpt.rag import ChunkParameters
  6. from dbgpt.rag.knowledge import KnowledgeType
  7. from dbgpt.rag.operators import EmbeddingAssemblerOperator, KnowledgeOperator, EmbeddingRetrieverOperator
  8. from dbgpt.rag.embedding import DefaultEmbeddingFactory
  9. from dbgpt.storage.vector_store.chroma_store import ChromaVectorConfig
  10. from dbgpt.storage.vector_store.connector import VectorStoreConnector
  11. from dbgpt.model.operators import LLMOperator
  12. from dbgpt.model.proxy import OpenAILLMClient
  14. # Here we use the openai embedding model, if you want to use other models, you can 
  15. # replace it according to the previous example.
  16. embeddings = DefaultEmbeddingFactory.openai()
  17. # Here we use the openai LLM model, if you want to use other models, you can replace
  18. # it according to the previous example.
  19. llm_client = OpenAILLMClient()
  21. # Delete old vector store directory(/tmp/awel_rag_test_vector_store)
  22. shutil.rmtree("/tmp/awel_rag_test_vector_store", ignore_errors=True)
  24. vector_connector = VectorStoreConnector.from_default(
  25.     "Chroma",
  26.     vector_store_config=ChromaVectorConfig(
  27.         name="test_vstore",
  28.         persist_path="/tmp/awel_rag_test_vector_store",
  29.     ),
  30.     embedding_fn=embeddings
  31. )
  33. with DAG("load_knowledge_dag") as knowledge_dag:
  34.     # Load knowledge from URL
  35.     knowledge_task = KnowledgeOperator(knowledge_type=KnowledgeType.URL.name)
  36.     assembler_task = EmbeddingAssemblerOperator(
  37.         vector_store_connector=vector_connector,
  38.         chunk_parameters=ChunkParameters(chunk_strategy="CHUNK_BY_SIZE")
  39.     )
  40.     knowledge_task >> assembler_task
  42. chunks = asyncio.run(assembler_task.call("https://docs.dbgpt.site/docs/latest/awel/"))
  43. print(f"Chunk length: {len(chunks)}\n")
  46. prompt = """Based on the known information below, provide users with professional and concise answers to their questions. 
  47. If the answer cannot be obtained from the provided content, please say: 
  48. "The information provided in the knowledge base is not sufficient to answer this question.". 
  49. It is forbidden to make up information randomly. When answering, it is best to summarize according to points 1.2.3.
  50.           known information: 
  51.           {context}
  52.           question:
  53.           {question}
  54. """
  57. with DAG("llm_rag_dag") as rag_dag:
  58.     input_task = InputOperator(input_source=InputSource.from_callable())
  59.     retriever_task = EmbeddingRetrieverOperator(
  60.         top_k=3,
  61.         vector_store_connector=vector_connector,
  62.     )
  63.     content_task = MapOperator(lambda cks: "\n".join(c.content for c in cks))
  65.     merge_task = JoinOperator(lambda context, question: {"context": context, "question": question})
  67.     prompt_task = PromptBuilderOperator(prompt)
  68.     # The model is gpt-3.5-turbo, you can replace it with other models.
  69.     req_build_task = RequestBuilderOperator(model="gpt-3.5-turbo")
  70.     llm_task = LLMOperator(llm_client=llm_client)
  71.     result_task = MapOperator(lambda r: r.text)
  73.     input_task >> retriever_task >> content_task >> merge_task
  74.     input_task >> merge_task
  76.     merge_task >> prompt_task >> req_build_task >> llm_task >> result_task
  78. print(asyncio.run(result_task.call("What is the AWEL?")))

可视化DAGs

可以通过如下代码进行DAG可视化

  1. knowledge_dag.visualize_dag()
  2. rag_dag.visualize_dag()

如果在 Jupyter Notebook 中执行代码，可以在笔记本中看到 DAG。

  1. display(knowledge_dag.show())
  2. display(rag_dag.show())

Knowledge_dag可视化图如下:

AWEL RAG开发入门 - 图1

rag_dag可视化图如下AWEL RAG开发入门 - 图2

英文文档地址