让大模型使用 byzerllm 从 0 实现RAG应用

/add_files /Users/allwefantasy/projects/byzer-llm/docs/zh/003_byzerllm_大模型编程快速指南.md/add_files /Users/allwefantasy/projects/byzer-llm/docs/zh/004_byzerllm_storage_一个好用的AI存储引擎.md

byzerllm storage start --enable_emb

/coding 根据大模型和存储的使用文档,我们要在 rag_exmaples 目录下实现一个 rag.py 文件,该文件指定一个目录,对目录里的md 文档按1000字最大长度做切割(按换行做分隔符,不断追加,直到超过1000时停止追加,得到一个chunk),然后对chunk 进行向量化,存储到存储引擎里.之后再根据query 召回并使用大模型回答.请注意使用 prompt 函数。

import osimport byzerllmfrom byzerllm.apps.byzer_storage.simple_api import (    ByzerStorage,    DataType,    FieldOption,    SortOption,)
def chunk_text(text, max_length=1000):    chunks = []    current_chunk = []    current_length = 0
    for line in text.split("\n"):        if current_length + len(line) > max_length and current_chunk:            chunks.append("\n".join(current_chunk))            current_chunk = []            current_length = 0        current_chunk.append(line)        current_length += len(line)
    if current_chunk:        chunks.append("\n".join(current_chunk))
    return chunks
@byzerllm.prompt()def process_query(context: str, query: str) -> str:    """    Based on the following context, please answer the query:
    Context:    {{ context }}
    Query: {{ query }}
    Please provide a concise and accurate answer based on the given context.    """
class RAG:    def __init__(        self, llm_model="deepseek_chat", emb_model="emb", storage_name="byzerai_store"    ):        self.storage = ByzerStorage(            storage_name, "rag_database", "rag_table", emb_model=emb_model        )        self.llm = byzerllm.ByzerLLM()        self.llm.setup_default_model_name(llm_model)
        # Create schema if not exists                _ = (            self.storage.schema_builder()            .add_field("_id", DataType.STRING)            .add_field("content", DataType.STRING, [FieldOption.ANALYZE])            .add_field("raw_content", DataType.STRING, [FieldOption.NO_INDEX])            .add_array_field("vector", DataType.FLOAT)            .execute()        )
    def index_directory(self, directory):        for filename in os.listdir(directory):            if filename.endswith(".md"):                with open(os.path.join(directory, filename), "r") as file:                    content = file.read()                    chunks = chunk_text(content)
                    for i, chunk in enumerate(chunks):                        item = {                            "_id": f"{filename}_{i}",                            "content": chunk,                            "raw_content": chunk,                            "vector": chunk,                        }                        self.storage.write_builder().add_items(                            [item], vector_fields=["vector"], search_fields=["content"]                        ).execute()
        self.storage.commit()
    def query(self, query_text):        query = self.storage.query_builder()        query.set_vector_query(query_text, fields=["vector"])        results = query.execute()
        if results:            context = results[0]["raw_content"]            response = process_query.with_llm(self.llm).run(                context=context, query=query_text            )            return response        else:            return "Sorry, I couldn't find relevant information to answer your query."
if __name__ == "__main__":    rag = RAG()
    # Index documents in a directory    rag.index_directory("/Users/allwefantasy/projects/auto-coder/docs/zh")
    # Query example    query = "AutoCoder 如何配置搜索引擎?"    answer = rag.query(query)    print(f"Query: {query}")    print(f"Answer: {answer}")

Query: AutoCoder 如何配置搜索引擎?Answer: 要配置 AutoCoder 使用搜索引擎，您需要完成以下步骤：
1. **申请搜索API Token**：推荐使用 Bing 搜索引擎，访问 [Bing Web Search API](https://www.microsoft.com/en-us/bing/apis/bing-web-search-api) 获取 token。
2. **在 AutoCoder 参数中配置搜索引擎**：在配置文件中添加以下参数：   ```yml   search_engine: bing   search_engine_token: ENV {{BING_SEARCH_TOKEN}}   ```
这样，AutoCoder 就可以使用 Bing 搜索引擎来辅助代码编写和文档查找。