当前位置：首页 > article >正文

Speckly：基于Speckle文档的RAG智能问答机器人

article 2024/12/28 16:52:19

前言

Speckly 是一个基于 检索增强生成 (RAG) 技术的智能问答机器人，它能像一位经验丰富的工程师，理解你的问题，并从 Speckle 文档中精准地找到答案。更厉害的是，它甚至可以帮你生成代码片段！🚀

本文将详细介绍 Speckly 的完整开发流程，涵盖从创建图管道到搭建服务器，再到设计用户界面的所有环节，最终实现一个可交互的智能问答系统。

您将学习如何：

构建处理用户提问和文档信息的核心逻辑（图管道）。
搭建本地服务器，模拟 Speckly 的线上运行环境。
使用 Streamlit 和 Gradio 设计用户友好的交互界面。

通过学习本项目，您将掌握在部署模型到生产环境之前进行本地测试的方法，并了解如何构建简洁易用的用户界面。

步骤 1：导入 API 密钥

首先，我们从 .env 文件中导入 API 密钥。另外，我们还可以使用 LangSmith 设置跟踪。

import osfrom dotenv import load_dotenv, find_dotenvload_dotenv(find_dotenv()) 重要提示：如果无法加载 API 密钥，请检查此行Getting the api keys from the .env fileos.environ['OPENAI_API_KEY'] = os.getenv('OPENAI_API_KEY')os.environ['LANGCHAIN_API_KEY'] = os.getenv('LANGCHAIN_API_KEY')Langsmith Tracingos.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 APIos.environ['FIRE_API_KEY']=os.getenv('FIRE_API_KEY')

以下是一个示例 .env 文件。请获取您的 API 密钥（如果您没有），并将它们粘贴在字符串之间。我在第一篇博文中详细描述了这一点。

OPENAI_API_KEY=''
LANGCHAIN_API_KEY=''
LANGCHAIN_TRACING_V2='true'
LANGCHAIN_ENDPOINT='https://api.smith.langchain.com'
LANGCHAIN_PROJECT=''

步骤 2：加载文档

我们将使用 Mendable.ai 创建的名为 FireCrawl 的产品，它可以将网站转换为对大语言模型友好的文档。这正是我们想要的。我们将抓取 Speckle 的开发者文档，并将所有页面和子页面转换为文档列表。您需要一个 API 密钥才能在加载器函数中使用。

我创建了 DocumentLoader 类，它将 API 密钥作为字符串输入，并使用 get_docs 函数，该函数将 URL 作为输入，并提供一个文档列表（包括元数据）作为输出。

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 docs

就我而言，我已经抓取了文档，并将文档保存在本地，这样我就不会重复这个过程并浪费我的积分了。第一次使用时，您可以使用 get_docs 函数；之后，您可以直接加载文档。

import 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"}

幻觉评分器

接下来，我们定义一个幻觉评分器，用于评估从大语言模型获得的答案是否基于一组事实。该评分器提供一个二进制分数（“yes”或“no”），表示答案是否合理。提示模板将包括事实 ({documents}) 和答案 ({generation}) 的占位符，这些占位符将在使用提示时填充。

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：创建图

现在我们已经拥有了所有组件，接下来我们将使用 LangGraph 创建图。

定义图的状态

首先，我们定义一个 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 中定义的状态在整个图中全局可访问，并且这些属性是节点函数可以修改的唯一变量。

节点

接下来，我们定义节点。节点是 Python 函数，它们接收图的状态，执行一些操作，并修改状态变量。我们定义一个名为 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}

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 启动服务器

现在，我们将使用 FastAPI 启动服务器。

首先，我们创建一个 FastAPI 应用程序。

app = FastAPI(
    title="Speckle服务器",
    version="1.0",
    description="用于回答有关Speckle Developer Docs的问题的API服务器"
)

接下来，我们为根 URL (/) 定义一个路由，该路由重定向到文档 URL (/docs)。

@app.get("/")
async def redirect_root_to_docs():
    return RedirectResponse("/docs")

然后，我们使用 Pydantic 的 BaseModel 定义两个模型：Input 和 Ou``tput。这些模型将用于定义 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",
)

最后，我们使用 Uvicorn 运行服务器。

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="localhost", port=8000)

现在，我们已经创建了一个 FastAPI 应用程序，并启动了一个可以从 http://localhost:8000 访问的服务器。

步骤 8：使用 Streamlit/Gradio 创建客户端

现在，我们将创建客户端来与服务器进行交互。

使用 Streamlit 创建 UI

首先，我们使用 Python 中的 Streamlit 库创建一个客户端。

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}")

我们首先设置 Streamlit 应用程序，包括一个标题和一个文本输入字段，供用户输入问题。当用户输入任何文本时，应用程序会显示一个微调器，表示正在处理输入。然后，应用程序使用 langserve 中的 RemoteRunnable 模块和服务器 URL 连接到服务器。它使用 stream 命令流式传输来自大语言模型的响应，同时打印图工作流中触发的节点。最后，我们从存储在值字典中的 generation 键中检索最终输出。如果在处理过程中出现错误，它将显示错误消息。

在这里插入图片描述

使用 Gradio 创建 UI

我们还可以使用 Gradio 来创建客户端 UI。Gradio 是一个开源 Python 库，用于为机器学习模型、API 和任意 Python 函数创建交互式基于 Web 的用户界面。

首先，我们创建一个函数，该函数将允许从大语言模型中获取最终响应。

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

然后，我们创建一个简单的 Gradio UI。

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 都为用户提供了一个友好的界面，让他们可以输入查询并实时接收服务器的响应。这是一个端到端的项目，允许开发人员在将应用程序部署到生产环境之前构建应用程序并在本地对其进行测试。