AI智能体路由机制：让大模型拥有动态决策能力

AI智能体在处理任务时，不能总是按部就班地执行。在现实世界中，AI需要根据具体情况（比如用户的提问或当前环境）来决定下一步做什么。这种动态决策能力，就是通过一种叫路由的机制实现的。简单来说，路由就是给AI加一个“大脑”，让它能根据不同的条件，选择不同的处理路径。比如一个客服AI，它可以先判断用户的意图（是问产品信息，还是查订单），然后把问题转给专门处理这个领域的模块，而不是一律回复固定的答案。基于

Android女王

683人浏览 · 2025-10-11 21:05:42

Android女王 · 2025-10-11 21:05:42 发布

文章详细介绍了AI智能体中的路由机制，这是一种让AI能够根据不同情况动态决策的能力。文章解释了路由的四种主要实现方法（基于AI模型、规则、嵌入和机器学习模型），展示了其在智能客服、数据处理和多工具协作等场景的应用，并通过LangGraph和LlamaIndex框架的代码示例，帮助开发者理解如何在实际项目中实现路由功能，使AI从死板的执行者转变为灵活的智能系统。

概述

AI智能体在处理任务时，不能总是按部就班地执行。在现实世界中，AI需要根据具体情况（比如用户的提问或当前环境）来决定下一步做什么。这种动态决策能力，就是通过一种叫路由的机制实现的。

简单来说，路由就是给AI加一个“大脑”，让它能根据不同的条件，选择不同的处理路径。比如一个客服AI，它可以先判断用户的意图（是问产品信息，还是查订单），然后把问题转给专门处理这个领域的模块，而不是一律回复固定的答案。

实现路由有几种主要方法：

基于AI模型的路由：让大语言模型自己去分析用户问题，然后直接给出下一步的指令。
基于规则的路由：使用预先设定的规则（比如关键字）来判断，这种方法快但不够灵活。
基于嵌入（Embedding）的路由：通过比较用户问题和不同处理模块的“意思”（向量），来选择最匹配的路径。
基于机器学习模型的路由：用一个专门训练过的模型来做判断，这种方法比直接用大模型更快、更稳定。

路由可以在AI处理任务的任何阶段启用，比如一开始就对任务分类，或者在处理过程中根据需要选择工具。

像 LangGraph 这样的框架，就提供了专门的工具来构建这种复杂的路由系统，让AI能根据整个任务的进展来做出更聪明的决策。最终，路由让AI从一个死板的执行者，变成一个能灵活应对各种状况的智能助手。

实际应用场景

智能客服：路由让客服AI能读懂用户意图。例如，当用户提问时，AI会先判断他是想查订单、问产品还是有技术问题，然后把问题转给相应的处理模块或人工客服，而不是只会说“我没听懂”。
自动化数据处理：路由可以像一个“分拣中心”，自动根据数据类型、内容等，将数据分配给不同的处理流程。比如，将邮件转给销售团队，将工单转给客服，或者将特定格式的文件转给数据转换工具。
多工具协作：在复杂的系统中，路由就像一个“总调度师”。它能判断当前任务需要什么工具，然后将任务分配给最合适的AI或工具。比如，一个AI编程助手会先识别用户的意图（是想调试代码还是翻译代码），然后将代码片段交给正确的专用工具来处理。

总之，路由让AI从一个只会按部就班的执行者，变成一个能灵活思考、动态决策的智能系统，从而更好地应对各种复杂情况。

实际代码示例 (LangGraph)

代码通过langgraph实现了根据用户查询来进行后面处理的机制，其逻辑如下：

完整代码如下：

fromlanggraph.graphimportStateGraph, END

fromlangchain.promptsimportPromptTemplate

fromlangchain_openaiimportChatOpenAI

fromdotenvimportload_dotenv

importos

fromtypingimportTypedDict

classState(TypedDict):

    query: str

    response: str

# 1. 定义路由函数

# 这个函数是路由模式的核心，它将根据LLM的输出决定下一步的执行路径

defroute_query(state):

    query=state['query']

    # 使用一个LLM作为路由决策器

    # 提示词要求LLM只返回一个特定的类别字符串

    router_prompt=PromptTemplate(

        template="""你是一个智能客服机器人，请分析以下用户问题并将其归类为以下四种类型之一：

        - 产品信息

        - 订单状态

        - 其他问题

        请只返回类别名称，不要有任何多余的解释。

        用户问题: {query}

        类别:

        """,

        input_variables=["query"]

    )

    # --- Configuration ---

    load_dotenv()

    SL_API_KEY=os.getenv("SL_API_KEY")

    local_llm="Qwen/Qwen2.5-7B-Instruct"

    base_url="https://api.siliconflow.cn/v1"

    # Initialize the LLM model (using OpenAI in this example)

    llm=ChatOpenAI(model=local_llm, base_url=base_url, temperature=0, api_key=SL_API_KEY)

    # llm = ChatOpenAI(temperature=0)

    # LLM在这里扮演了路由角色

    classification=llm.invoke(router_prompt.format(query=query)).content

    print(f"用户问题: {query}")

    print(f"路由结果: {classification}")

    if"产品信息"inclassification:

        return"product_info"

    elif"订单状态"inclassification:

        return"order_status"

    else:

        return"general_response"

