<案例> 在 auto-coder.chat 中使用 byzerllm 从 0 实现RAG应用
输入“/”快速插入内容
<案例> 在 auto-coder.chat 中使用 byzerllm 从 0 实现RAG应用
<案例> 在 auto-coder.chat 中使用 byzerllm 从 0 实现RAG应用
飞书用户21063月13日修改
首先添加byzer-llm库，这样大模型就可以学习文档来学会如何使用 byzerllm:
下面的指令是运行在 auto-coder.chat 里的。
代码块
/lib /add byzer-llm
启动 byzerllm storage (向量加全文检索融合引擎):
注意该命令不要在 auto-coder.chat 的shell里执行，要在独立的terminal里执行。
代码块
byzerllm storage start --enable_emb
现在可以开始编程(我们是使用了 sonnet3.5)：
下面的指令是运行在 auto-coder.chat 里的。
代码块
/coding 根据大模型和存储的使用文档,我们要在 rag_exmaples 目录下实现一个 rag.py 文件,该文件指定一个目录,对目录里的md 文档按1000字最大长度做切割(按换行做分隔符,不断追加,直到超过1000时停止追加,得到一个chunk),然后对chunk 进行向量化,存储到存储引擎里.之后再根据query 召回并使用大模型回答.请注意使用 prompt 函数。
生成的代码（我大概人工调整了两到三行代码）：
代码块
import os
import byzerllm
from 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}")