让双子座选择数据源：FAISS、Tavily以及LangGraph中的直接回答功能

大多数问答原型都会将每个查询都送入固定的处理流程，尽管不同问题的需求各异：有些需要依赖公司的内部文档，有些则需要实时变化的事实，还有些仅需一般性知识即可。本指南将构建一个简洁的 Python 工作流，让语言模型首先分析每个问题，然后将其发送到合适的地方——用于存储内部知识的 FAISS 向量库、用于获取实时网络结果的 Tavily，或直接发送给 Gemini。学习完成后，你将了解在 LangGraph 中状态、节点和条件边是如何相互关联的，为何结构化输出能让大型语言模型路由器更可靠，以及这种用于教学的设计在面向真实用户之前还需要哪些改进。如果你想了解更多类型的图结构，可以阅读关于路由、扇出结构的文章。

“LangGraph中的评价与审批模式”为这一实践性构建提供了补充。

三个问题，三种不同的数据来源

想象有三名用户。第一人询问公司的休假政策，只有内部文档才能回答；第二人想了解北阿坎德邦今天的天气，由于没有内部文档包含此类信息，系统必须上网搜索；第三人则询问RAG是什么，模型已经知晓答案，再获取其他信息只会增加延迟。

因此，目标架构在三个分支之前设置了一个由Gemini驱动的路由器，所有分支最终都会汇聚到同一个答案生成步骤：

User Question
                               |
                               v
                       +----------------+
                       |   AI Router    |
                       |    (Gemini)    |
                       +----------------+
                         /      |      \
                        /       |       \
                       v        v        v
                    FAISS    Tavily    Gemini
                 Internal DB  Web      Direct
                       \        |        /
                        \       |       /
                         v      v       v
                       +----------------+
                       | Generate Answer|
                       |     Gemini     |
                       +----------------+
                               |
                               v
                            Answer

该路由器是整个设计的核心。一种诱人的捷径是关键词匹配，如下方伪代码所示，通过问题中的特定词汇来选择对应的分支：

if "weather" in question:
    use_tavily()
elif "leave" in question:
    use_faiss()
else:
    use_gemini()

在此情况下，该决策被交由 Gemini 来处理。由于模型是理解问题内容而非仅扫描其中的触发词，因此这种工作流程被称为“智能体式”工作流程。

生成式 AI、智能体与智能体式工作流程

这些术语经常被混用，但它们实际上描述了模型参与程度的不同层次。

普通的生成式 AI

最简单的应用方式是将提示语传递给模型，然后获取模型生成的输出内容：

User Question
      |
      v
     LLM
      |
      v
    Answer

它能够根据训练数据解释 RAG 技术，但对你的休假政策却一无所知。

调用工具的智能体

这类智能体会为模型添加数据库、搜索引擎或外部 API 等工具，让模型在回复之前决定是否需要调用这些工具：

User Question
      |
      v
     LLM
      |
      +------> Database
      |
      +------> Web Search
      |
      +------> API
      |
      v
    Answer

例如，天气相关的问题可以通过查询天气服务来获得答案，而非让模型自行编造看似合理的预测。

智能体工作流

智能体工作流能让模型在运行时决定执行路径。在这个项目中，流程如下：

Question
   |
   v
Router
   |
   +----> FAISS
   |
   +----> Tavily
   |
   +----> Gemini

开发者会定义三种可能的处理路径；模型只需针对每个输入问题从中选择一种。它并没有获得对应用程序的完全控制权，只有一份允许执行的操作列表。“智能体路由工作流”才是对这种架构最准确的描述。

项目设置

你需要安装 Python 3.10 或更高版本，以及 Google Gemini 和 Tavily 的 API 密钥。可以一次性安装所有相关库：

pip install -U langchain langchain-google-genai langchain-community langgraph faiss-cpu tavily-python python-dotenv pydantic

请将这两个密钥保存在 .env 文件中，而非源代码里：

GOOGLE_API_KEY=your_google_api_key
TAVILY_API_KEY=your_tavily_api_key

