LangGraph FastAPI 和 StreamlitGradioLLM 驱动应用的完美三重奏

在我开始之前，快速插入一个🔌 :-)这是关于 LangChain 初学者系列的第 5 篇博客。我通过这些帖子记录我的 GenAI 学习（顺便提一下，我的背景是土木工程和机器学习）。我承诺会让内容简单易懂，适合像我一样的初学者，并随着每一篇新帖子逐步深入。通过这个系列，你将能够一步一步构建自己的 GenAI 应用。祝学习愉快！🤗

在我开始之前，这里有一个背景故事。这是一个来自名为 Beyond the Speckleverse 的黑客马拉松的项目，我本打算参加。这个活动由 Speckle 组织，该公司为软件开发人员和工程师/建筑师提供一个平台，以便他们能够协作并为 AEC（建筑、工程和施工）行业构建工具。我直到最后一晚才开始，因为我无法想出一个可以与 Speckle 平台集成的酷项目。当我试图学习开发者文档时，我意识到……

如果我们能创建一个 AI 代码助手，能够浏览文档并根据用户的查询检索答案，那将是多么酷啊。这个项目还可以进一步发展成一个调试助手，通过查看社区论坛来实现。

于是我试图在一夜之间完成整个设置 😄！当然我没能完成。我低估了手头的任务，过高估计了我的编码能力。但是，在接下来的两天里，我成功构建了一个生产就绪的服务器，并建立了一个简单的用户界面来与文档进行聊天。还有更多的工作要做，但我认为这是我第五篇博客的完美内容。让我们开始吧 🎬

构建服务器-客户端交互

TL;DR

这是一篇稍长的文章，因为它包含一个端到端的项目；创建图形管道、启动服务器以及创建与服务器交互的客户端。通过这个项目，您将学习如何在将模型部署到生产环境之前在本地测试您的项目。我们还将看到如何使用 Streamlit 和 Gradio 为我们的代码助手创建一个简单的用户界面。那么，废话不多说，让我们开始吧！

目录

1. 导入 API 密钥
2. 加载文档
3. 创建向量存储和检索器
4. 创建响应生成的检索链
5. 创建评分器
6. 创建图形
7. 使用 FastAPI 启动服务器
8. 创建带有 Streamlit/Gradio UI 的客户端

第一步：导入 API 密钥

让我们开始从 .env 文件中导入 API 密钥。可选地，我们还可以使用 Langsmith 设置追踪。

import os
from dotenv import load_dotenv, find_dotenv
load_dotenv(find_dotenv()) # important line if cannot load api key

## 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')

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

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

第2步：加载文档

我们将使用一个名为 FireCrawl 的产品，它由 Mendable.ai 创建，可以将网站转换为适合 LLM 的文档。这正是我们所需要的。我们将爬取 Speckle 的开发者文档，并将所有页面和子页面转换为文档列表。您需要一个 API 密钥来在加载函数中使用。仅供参考：您将获得 500 个免费积分（顺便说一下，我已经超过了），所以请明智地使用。

我创建了 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]:
        """
        Retrieves documents from the specified URL using the FireCrawlLoader.

        Args:
            url (str): The URL to crawl for documents.

        Returns:
            List[Document]: A list of Document objects containing the retrieved content.
        """
        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

# Load the crawled saved docs from the local file
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:
    """
    Creates a FAISS vector store from a list of documents.

    Args:
        docs (List[Document]): A list of Document objects containing the content to be stored.
        store_path (Optional[str]): The path to store the vector store locally. If None, the vector store will not be stored.

    Returns:
        FAISS: The FAISS vector store containing the documents.
    """
    # Creating text splitter
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200,
    )

    texts = text_splitter.split_documents(docs)

    # Embedding object
    embedding_model = OpenAIEmbeddings()

    # Create the FAISS vector store
    store = FAISS.from_documents(texts, embedding_model)

    # Save the vector store locally if a path is provided
    if store_path:
        store.save_local(store_path)

    return store


# create vector store
store = create_vector_store(saved_docs)