# 2. 定义处理函数

# 每个函数代表一个独立的业务逻辑或工具调用

defproduct_info_handler(state):

    print("正在处理产品信息...")

    return {"response": "这是关于产品信息的处理结果。"}

deforder_status_handler(state):

    print("正在查询订单状态...")

    return {"response": "这是关于订单状态的查询结果。"}

defgeneral_response_handler(state):

    print("正在提供通用回复...")

    return {"response": "抱歉，我无法理解您的问题，这是我的通用回复。"}

# 3. 构建 LangGraph 图

# 定义图的状态

workflow=StateGraph(State)

# 添加节点

workflow.add_node("product_info", product_info_handler)

workflow.add_node("order_status", order_status_handler)

workflow.add_node("general_response", general_response_handler)

# 设置路由

# 这里我们将路由函数作为图的条件边

workflow.add_conditional_edges(

    "__start__",  # 从开始节点

    route_query,  # 路由函数决定下一个节点

    {

        "product_info": "product_info",

        "order_status": "order_status",

        "general_response": "general_response",

    }

)

# 连接所有处理节点到END节点

workflow.add_edge("product_info", END)

workflow.add_edge("order_status", END)

workflow.add_edge("general_response", END)

# 编译图

app=workflow.compile()

print("\n=== 开始生成图形可视化 ===")

display_graph(app, file_name=os.path.basename(__file__))

print("=== 图形可视化完成 ===\n")

# 4. 运行示例

print("--- 运行示例 1：查询产品信息 ---")

result1=app.invoke({"query": "你们的最新手机型号有什么特点？"})

print(f"结果: {result1.get('response', '')}")

print("\n")

print("--- 运行示例 2：查询订单状态 ---")

result2=app.invoke({"query": "我的订单号123456发货了吗？"})

print(f"结果: {result2.get('response', '')}")

print("\n")

print("--- 运行示例 3：其他问题 ---")

result3=app.invoke({"query": "你好，今天天气怎么样？"})

print(f"结果: {result3.get('response', '')}")

实际代码示例 (LlamaIndex)

LlamaIndex也提供了路由机制，LlamaIndex的路由功能主要通过其 RouterQueryEngine 来实现，它的核心作用是根据用户的查询意图，将请求智能地分发给最合适的“子查询引擎”（Sub Query Engine），而不是像传统方式那样将所有请求都发给同一个引擎。

LlamaIndex 的路由机制主要依赖一个 Router（路由器）来做决策。这个路由器通常是一个大语言模型（LLM），它会分析用户的原始查询，并从一系列预设的选项中选择最合适的那个。

整个过程分为两个步骤：

意图分析：Router 首先会分析用户的查询。开发者需要为每个子查询引擎提供一个简洁的描述，告诉 Router 这个引擎擅长处理什么类型的问题。
动态分发：Router 会根据意图分析的结果，将查询分发给对应的子查询引擎。

importos

fromllama_index.coreimportVectorStoreIndex, SimpleDirectoryReader

fromllama_index.core.query_engineimportRouterQueryEngine

fromllama_index.core.toolsimportQueryEngineTool

fromllama_index.core.selectorsimportLLMSingleSelector

# 确保你的OpenAI API key已设置

os.environ["OPENAI_API_KEY"] ="你的OpenAI API Key"

# 加载产品文档数据并创建索引

docs_products=SimpleDirectoryReader(input_files=["./product_docs.txt"]).load_data()

index_products=VectorStoreIndex.from_documents(docs_products)

query_engine_products=index_products.as_query_engine()

# 加载销售数据并创建索引

docs_sales=SimpleDirectoryReader(input_files=["./sales_data.txt"]).load_data()

index_sales=VectorStoreIndex.from_documents(docs_sales)

query_engine_sales=index_sales.as_query_engine()

# 将查询引擎封装成工具

tool_products=QueryEngineTool.from_defaults(

    query_engine=query_engine_products,

    # 这里的描述非常重要，Router会根据它来做决策

    description="专门用于回答关于公司产品信息、功能、定价等问题。"

)

tool_sales=QueryEngineTool.from_defaults(

    query_engine=query_engine_sales,

    # 这里的描述非常重要，Router会根据它来做决策

    description="专门用于回答关于公司销售数据、业绩、市场份额等问题。"

)

######################### 这里是路由设置的关键 ############################3

# 构建路由查询引擎，并传入两个工具

query_engine_router=RouterQueryEngine(

    selector=LLMSingleSelector.from_defaults(),

    query_engine_tools=[

        tool_products,

        tool_sales,

    ]

)

# 问题1: 询问产品信息

response1=query_engine_router.query("我们的最新产品有什么特点？")

print("问题1的路由结果：")

print(response1)

print("\n")

# 问题2: 询问销售数据

response2=query_engine_router.query("上个月的销售业绩如何？")

print("问题2的路由结果：")

print(response2)