这些导入包括类型定义辅助工具、用于路由方案的Pydantic库、LangChain的文档处理及FAISS封装类、Gemini的聊天与嵌入相关类、LangGraph的图结构处理功能以及Tavily客户端：

import os
from typing import List, Literal
from typing_extensions import TypedDict
from dotenv import load_dotenv
from pydantic import BaseModel
from langchain_core.documents import Document
from langchain_community.vectorstores import FAISS
from langchain_google_genai import (
    ChatGoogleGenerativeAI,
    GoogleGenerativeAIEmbeddings,
)
from langgraph.graph import StateGraph, START, END
from tavily import TavilyClient

下一步，如单行代码片段所示，仅需加载环境变量即可：

Load the environment variables:

当设置override=True时，.env文件中的值会优先于已在shell中设置的变量：

load_dotenv(override=True)

配置Gemini用于路由、问答及嵌入功能

该聊天模型承担双重职责：首先决定由哪个模块处理问题，随后根据收集到的上下文生成最终答案。以下配置启用了两次自动重试功能，同时未设置令牌数和超时限制：

model = ChatGoogleGenerativeAI(
    model="gemini-3.6-flash",
    max_tokens=None,
    timeout=None,
    max_retries=2,
    api_key=os.getenv("GOOGLE_API_KEY"),
)

模型标识符会频繁变化，因此在运行代码之前，请务必根据当前的 Gemini 模型列表确认代码片段中的名称是否正确。

嵌入值是另一个独立的问题，由专门的模型处理。Gemini Embedding 会将每份文档转换为数值向量：

embeddings = GoogleGenerativeAIEmbeddings(
    model="models/gemini-embedding-001",
    google_api_key=os.getenv("GOOGLE_API_KEY"),
)

正是向量使得语义搜索成为可能：含义相似的文本在向量空间中会处于相近位置，因此即使问题与相关文本仅有少量完全相同的词汇，也能检索到对应的段落。

FAISS 中的小型内部知识库

为让用户专注于工作流程，知识库中仅包含三份简短的文档：30天退款政策、需提前7天申请的可享受20天带薪假的规定，以及工作日服务时间。在真实系统中，这些文本可能来自手册、PDF文件、Notion页面、支持工单、内部文档或数据库记录。

documents = [
    Document(
        page_content="""
        Our company provides a 30-day refund policy.
        Customers can request a refund within 30 days of purchase.
        """
    ),
    Document(
        page_content="""
        Employees receive 20 paid vacation days per year.
        Vacation requests must be submitted at least 7 days in advance.
        """
    ),
    Document(
        page_content="""
        The company provides technical support from Monday to Friday,
        9 AM to 6 PM IST.
        """
    ),
]

构建存储系统并将其封装为检索工具需要两次调用。将k设置为3时，系统会查找距离最近的三份文档；在这个小型语料库中，这意味着所有文档都会按照相似度排序后返回：

vector_store = FAISS.from_documents(
    documents,
    embeddings,
)

retriever = vector_store.as_retriever(
    search_kwargs={"k": 3}
)

FAISS并不理解语言，它只是用于最近邻搜索的索引工具。嵌入模型会将输入的问题转换为向量，FAISS则会返回那些向量与之最接近的存储文档。

添加Tavily客户端以获取实时信息

网络搜索只需使用第二个 API 密钥构建客户端实例即可：

tavily_client = TavilyClient(
    api_key=os.getenv("TAVILY_API_KEY")
)

共享状态的设计

在 LangGraph 中，状态是核心概念：它是一种类型化的字典，会在图中传递，每个节点都会读取并更新它。这一工作流程需要问题、检索到的所有文档、Tavily 的结果、选定的来源以及最终答案：

class AgentState(TypedDict):
    question: str
    documents: List[Document]
    tavily_response: str
    source: str
    answer: str

从概念上讲，它就像一个所有步骤都能查看的共享记录：

AgentState
                    |
        +-----------+-----------+
        |           |           |
    question    documents    source
                    |
             tavily_response
                    |
                  answer