# creating retriever
retriever = store.as_retriever()

第4步：创建用于生成响应的检索链

现在，我们将创建 create_generate_chain 函数，该函数将创建一个生成响应的链。为了创建这个链，我们将首先使用 generate_template，在其中提供关于该过程的详细说明。模板有两个占位符：{context} 用于存储相关信息，{input} 用于问题。然后，我们将使用 LangChain 的 PromptTemplate 模块，它接受两个变量：template = generate_template 和 input_variables = ["context", "input"]。

最后一步是使用 generate_prompt、llm 模型和 StrOutputParser() 创建 generate_chain。

# generate_chain.py
from langchain.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser

def create_generate_chain(llm):
    """
    创建一个用于回答代码相关问题的生成链。

    参数:
        llm (LLM): 用于生成响应的语言模型。

    返回:
        一个可调用的函数，该函数接受上下文和问题作为输入，并返回一个字符串响应。
    """
    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() 用于从 LLM 获取字符串输出，否则输出可能会很复杂，例如 JSON 或结构化消息对象，可能无法直接用于进一步处理或显示给用户。例如，未使用 StrOutputParser() 的输出可能如下所示：

{
    "content": "这是来自 LLM 的响应。",
    "metadata": {
        "confidence": 0.8,
        "response_time": 0.5
    }
}

而使用 StrOutputParser() 后，输出将如下所示：

这是来自 LLM 的响应。

第5步：创建评分器

在此步骤中，我们将创建不同的评分器，以评估检索到的文档的相关性、评估生成的答案、检查答案是否是虚构的，以及在未获得相关文档时的查询重写器。我们将逐步进行每一个部分。

检索评分器

我们将首先创建一个检索评分器，以评估检索到的文档与用户问题的相关性。为此，我们将定义一个 create_retrieval_grader 函数，该函数接受一个带有新指令的提示模板 grade_prompt。

它表示评分器应在文档中查找与用户问题相关的关键词。如果存在这样的关键词，则该文档被视为相关。然后，它应提供一个二元评分，即“是”或“否”，以指示文档是否与问题相关，并以 JSON 格式提供结果，只有一个键“score”。

def create_retrieval_grader(model):
    """
    创建一个检索评分器，以评估检索到的文档与用户问题的相关性。

    返回：
        一个可调用的函数，接受文档和问题作为输入，并返回一个 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 = "法国是一个位于欧洲的国家。巴黎是法国的首都。"
question = "法国的首都是什么？"
score = grader(document, question)
print(score)  # 输出: {"score": "yes"}

虚构评分器

接下来，我们将定义一个虚构评分器，以评估从 LLM 获得的答案是否基于或得到一组事实的支持。然后，它提供一个二元评分（“是”或“否”），指示答案是否有依据。提示模板将包括事实的占位符（{documents}）和答案的占位符（{generation}），在使用提示时将填充这些占位符。

def create_hallucination_grader(self):
    """
    创建一个虚构评分器，以评估答案是否基于/得到一组事实的支持。

    返回：
        一个可调用的函数，接受一个生成的答案和一组文档（事实）作为输入，并返回一个 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 = ChatOpenAI(model="gpt-4o", temperature=0)
## 评分器
grader = create_hallucination_grader(model)
answer = "法国的首都为巴黎。"
facts = ["法国是一个位于欧洲的国家。", "巴黎是法国的首都。"]
score = grader(answer, facts)
print(score)  # 输出: {"score": "yes"}

代码评估器

接下来，我们将定义一个 create_code_evaluator 函数，创建一个代码评估器，以评估生成的代码是否正确且与给定问题相关。它使用一个 PromptTemplate 来指示评估器提供一个 JSON 响应，包含一个二元评分和反馈。评估器接受一个生成的代码、一个问题和一组文档作为输入，并返回一个 JSON 对象，包含一个评分，指示代码是否正确且相关，以及对评估的简要说明。

