点击上方蓝字关注我们
前言
构建处理用户提问和文档信息的核心逻辑（图管道）。 搭建本地服务器，模拟 Speckly 的线上运行环境。 使用 Streamlit 和 Gradio 设计用户友好的交互界面。
步骤 1：导入 API 密钥
import os
from dotenv import load_dotenv, find_dotenv
load_dotenv(find_dotenv()) 重要提示：如果无法加载 API 密钥，请检查此行Getting the api keys from the .env file
os.environ['OPENAI_API_KEY'] = os.getenv('OPENAI_API_KEY')
os.environ['LANGCHAIN_API_KEY'] = os.getenv('LANGCHAIN_API_KEY')
Langsmith Tracing
os.environ['LANGCHAIN_TRACING_V2'] = os.getenv('LANGCHAIN_TRACING_V2')
os.environ['LANGCHAIN_ENDPOINT'] = os.getenv('LANGCHAIN_ENDPOINT')
os.environ['LANGCHAIN_PROJECT'] = os.getenv('LANGCHAIN_PROJECT')
Fire Crawl API
os.environ['FIRE_API_KEY']=os.getenv('FIRE_API_KEY')OPENAI_API_KEY=''
LANGCHAIN_API_KEY=''
LANGCHAIN_TRACING_V2='true'
LANGCHAIN_ENDPOINT='https://api.smith.langchain.com'
LANGCHAIN_PROJECT=''
步骤 2：加载文档
from typing import List
from langchain_community.document_loaders import FireCrawlLoader
from document import Document
class DocumentLoader:
def __init__(self, api_key: str):
self.api_key = api_key
def get_docs(self, url: str) -> List[Document]:
"""
使用 FireCrawlLoader 从指定的 URL 检索文档。
Args:
url (str): 要抓取文档的 URL。
Returns:
List[Document]: 包含检索到的内容的 Document 对象列表。
"""
loader = FireCrawlLoader(
api_key=self.api_key, url=url, mode="crawl"
)
raw_docs = loader.load()
docs = [Document(page_content=doc.page_content, metadata=doc.metadata) for doc in raw_docs]
return docsimport pickle
# 从本地文件加载已抓取并保存的文档
with open("crawled_docs/saved_docs.pkl", "rb") as f:
saved_docs = pickle.load(f)
步骤 3：创建向量存储和检索器
现在我们有了文档，我们希望将它们分成更小的部分，并将嵌入存储在一个开源向量存储中以供检索。我们将依赖 OpenAI 嵌入模型和 FAISS 向量存储。您还可以选择提供一个路径，以便将向量存储保存在本地。
from typing import List, Optional
from langchain_openai import OpenAIEmbeddings
from langchain.vectorstores import FAISS
from langchain.text_splitter import RecursiveCharacterTextSplitter
def create_vector_store(docs, store_path: Optional[str] = None) -> FAISS:
"""
从文档列表创建 FAISS 向量存储。
Args:
docs (List[Document]): 包含要存储的内容的 Document 对象列表。
store_path (Optional[str]): 用于在本地存储向量存储的路径。如果为 None，则不会存储向量存储。
Returns:
FAISS: 包含文档的 FAISS 向量存储。
"""
# 创建文本拆分器
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
)
texts = text_splitter.split_documents(docs)
# 嵌入对象
embedding_model = OpenAIEmbeddings()
# 创建 FAISS 向量存储
store = FAISS.from_documents(texts, embedding_model)
# 如果提供了路径，则将向量存储保存在本地
if store_path:
store.save_local(store_path)
return store
# 创建向量存储
store = create_vector_store(saved_docs)
# 创建检索器
retriever = store.as_retriever()
步骤 4：创建用于响应生成的检索链
create_generate_chain 函数来创建一个响应生成链。该函数首先使用一个 generate_template 来提供有关该过程的详细说明。这个模板有两个占位符：{context} 用于存储相关信息，{input} 用于存储问题。然后，使用 LangChain 中的 PromptTemplate 模块，它接受两个变量：template = generate_template 和 input_variables = ["context", "input"]。最后一步是使用 generate_prompt、大语言模型和 StrOutputParser() 创建 generate_chain。# generate_chain.py
from langchain.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
def create_generate_chain(llm):
"""
创建一个用于回答代码相关问题的生成链。
Args:
llm (LLM): 用于生成响应的语言模型。
Returns:
一个可调用函数，它接受上下文和问题作为输入，并返回字符串响应。
"""
generate_template = """
你是一个名为 Speckly 的乐于助人的代码助手。用户向你提供了一个与代码相关的问题，其内容由以下上下文部分表示（由 <context></context> 分隔）。
使用这些来回答最后的问题。
这些文件涉及 Speckle 开发者文档。你可以假设用户是土木工程师、建筑师或软件开发人员。
如果你不知道答案，就说你不知道。不要试图编造答案。
如果问题与上下文无关，请礼貌地回复说你只回答与上下文相关的问题。
提供尽可能详细的答案，并使用 Python（默认）生成代码，除非用户在问题中特别说明。
<context>
{context}
</context>
<question>
{input}
</question>
"""
generate_prompt = PromptTemplate(template=generate_template, input_variables=["context", "input"])
# 创建生成链
generate_chain = generate_prompt | llm | StrOutputParser()
return generate_chain
# 创建生成链
generate_chain = create_generate_chain(llm)
StrOutputParser() 用于从大语言模型获取字符串输出。否则，输出可能很复杂，例如 JSON 或结构化消息对象，这些对象无法直接用于进一步处理或向用户显示。没有 StrOutputParser() 的输出可能如下所示：{
"content": "This is the response from the LLM.",
"metadata": {
"confidence": 0.8,
"response_time": 0.5
}
}StrOutputParser()，输出如下所示：This is the response from the LLM.
步骤 5：创建评分器
检索评分器
检索评分器
create_retrieval_grader 函数，该函数接受一个带有新指令的提示模板 grade_prompt。该函数指示评分器在文档中查找与用户问题相关的关键字。如果存在此类关键字，则该文档被视为相关。评分器应该提供一个二进制分数，“yes” 或 “no”，表示该文档是否与问题相关，并以 JSON 格式提供结果，其中包含一个键“score”。def create_retrieval_grader(model):
"""
创建一个检索评分器，用于评估检索到的文档与用户问题的相关性。
Returns:
一个可调用函数，它接受文档和问题作为输入，并返回一个 JSON 对象，其中包含一个二进制分数，表示该文档是否与问题相关。
"""
grade_prompt = PromptTemplate(
template="""
<|begin_of_text|><|start_header_id|>system<|end_header_id|>
你是一个评分器，负责评估检索到的文档与用户问题的相关性。如果文档包含与用户问题相关的关键字，则将其评级为相关。它不需要是一个严格的测试。目标是过滤掉错误的检索结果。
给出一个二进制分数“yes”或“no”，表示该文档是否与问题相关。
以 JSON 格式提供二进制分数，其中包含一个键“score”，并且没有前言或解释。
<|eot_id|>
<|start_header_id|>user<|end_header_id|>
以下是检索到的文档： \n\n {document} \n\n
以下是用户问题： {input} \n
<|eot_id|>
<|start_header_id|>assistant<|end_header_id|>
""",
input_variables=["document", "input"],
)
# 创建检索器链
retriever_grader = grade_prompt | model | JsonOutputParser()
return retriever_grader以下是一个示例：
model = ... 在此处提供您的 llm
grader = create_retrieval_grader(model)
document = "France is a country in Europe. Paris is the capital of France."
question = "What is the capital of France?"
score = grader(document, question)
print(score) 输出： {"score": "yes"}
幻觉评分器
幻觉评分器
def create_hallucination_grader(self):
"""
创建一个幻觉评分器，用于评估答案是否基于/得到一组事实的支持。
Returns:
一个可调用函数，它接受生成（答案）和文档列表（事实）作为输入，并返回一个 JSON 对象，其中包含一个二进制分数，表示答案是否基于/得到事实的支持。
"""
hallucination_prompt = PromptTemplate(
template="""<|begin_of_text|><|start_header_id|>system<|end_header_id|>
你是一个评分器，负责评估答案是否基于/得到一组事实的支持。给出一个二进制分数“yes”或“no”，表示答案是否基于/得到一组事实的支持。以 JSON 格式提供二进制分数，其中包含一个键“score”，并且没有前言或解释。
<|eot_id|>
<|start_header_id|>user<|end_header_id|>
以下是事实：
\n ------- \n
{documents}
\n ------- \n
以下是答案： {generation}
<|eot_id|>
<|start_header_id|>assistant<|end_header_id|>""",
input_variables=["generation", "documents"],
)
hallucination_grader = hallucination_prompt | self.model | JsonOutputParser()
return hallucination_grader以下是一个示例：
from langchain_openai import ChatOpenAI
## LLM model
model = ChatOpenAI(model="gpt-4o", temperature=0)
## Grader
grader = create_hallucination_grader(model)
answer = "The capital of France is Paris."
facts = ["France is a country in Europe.", "Paris is the capital of France."]
score = grader(answer, facts)
print(score) # 输出： {"score": "yes"}
代码评估器
代码评估器
接下来，我们定义一个 create_code_evaluator 函数，该函数创建一个代码评估器，以评估生成的代码是否正确以及是否与给定问题相关。它使用 PromptTemplate 指示评估器提供一个带有二进制分数和反馈的 JSON 响应。评估器接受生成（代码）、问题和文档列表作为输入，并返回一个 JSON 对象，其中包含一个分数（表示代码是否正确和相关）以及对评估的简要说明。
def create_code_evaluator(self):
"""
创建一个代码评估器，用于评估生成的代码是否正确以及是否与给定问题相关。
Returns:
一个可调用函数，它接受生成（代码）、问题和文档列表作为输入，并返回一个带有二进制分数和反馈的 JSON 对象。
"""
eval_template = PromptTemplate(
template="""<|begin_of_text|><|start_header_id|>system<|end_header_id|> 你是一个代码评估器，负责评估生成的代码是否正确以及是否与给定问题相关。
提供一个带有以下键的 JSON 响应：
“score”：一个二进制分数“yes”或“no”，表示代码是否正确和相关。
“feedback”：对你的评估的简要说明，包括任何问题或需要改进的地方。
<|eot_id|><|start_header_id|>user<|end_header_id|>
以下是生成的代码：
\n ------- \n
{generation}
\n ------- \n
以下是问题： {input}
\n ------- \n
以下是相关文档： {documents}
<|eot_id|><|start_header_id|>assistant<|end_header_id|>""",
input_variables=["generation", "input", "documents"],
)
code_evaluator = eval_template | self.model | JsonOutputParser()
return code_evaluator
model = ... # 初始化一个语言模型
code_evaluator = create_code_evaluator(model)
code = "def greet(name): return f'Hello, {name}!'"
question = "Write a function to greet someone by name."
documents = ["A function should take a name as input and return a greeting message."]
result = code_evaluator(code, question, documents)
print(result) # 输出： {"score": "yes", "feedback": "The code is correct and relevant to the question."}
问题重写器
create_question_rewriter 函数，该函数构建一个重写器链，用于优化给定问题，以增强其清晰度和相关性。此函数返回一个可调用函数，该函数接受一个问题作为输入，并输出重写后的问题作为字符串。def create_question_rewriter(model):
"""
创建一个问题重写器链，用于重写给定问题以提高其清晰度和相关性。
Returns:
一个可调用函数，它接受一个问题作为输入，并返回重写后的问题作为字符串。
"""
re_write_prompt = hub.pull("efriis/self-rag-question-rewriter")
question_rewriter = re_write_prompt | self.model | StrOutputParser()
return question_rewriter
rewriter = create_question_rewriter()
original_question = "how to use speckle's python sdk?"
rewritten_question = rewriter(original_question)
print(rewritten_question) # 输出： "How to install speckle's python sdk?"现在我们已经定义了所有组件，我们可以创建一个名为 GraderUtils 的类来包含所有这些函数。然后，我们可以使用我们的 LLM 模型作为唯一必要的输入来初始化这个类的一个实例。
from langchain_openai import ChatOpenAI
class GraderUtils:
def __init__(self, model):
self.model = model
def create_retrieval_grader(self):
...
def create_hallucination_grader(self):
...
def create_code_evaluator(self):
...
def create_question_rewriter(self):
...
## LLM model
llm = ChatOpenAI(model="gpt-4o", temperature=0)
# 创建 GraderUtils 类的一个实例
grader = GraderUtils(llm)
# 获取检索评分器
retrieval_grader = grader.create_retrieval_grader()
# 获取幻觉评分器
hallucination_grader = grader.create_hallucination_grader()
# 获取代码评估器
code_evaluator = grader.create_code_evaluator()
# 获取问题重写器
question_rewriter = grader.create_question_rewriter()
想要了解更多信息，您可以参考 langchain-ai 仓库中的 RAG 笔记本。
步骤 6：创建图
定义图的状态
定义图的状态
GraphState 类来表示图的状态，该状态包含三个关键属性：input、generation 和 documents。其中，input 属性存储用户输入的问题，generation 属性存储大语言模型根据输入生成的答案，documents 属性存储相关文档列表。from typing_extensions import TypedDict
from typing import List
class GraphState(TypedDict):
"""
表示图的状态。
Attributes:
question: 问题
generation: LLM 生成
documents: 文档列表
"""
input: str
generation: str
documents: str #List[str]
GraphState 中定义的状态在整个图中全局可访问，并且这些属性是节点函数可以修改的唯一变量。节点
节点
GraphNodes 的类来包含所有节点函数。from document import Document
from utils.generate_chain import create_generate_chain
class GraphNodes:
def __init__(self, llm, retriever, retrieval_grader, hallucination_grader, code_evaluator, question_rewriter):
self.llm = llm
self.retriever = retriever
self.retrieval_grader = retrieval_grader
self.hallucination_grader = hallucination_grader
self.code_evaluator = code_evaluator
self.question_rewriter = question_rewriter
self.generate_chain = create_generate_chain(llm)
def retrieve(self, state):
"""
检索文档
Args:
state (dict): 当前图状态
Returns:
state (dict): 添加到状态的新键，文档，其中包含检索到的文档
"""
print("---RETRIEVE---")
question = state["input"]
# 检索
documents = self.retriever.invoke(question)
return {"documents": documents, "input": question}
def generate(self, state):
"""
生成答案
Args:
state (dict): 当前图状态
Returns:
state (dict): 添加到状态的新键，生成，其中包含 LLM 生成
"""
print("---GENERATE---")
question = state["input"]
documents = state["documents"]
# RAG 生成
generation = self.generate_chain.invoke({"context": documents, "input": question})
return {"documents": documents, "input": question, "generation": generation}
def grade_documents(self, state):
"""
确定检索到的文档是否与问题相关。
Args:
state (dict): 当前图状态
Returns:
state (dict): 使用仅过滤后的相关文档更新文档键
"""
print("---CHECK DOCUMENT RELEVANCE TO QUESTION---")
question = state["input"]
documents = state["documents"]
# 对每个文档进行评分
filtered_docs = []
for d in documents:
score = self.retrieval_grader.invoke({"input": question, "document": d.page_content})
grade = score["score"]
if grade == "yes":
print("---GRADE: DOCUMENT RELEVANT---")
filtered_docs.append(d)
else:
print("---GRADE: DOCUMENT IR-RELEVANT---")
continue
return {"documents": filtered_docs, "input": question}
def transform_query(self, state):
"""
转换查询以生成更好的问题。
Args:
state (dict): 当前图状态
Returns:
state (dict): 使用重新表述的问题更新问题键
"""
print("---TRANSFORM QUERY---")
question = state["input"]
documents = state["documents"]
# 重新编写问题
better_question = self.question_rewriter.invoke({"input": question})
return {"documents": documents, "input": better_question}
GraphNodes类定义了以下节点函数：
retrieve：根据输入问题检索文档，并将它们添加到图状态中。generate：使用输入问题和检索到的文档生成答案，并将生成添加到图状态中。grade_documents：根据检索到的文档与输入问题的相关性对其进行过滤，仅使用相关文档更新图状态。transform_query：重新表述输入问题以提高其清晰度和相关性，使用转换后的问题更新图状态。
边
边
class EdgeGraph:
def __init__(self, hallucination_grader, code_evaluator):
self.hallucination_grader = hallucination_grader
self.code_evaluator = code_evaluator
def decide_to_generate(self, state):
"""
确定是生成答案还是重新生成问题。
Args:
state (dict): 当前图状态
Returns:
str: 对要调用的下一个节点的二进制决策
"""
print("---ASSESS GRADED DOCUMENTS---")
question = state["input"]
filtered_documents = state["documents"]
if not filtered_documents:
# 所有文档都已过滤 check_relevance
# 我们将重新生成一个新查询
print("---DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY---")
return "transform_query" # "retrieve_from_community_page", "transform_query"
else:
# 我们有相关文档，因此生成答案
print("---DECISION: GENERATE---")
return "generate"
def grade_generation_v_documents_and_question(self, state):
"""
确定生成是否基于文档并回答问题。
Args:
state (dict): 当前图状态
Returns:
str: 对要调用的下一个节点的决策
"""
print("---CHECK HALLUCINATIONS---")
question = state["input"]
documents = state["documents"]
generation = state["generation"]
score = self.hallucination_grader.invoke({"documents": documents, "generation": generation})
grade = score["score"]
# 检查幻觉
if grade == "yes":
print("---DECISION: GENERATION IS GROUNDED IN DOCUMENTS---")
# 检查问答
print("---GRADE GENERATION vs QUESTION---")
score = self.code_evaluator.invoke({"input": question, "generation": generation, "documents": documents})
grade = score["score"]
if grade == "yes":
print("---DECISION: GENERATION ADDRESSES QUESTION---")
return "useful"
else:
print("---DECISION: GENERATION DOES NOT ADDRESS QUESTION---")
return "not useful"
else:
print("---DECISION: GENERATIONS ARE HALLUCINATED, RE-TRY---")
return "not supported"EdgeGraph 类定义了以下边函数：decide_to_generate：根据过滤后的文档与输入问题的相关性，决定是生成答案还是重新生成问题。grade_generation_v_documents_and_question：根据生成的答案是否基于文档及其回答问题的能力来评估生成的答案。
构建图
构建图
# 初始化图
workflow = StateGraph(GraphState)
# 创建 GraphNodes 类的一个实例
graph_nodes = GraphNodes(llm, retriever, retrieval_grader, hallucination_grader, code_evaluator, question_rewriter)
# 创建 EdgeGraph 类的一个实例
edge_graph = EdgeGraph(hallucination_grader, code_evaluator)
# 定义节点
workflow.add_node("retrieve", graph_nodes.retrieve) # 检索文档
workflow.add_node("grade_documents", graph_nodes.grade_documents) # 对文档进行评分
workflow.add_node("generate", graph_nodes.generate) # 生成答案
workflow.add_node("transform_query", graph_nodes.transform_query) # 转换查询
# 构建图
workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
"grade_documents",
edge_graph.decide_to_generate,
{
"transform_query": "transform_query", # "transform_query": "transform_query",
"generate": "generate",
},
)
workflow.add_edge("transform_query", "retrieve")
workflow.add_conditional_edges(
"generate",
edge_graph.grade_generation_v_documents_and_question,
{
"not supported": "generate",
"useful": END,
"not useful": "transform_query", # "transform_query"
},
)
# 编译
chain = workflow.compile()
StateGraph 类初始化图。然后，我们创建 GraphNodes 和 EdgeGraph 类的实例。接下来，我们添加已经定义函数的节点：retrieve：根据输入问题检索相关文档。grade_documents：根据检索到的文档与问题的相关性对其进行过滤。generate：根据过滤后的文档生成答案。transform_query：转换输入问题以提高其清晰度和相关性。
retrieve 节点。retrieve 节点和 grade_documents 节点之间有一条普通边连接。在 grade_documents 节点之后，工作流到达一个条件边。此时，会调用 edge_graph.decide_to_generate 函数来决定工作流的下一步。该函数评估已评分的文档，并决定是转换查询还是生成答案。如果函数返回 transform_query，则工作流将移动到 transform_query 节点，该节点转换输入问题以提高其清晰度和相关性。如果函数返回 generate，则工作流将移动到 generate 节点，该节点根据过滤后的文档生成答案。transform_query 和 retrieve 之间也有一条普通边连接。这是因为在转换查询之后，工作流会移回 retrieve 节点，以根据转换后的查询检索新文档。edge_graph.grade_generation_v_documents_and_question 函数，根据生成的答案是否基于文档及其回答问题的能力来评估生成的答案。如果函数返回 not supported，则工作流将移回 generate 节点以重新生成答案。此步骤对于确保工作流生成受文档支持的答案是必需的。如果函数返回 useful，则工作流将结束，表示已生成有用的答案。如果函数返回 not useful，则工作流将移动到 transform_query 节点以再次转换查询。
步骤 7：使用 FastAPI 启动服务器
app = FastAPI(
title="Speckle服务器",
version="1.0",
description="用于回答有关Speckle Developer Docs的问题的API服务器"
)
@app.get("/")
async def redirect_root_to_docs():
return RedirectResponse("/docs")BaseModel 定义两个模型：Input 和 Output。这些模型将用于定义 API 的输入和输出数据的结构。class Input(BaseModel):
input: str
class Output(BaseModel):
output: dict
add_routes 函数向应用程序添加路由。add_routes(
app,
chain.with_types(input_type=Input, output_type=Output),
path="/speckle_chat",
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="localhost", port=8000)现在，我们已经创建了一个 FastAPI 应用程序，并启动了一个可以从 http://localhost:8000 访问的服务器。
步骤 8：使用 Streamlit/Gradio 创建客户端
使用 Streamlit 创建 UI
使用 Streamlit 创建 UI
import streamlit as st
from langserve import RemoteRunnable
from pprint import pprint
st.title('Welcome to Speckle Server')
input_text = st.text_input('ask speckle related question here')
if input_text:
with st.spinner("Processing..."):
try:
app = RemoteRunnable("http://localhost:8000/speckle_chat/")
for output in app.stream({"input": input_text}):
for key, value in output.items():
# 节点
pprint(f"Node '{key}':")
# 可选：在每个节点打印完整状态
# pprint.pprint(value["keys"], indent=2, width=80, depth=None)
pprint("\n---\n")
output = value['generation']
st.write(output)
except Exception as e:
st.error(f"Error: {e}")
langserve 中的 RemoteRunnable 模块和服务器 URL 连接到服务器。它使用 stream 命令流式传输来自大语言模型的响应，同时打印图工作流中触发的节点。最后，我们从存储在值字典中的 generation 键中检索最终输出。如果在处理过程中出现错误，它将显示错误消息。
使用 Gradio 创建 UI
使用 Gradio 创建 UI
def get_response(input_text):
app = RemoteRunnable("http://localhost:8000/speckle_chat/")
for output in app.stream({"input": input_text}):
for key, value in output.items():
# 节点
pprint(f"Node '{key}':")
# 可选：在每个节点打印完整状态
# pprint.pprint(value["keys"], indent=2, width=80, depth=None)
pprint("\n---\n")
output = value['generation']
return output
import gradio as gr
from langserve import RemoteRunnable
from pprint import pprint
# 在 Gradio 中创建 UI
iface = gr.Interface(fn=get_response,
inputs=gr.Textbox(
value="Enter your question"),
outputs="textbox",
title="Q&A over Speckle's developer docs",
description="Ask a question about Speckle's developer docs and get an answer from the code assistant. This assistant looks up relevant documents and answers your code-related question.",
examples=[["How do I install Speckle's python sdk?"],
["How to commit and retrieve an object from Speckle?"],
],
theme=gr.themes.Soft(),
allow_flagging="never",)
iface.launch(share=True) # 将 share 设置为 True 以获取公共 URL
在 launch 函数中设置 share=True 可以获取公共 URL。
总结
今天，我们探讨了如何为包含高级 RAG（检索增强生成）概念的图工作流开发服务器-客户端架构。我们创建了一个服务器组件，该组件涵盖了一个全面的管道，包括对检索到的文档进行评分、对响应进行评分、检查幻觉和查询重写。为了与此本地服务器交互，我们创建了两个客户端应用程序，一个使用 Streamlit，另一个使用 Gradio。这两个 UI 都为用户提供了一个友好的界面，让他们可以输入查询并实时接收服务器的响应。这是一个端到端的项目，允许开发人员在将应用程序部署到生产环境之前构建应用程序并在本地对其进行测试。