节点接收当前状态，使用其关心的字段并返回更新内容，LangGraph 会合并这些更新后再将状态传递给下一个节点。

FAISS 检索节点

该节点从状态中获取问题，通过检索器进行处理，并存储匹配到的文档：

def retrieve_from_faiss(state : AgentState) -> AgentState:
    question = state['question']

    """ Fetch the details from the FAISS vector database
    """

    result = retriever.invoke(question)

    return {**state, "documents": result}

仅当路由器选择了内部知识库时才会执行此操作。如下所示的问题就是典型的触发条件：

"What is our company leave policy?"

对于该输入，检索器应调出相关假期文档：

Employees receive 20 paid vacation days per year.
Vacation requests must be submitted at least 7 days in advance.

Tavily搜索节点

网页搜索节点会将问题发送给Tavily，并将search_depth设置为advanced，然后从每个结果中提取content字段：

def search_with_tavily(state: AgentState) -> AgentState:
    question = state['question']
    """ Using the Tavily to search the web and
      fetch the latest information about user query
    """

    response = tavilyClient.search(
        query=question,
        search_depth='advanced'
    )

    contents = [result["content"] for result in response["results"]]
    return {**state, "tavilyResponse":contents}

这些片段将成为Gemini在生成答案时所参考的上下文。请注意，该节点返回的是字符串列表；如果希望得到一段完整的文本，在存储之前先将这些内容拼接起来，以便该字段与状态中声明的str类型相匹配。

通过结构化输出提升路由器的可靠性

简单的路由器会要求模型用三个基本词汇之一来回复：

Return only:
faiss
tavily
gemini

语言模型并不总是遵循格式化要求。你可能会得到一个完整的句子作为回复：

I would choose tavily.

或者是以略有不同的表述方式：

The best option is: tavily

无论是哪种回复都会破坏需要精确字符串来选择边的图结构。结构化输出可以填补这一缺陷。首先将允许的决策描述为一个Pydantic模型，该模型的单个字段仅限于三个固定值：

class RouteDecision(BaseModel):
    source: Literal["faiss", "tavily", "gemini"]

然后定义一个路由器模型，要求其返回该模式的实例。json_schema方法促使生成过程遵循该模式，而非仅依赖提示词的内容：

router_model = model.with_structured_output(
    RouteDecision,
    method="json_schema",
)

输出内容会依据Literal类型进行验证，因此无效值会以错误形式呈现，而不会导致图结构指向未知方向。

编写决策节点

路由提示会说明每种选项的用途：对于与公司政策及其他内部知识相关的问题，使用 faiss；对于需要最新信息或网络信息的查询，使用 tavily；而对于既不需要前者也不需要后者的普通知识问题，则使用 gemini。

def decide_source(state:AgentState)-> AgentState:
    question = state["question"]
    prompt = f"""
    Decide the best source for answering this question.

    Choose exactly one:

    faiss:
    Use when the question can be answered using our internal
    knowledge base.related to company policy and all

    tavily:
    Use when the question requires current, recent, or web-based
    information.

    gemini:
    Use when the question is general knowledge and does not
    require our internal documents or current web information.

    Question:
    {question}

    Return only one word:
    faiss, tavily, or gemini
    """

    response = model.invoke(prompt)
    # return response

    return {**state,"source":response.text}

仔细查看最后几行代码。按照当前的写法，节点仍然会调用普通的 model 并存储 response.text，这正是上文所描述的那种不稳定的自由文本处理方式。若要利用模式化的优势，应改为调用 router_model.invoke(prompt) 并存储结果的 source 属性。经过这样的修改后，路由器返回的将是类似这样的类型化对象，而非随意的文字内容。

RouteDecision(source="faiss")

告知 LangGraph 下一步该去何处

条件边需要一个函数来指定应遵循哪条分支。这个函数不会自行做决策，而是读取路由器已保存在状态中的选择，并将其传回给LangGraph：

def route_source(
    state: AgentState,
) -> Literal["faiss", "tavily", "gemini"]:
    return state["source"]

每条路由对应一个答案节点