def create_code_evaluator(self):
    """
    创建一个代码评估器，以评估生成的代码是否正确且与给定问题相关。

    返回：
        一个可调用的函数，接受一个生成的代码、一个问题和一组文档作为输入，并返回一个 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 = "写一个函数来根据名字问候某人。"
documents = ["一个函数应该接受一个名字作为输入并返回一个问候消息。"]

result = code_evaluator(code, question, documents)
print(result)  # 输出: {"score": "yes", "feedback": "代码正确且与问题相关。"}

问题重写器

最后，我们将创建 create_question_rewriter 函数，该函数构建一个重写链，以改进给定问题的清晰度和相关性。此函数返回一个可调用的函数，接受一个问题作为输入，并将重写的问题作为字符串输出。

def create_question_rewriter(model):
    """
    创建一个问题重写链，以重写给定问题以提高其清晰度和相关性。

    返回：
        一个可调用的函数，接受一个问题作为输入，并返回重写的问题作为字符串。
    """
    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 = "如何使用 speckle 的 python sdk？"
rewritten_question = rewriter(original_question)
print(rewritten_question)  # 输出: "如何安装 speckle 的 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 模型
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 笔记本 。还有另一篇很棒的 文章 ，由 Philipp Kaindl 撰写，解释了高级 RAG 技术以及与 AWS bedrock 的部署。

第6步：创建图形

现在我们已经拥有所有组件，可以开始使用 LangGraph 创建我们的图形。在我之前的 博客文章 中，我详细介绍了图形工作流的核心概念。在此，我假设您具备必要的工作知识。

定义图形的状态

最初，我们将定义一个 GraphState 类，该类定义了图形的状态，由三个关键属性组成：input、generation 和 documents。input 属性保存作为字符串处理的输入或问题，而 generation 属性存储基于输入的语言模型（LLM）输出，同样为字符串。documents 属性表示相关文档的字符串列表。

## Start the Graph
from typing_extensions import TypedDict
from typing import List

class GraphState(TypedDict):
    """
    Represents the state of our graph.

    Attributes:
        question: question
        generation: LLM generation
        documents: list of documents
    """

    input: str
    generation: str
    documents: str #List[str]

该状态在整个图形中全局可访问，这些属性是唯一可以被节点内的函数修改的变量。这将引导我们定义节点。

节点

节点可以是 Python 函数，这些函数将获取图形的状态，执行一些操作，并修改任何状态变量。让我们定义一个名为 GraphNodes 的类。在当前目录中，utils 文件夹包含所有模块，因此我们将从 utils.generate_chain 导入 create_generate_chain 函数作为模块。

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):
        """
        Retrieve documents

        Args:
            state (dict): The current graph state

        Returns:
            state (dict): New key added to state, documents, that contains retrieved documents
        """
        print("---RETRIEVE---")
        question = state["input"]

        # Retrieval
        documents = self.retriever.invoke(question)
        return {"documents": documents, "input": question}

    def generate(self, state):
        """
        Generate answer

        Args:
            state (dict): The current graph state

        Returns:
            state (dict): New key added to state, generation, that contains LLM generation
        """
        print("---GENERATE---")
        question = state["input"]
        documents = state["documents"]

        # RAG generation
        generation = self.generate_chain.invoke({"context": documents, "input": question})
        return {"documents": documents, "input": question, "generation": generation}

    def grade_documents(self, state):
        """
        Determines whether the retrieved documents are relevant to the question.

        Args:
            state (dict): The current graph state

        Returns:
            state (dict): Updates documents key with only filtered relevant documents
        """
        print("---CHECK DOCUMENT RELEVANCE TO QUESTION---")
        question = state["input"]
        documents = state["documents"]

        # score each doc
        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):
        """
        Transform the query to produce a better question.

        Args:
            state (dict): The current graph state

        Returns:
            state (dict): Updates question key with a re-phrased question
        """
        print("---TRANSFORM QUERY---")
        question = state["input"]
        documents = state["documents"]

        # Re-write question
        better_question = self.question_rewriter.invoke({"input": question})
        return {"documents": documents, "input": better_question}

该类定义了图形的节点函数，负责图形工作流中的各种任务。以下是每个函数的描述：

1. retrieve：根据输入问题检索文档，并将其添加到图形状态中。
2. generate：使用输入问题和检索到的文档生成答案，并将生成结果添加到图形状态中。
3. grade_documents：根据检索到的文档与输入问题的相关性进行过滤，更新图形状态，仅保留相关文档。
4. transform_query：重新表述输入问题，以提高其清晰度和相关性，更新图形状态中的转换问题。

接下来，我们将定义 EdgeGraph 类，该类定义了图形的边函数。

边

边函数引导图形处理管道，根据当前状态和各种节点函数的结果做出决策。

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):
        """
        Determines whether to generate an answer, or re-generate a question.

        Args:
            state (dict): The current graph state

        Returns:
            str: Binary decision for next node to call
        """
        print("---ASSESS GRADED DOCUMENTS---")
        question = state["input"]
        filtered_documents = state["documents"]

        if not filtered_documents:
            # All documents have been filtered check_relevance
            # We will re-generate a new query
            print("---DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY---")
            return "transform_query"  # "retrieve_from_community_page", "transform_query"
        else:
            # We have relevant documents, so generate answer
            print("---DECISION: GENERATE---")
            return "generate"

    def grade_generation_v_documents_and_question(self, state):
        """
        Determines whether the generation is grounded in the document and answers question.

        Args:
            state (dict): The current graph state

        Returns:
            str: Decision for next node to call
        """
        print("---CHECK HALLUCINATIONS---")
        question = state["input"]
        documents = state["documents"]
        generation = state["generation"]

        score = self.hallucination_grader.invoke({"documents": documents, "generation": generation})
        grade = score["score"]

        # Check hallucination
        if grade == "yes":
            print("---DECISION: GENERATION IS GROUNDED IN DOCUMENTS---")
            # Check question-answering
            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"

以下是每个函数的描述：

1. decide_to_generate：根据过滤文档与输入问题的相关性，决定是生成答案还是重新生成问题。如果所有文档都不相关，则决定转换查询；否则，决定生成答案。
2. grade_generation_v_documents_and_question：根据生成的答案是否基于文档以及是否能够解决问题来评估生成的答案。如果生成是基于文档并解决了问题，则被视为有用；否则，视为不支持或无用。

现在我们已经定义了图形状态、节点和边函数，我们可以最终开始构建我们的图形。

构建图形

# Initiating the Graph
workflow = StateGraph(GraphState)
# Create an instance of the GraphNodes class
graph_nodes = GraphNodes(llm, retriever, retrieval_grader, hallucination_grader, code_evaluator, question_rewriter)
# Create an instance of the EdgeGraph class
edge_graph = EdgeGraph(hallucination_grader, code_evaluator)
# Define the nodes
workflow.add_node("retrieve", graph_nodes.retrieve) # retrieve documents
workflow.add_node("grade_documents", graph_nodes.grade_documents)  # grade documents
workflow.add_node("generate", graph_nodes.generate) # generate answers
workflow.add_node("transform_query", graph_nodes.transform_query)  # transform_query
# Build graph
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"
    },
)
# Compile
chain = workflow.compile()

首先，我们将从已经定义的 StateGraph 类初始化图形。接下来，我们将创建 graph_nodes 和 edge_graph 实例，分别来自 GraphNodes 和 EdgeGraph 类。

然后，我们将添加已经定义了函数的节点：

1. Retrieve: 根据输入问题检索相关文档。
2. Grade Documents: 根据文档与问题的相关性过滤检索到的文档。
3. Generate: 根据过滤后的文档生成答案。
4. 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 应用。我们通过导入 FastAPI 并创建 FastAPI 类的实例来实现。我们传入一些元数据，例如应用的标题、版本和描述。

app = FastAPI(
    title="Speckle Server",
    version="1.0",
    description="An API server to answer questions regarding the Speckle Developer Docs"
)

接下来，我们定义一个根 URL ("/") 的路由，该路由重定向到文档 URL ("/docs")。这是 FastAPI 应用中的一种常见模式，因为它允许用户轻松访问文档。

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

我们使用 Pydantic 的 BaseModel 定义两个模型：Input 和 Output。这些模型将用于定义我们 API 的输入和输出数据的结构。

class Input(BaseModel):
    input: str

class Output(BaseModel):
    output: dict

我们使用 add_routes 函数向应用添加路由。该函数接受三个参数：应用实例、链实例和路由的路径。在这种情况下，我们为 /speckle_chat 端点添加了一个路由。

add_routes(
    app,
    chain.with_types(input_type=Input, output_type=Output),
    path="/speckle_chat",
)

最后，我们使用 Uvicorn 运行服务器。我们导入 Uvicorn 并调用 run 函数，传入应用实例、主机和端口。

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

就这样。通过这些步骤，我们创建了一个 FastAPI 应用并启动了一个可以在 http://localhost:8000 访问的服务器。

第8步：创建一个带有 Streamlit/Gradio UI 的客户端

我们现在将创建一个 client.py 文件，该文件将使用 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():
                    # Node
                    pprint(f"Node '{key}':")
                    # Optional: print full state at each node
                    # 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 命令从 LLM 模型流式传输响应，同时打印图形工作流中被触发的节点。最后，我们将从值字典中检索存储在 'generation' 键中的最终输出。如果在处理过程中出现错误，将显示错误消息。它的样子是这样的！ 😎

可选：使用 Gradio 创建用户界面

您还可以使用 Gradio；这是一个开源的 Python 库，用于为机器学习模型、API 和任意 Python 函数创建交互式基于 Web 的用户界面。它的主要目的是通过提供易于使用的界面来弥合机器学习模型与最终用户之间的差距，从而便于部署和与这些模型进行交互。

让我们开始创建一个函数，以便从 LLM 模型获取最终响应。

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():
            # Node
            pprint(f"Node '{key}':")
            # Optional: print full state at each node
            # pprint.pprint(value["keys"], indent=2, width=80, depth=None)
        pprint("\n---\n")
    output = value['generation']
    return output      

现在，我们将创建一个简单的 Gradio 用户界面，在 Gradio 的 Interface 函数中将 get_response 函数分配给 fn 变量。

import gradio as gr
from langserve import RemoteRunnable
from pprint import pprint

# Create the UI In Gradio
iface = gr.Interface(fn=get_response, 
          inputs=gr.Textbox(
          value="输入您的问题"), 
          outputs="textbox",  
          title="关于 Speckle 开发文档的问答",
          description="询问有关 Speckle 开发文档的问题，并从代码助手那里获得答案。该助手查找相关文档并回答您的代码相关问题。",
          examples=[["如何安装 Speckle 的 Python SDK？"], 
                  ["如何从 Speckle 提交和检索对象？"],
                  ],
          theme=gr.themes.Soft(),
          allow_flagging="never",)

iface.launch(share=True) # put share equal to True for public URL

这就是它的样子！ 👇🏼

您只需在 launch 函数中包含 share=True，即可在本地 URL 之上获取公共 URL。

结论

在这篇博客中，我们探讨了一个服务器-客户端架构的图形工作流的开发，该架构结合了先进的RAG（检索增强生成）概念。服务器组件涵盖了一个全面的管道，包括对检索到的文档进行评分、对响应进行评分、检查幻觉以及查询重写。

为了与这个本地服务器进行交互，我们创建了两个客户端应用程序，一个使用Streamlit，另一个使用Gradio。两个用户界面都提供了友好的界面，供用户输入查询并实时接收服务器的响应。这是一个端到端的项目，将允许开发人员构建一个应用程序并在本地进行测试，然后再部署到生产环境中。

在下一篇文章中，我将介绍如何使用Docker对这个应用程序进行容器化，并使用Google Cloud Platform进行部署。

这里是Speckly机器人的GitHub仓库 。

随着我继续学习，我将添加更多内容。如果你喜欢我的贡献，可以用⭐️来支持这个仓库 :-)

你也可以在LinkedIn 上与我联系！

À bientôt!