所有三条分支都在同一生成步骤中完成处理。它会检查source内容并据此构建提示词：将检索到的文档整合为FAISS所需的上下文，将搜索片段作为Tavily的参考资料，或是直接使用问题本身来获取答案：

# Generate the answer for the user
def generateAnswer(state:AgentState) -> AgentState:
    source = state["source"]
    question = state['question']
    documents = state['documents']
    tavilyResponse = state['tavilyResponse']

    if source == "faiss":
        context = "\n\n".join([doc.page_content for doc in documents])
        prompt = f"""Based on the following context answer the question below
           Context:
           {context}

        Question:
        {question}
           """
    elif source == "tavily":
        prompt = f""" Based on the following search result , use this as an reference and provdie the
         answer to the below question

         Context:
         {tavilyResponse}

         Question:
         {question}
        """
    else:
        prompt = f" Answer the following question : {question}"
    response = model.invoke(prompt)
    answer = response.content

    return {**state, "answer":answer}

所有的答案都由同一个Gemini模型生成。唯一不同的就是放在其前的上下文，而这正是RAG的核心理念：更好的输入而非不同的模型才能产生有依据的响应。

在整合片段时保持名称一致

这些代码片段混用了两种命名风格，如果直接原样拼接就会导致错误。状态中声明的是 tavily_response，而节点的读写操作却使用 tavilyResponse；客户端被创建为 tavily_client，但调用时却用的是 tavilyClient；答案函数定义为 generateAnswer，但注册时却写成了 generate_answer。在运行图模型之前，请统一选择一种命名规范并全程使用。

组装图模型

目前这些组成部分包括一个决策节点以及三种可能的后续处理路径：

decide_source
      |
      +----> faiss
      |
      +----> tavily
      |
      +----> gemini

这里有一个细节需要注意。gemini 路径根本不属于检索步骤，它的含义是“跳过检索直接给出答案”。因此无需为它创建空节点，该分支可以直接指向共享的生成节点。

首先为状态类型创建一个图：

workflow = StateGraph(AgentState)

注册四个节点：

workflow.add_node("decide", decide_source)
workflow.add_node("faiss", retrieve_from_faiss)
workflow.add_node("tavily", search_with_tavily)
workflow.add_node("generate", generate_answer)

将决策节点设为入口点：

workflow.add_edge(START, "decide")

连接条件边。该映射会将路由函数可能返回的每个值转换为节点名称，gemini就会直接被发送到对应的节点以执行generate操作：

workflow.add_conditional_edges(
    "decide",
    route_source,
    {
        "faiss": "faiss",
        "tavily": "tavily",
        "gemini": "generate",
    },
)

检索和搜索分支随后需连接到答案生成环节：

workflow.add_edge("faiss", "generate")
workflow.add_edge("tavily", "generate")

生成是图结构结束前的最后一步：

workflow.add_edge("generate", END)

编译会将定义转换为可运行的应用程序：

app = workflow.compile()

完成的图结构

从开始到结束的完整流程：

START
                           |
                           v
                    +--------------+
                    |    decide    |
                    |    source    |
                    +--------------+
                     /      |      \
                    /       |       \
                   v        v        v
               +------+ +--------+ +---------+
               |FAISS | | Tavily | | Generate|
               |      | |        | | directly|
               +------+ +--------+ +---------+
                   \        |         /
                    \       |        /
                     v      v       v
                    +----------------+
                    |    generate    |
                    |     answer     |
                    +----------------+
                            |
                            v
                           END

请牢记核心原则：图结构决定了所有可能的路径集合，而模型在运行时从中选择一条路径。正是这种结合使得工作流具有主动性，同时又不至于变得不可预测。

尝试三种路径

一个小型辅助函数用于构建初始状态并调用已编译的应用程序：

def ask_question(question: str):
    initial_state = {
        "question":question,
        "documents":[],
        "tavilyResponse":"",
        "source":""
    }

    result = app.invoke(initial_state)
    return result

需要通过网络解答的问题

天气相关问题应发送至Tavily：

result = ask_question(
    "What is the current weather in Uttarakhand?"
)

同时打印所选择的来源以及生成的答案：

print("Source:", result["source"])
print("Answer:", result["answer"])

预期的处理路径：

Source: tavily

当前的功能仅存在于网络环境中。

常识类问题

接下来，可以询问有关检索增强生成技术的问题：

result = ask_question(
    "What is Retrieval Augmented Generation?"
)

以相同方式打印结果：

print("Source:", result["source"])
print("Answer:", result["answer"])

路由器应完全跳过检索步骤：

Source: gemini

该模型能够自行解释相关概念。

关于内部政策的问题

最后是关于休假政策的问题：

result = ask_question(
    "What is our company leave policy?"
)

以及相同的打印语句：

print("Source:", result["source"])
print("Answer:", result["answer"])

这次内部知识库应该能给出正确答案：

Source: faiss

大语言模型的路由是概率性的，因此应将这些视为预期结果而非必然保证。

为何语义路由优于关键词规则

这是另一种硬编码的解决方案：

if "weather" in question:
    use_tavily()
elif "leave" in question:
    use_faiss()
else:
    use_gemini()

这类规则很快就会失效。比如有用户询问办公室周六是否开放，句子中并未提及任何政策内容，但答案很可能存在于内部文档中。基于模型的路由器可以推断出用户的意图；而关键词列表则无法做到，除非有人能预见到所有可能的表达方式。

这种方法的扩展性也更好。当出现新的后端时，比如下面的这些，只需扩展模式和提示词，而无需增加复杂的条件语句：

SQL Database
Internal API
CRM
Customer Support System
Documentation
Web Search

其代价是在开始实际工作之前需要多进行一次模型调用，从而产生相应的成本和延迟。

这真的算是人工智能代理吗？

将其称为小型代理工作流比自主代理更为准确。可用的操作是预先固定的：

FAISS
Tavily
Direct Gemini

模型无法自行决定执行类似以下操作，因为这些能力从未被赋予它：

delete a database
send an email
call an arbitrary API

该架构呈现出责任链的结构：

Developer defines possible actions
              |
              v
        LLM chooses action
              |
              v
        LangGraph executes
              |
              v
           Result

这种限制其实是一种优势：有限的选项能让生产环境中的行为更具可预测性和可审计性。

各组件的作用

LangGraph：工作流引擎

LangGraph掌控着应用程序的结构：状态、节点、边、条件路由以及执行顺序。从抽象层面来看，每一步都遵循相同的模式：

State
  |
  v
Node
  |
  v
Updated State
  |
  v
Conditional Edge
  |
  +----> Node A
  |
  +----> Node B
  |
  +----> Node C

由多个小步骤构成的图比一个庞大的函数更易于测试和观察。

FAISS：检索层

FAISS负责实现系统中的RAG功能。简而言之：

Company Documents
       |
       v
   Embeddings
       |
       v
     FAISS
       |
       v
Similar Documents
       |
       v
     Gemini
       |
       v
     Answer

在实际部署中，还会在其前面添加数据摄取管道：

Documents
   |
   v
Load
   |
   v
Split into chunks
   |
   v
Generate embeddings
   |
   v
Store vectors
   |
   v
Retrieve relevant chunks
   |
   v
Generate answer

该示例为突出工作流程而省略了数据加载和分块处理；实际手册则需要这两部分内容。

Tavily：实时网络搜索

Tavily用于处理会随时间变化的信息：

User Question
      |
      v
   Router
      |
      v
    Tavily
      |
      v
 Search Results
      |
      v
    Gemini
      |
      v
    Answer

典型应用场景包括实时天气、突发新闻、最新产品发布、更新后的文档、近期活动以及实时市场数据。在实际应用中，还需确定如何引用、筛选、验证和展示搜索结果，因为未经检查的网页内容既无法保证准确性，也不安全，不能直接传递给模型。

值得记住的模式

这些组件是可以相互替换的。核心理念在于路由：将每个问题与最合适的处理能力匹配起来。

Internal Knowledge
        |
        +------ FAISS
Current Information
        |
        +------ Tavily
General Knowledge
        |
        +------ Gemini

这种按请求选择处理能力的方式几乎出现在所有成熟的智能体应用中。

下一步发展方向

添加更多工具

路由器可以从更多的后端选项中进行选择：

SQL Database
REST APIs
CRM
Email
Calendar
Internal Documentation

强化检索功能

这三份文档仅用于演示，并非知识库。真正的RAG系统需要文档加载器、分块功能、元数据、更优的检索策略、重排序机制、来源引用以及访问控制，这样才能确保用户只能获取其被允许查看的内容。

验证路由决策

在路由器与工具之间设置验证步骤，可以排除不合理的选择并安全地回退：

Router
   |
   v
Validator
   |
   +---- valid ----> Tool
   |
   +---- invalid --> Fallback

随着工具数量的增加，这一点愈发重要。

处理故障

结构化输出只能保证存在有效路径，而非工具调用一定成功。搜索API可能会超时或返回空结果：

Router
   |
   v
Tavily
   |
   X
Search failed
   |
   v
Fallback

生产系统需要重试机制和备用方案，例如直接给出带有明确说明的答复。关于如何通过重试和备用方案设计具备弹性的智能体图结构的文章对此进行了更深入的探讨。

实现可观测性

当答案出错时，需要知道是哪个环节出现了问题：

Wrong route?
      |
      v
Bad retrieval?
      |
      v
Bad search results?
      |
      v
Bad generation?

记录选定的输入源及中间结果有助于回答上述问题。

引入循环结构

当前的图结构仅能做出一个决策：

Question
   |
   v
Router
   |
   v
Tool
   |
   v
Answer

功能更强的智能体会评估其找到的结果，并决定是否再次采取行动：

Question
   |
   v
Reason
   |
   v
Tool
   |
   v
Evaluate Result
   |
   +---- Need more information?
   |            |
   |            v
   |          Tool
   |            |
   +------------+
   |
   v
Final Answer

这正是代理系统展现真正力量的地方：模型会判断手头的信息是否足够，或者是否需要采取其他行动。同时这也是设定迭代次数的必要环节，以避免循环无限运行。

关键要点

整个系统要回答一个问题：人工智能应用应如何决定答案的来源？完整的工作流程如下：

User Question
                   |
                   v
                Gemini
                Router
                   |
        +----------+----------+
        |          |          |
        v          v          v
      FAISS      Tavily    Gemini
   Internal DB    Web      Direct
        |          |          |
        +----------+----------+
                   |
                   v
                Gemini
              Final Answer

• 在图中定义各种能力与边界，让模型在运行时从中选择。
• 使用结构化输出来做出路由决策，并确保决策节点确实调用了结构化模型。
• 仅保留一个答案生成节点，改变的只是输入给它的上下文。
• 将路由功能视为可测试的组件：记录决策结果，并用典型问题对其进行验证。

在添加更多工具之前，应先实现验证、备用方案及可观测性功能，因为每新增一个工具都会增加请求失败的可能。

基于这一基础，同样的理念可自然延伸至 SQL 数据库、API、内存系统、人工审核、评估节点、重试机制以及多智能体架构。虽然系统结构会越来越复杂，但核心原则始终不变：为模型提供实用能力，明确这些能力的使用方式，让模型自行决定哪种能力最适合当前任务。

相关阅读

• 在 FastAPI 中嵌入 LangChain 智能体：工具、手动搜索与流式响应 —— 使用 FastAPI 和 LangChain 构建应用内助手：将 ChromaDB 中的 PDF 手册作为工具提供，支持用户专属上下文、历史记录检查点以及流式回复功能。
• 利用 LangGraph 和 Amazon Bedrock 设计四层代理内存 — 学习如何在 Bedrock 和 LangGraph 上为大型语言模型代理赋予工作记忆、情景记忆、语义记忆和程序记忆功能，同时防范数据污染、个人身份信息泄露以及租户间数据交叉污染的问题。