大型语言模型（LLM）的日益多样化带来了一个挑战：如何高效地将特定任务分配给最具能力的模型。传统系统依赖于预定义规则或手动干预，这可能效率低下且缺乏灵活性。我们需要一个更智能的解决方案，能够动态评估上下文并自动将任务路由到合适的模型。

介绍

在人工智能时代，我们经常面临为特定任务选择合适模型的挑战。随着为不同应用（从情感分析到代码生成）训练的专用大型语言模型（LLM）数量不断增加，高效地部署它们已成为一个关键问题。虽然这种专业化是一个显著优势，但它也引入了一个关键挑战：如何确定给定任务应使用哪个模型。

传统上，解决这一问题的方法通常依赖于预定义规则或人工干预来将任务与适当的模型匹配，或将所有任务传递给单个通用模型。然而，这些方法存在固有局限性。基于规则的系统可能过于僵化，无法适应细微或不可预见的上下文。另一方面，人工选择过程既耗时又容易出错，尤其是在任务复杂性和数量增加时。从根本上说，这些方法缺乏一种清晰且动态的机制来理解给定输入的上下文并将其路由到最合适的模型。

此外，专用模型在成本方面往往存在差异——有些模型的计算成本比其他模型更高。如果没有优化选择过程的机制，组织可能会面临不必要的支出，同时性能也会受到影响。

为了解决这一问题，我建议使用LLM作为语义路由器——这是一种能够分析消息历史的上下文并确定最适合该任务的专用模型的智能系统。通过将任务路由到专为应对特定挑战而设计的专用代理，这种方法不仅可以提高任务准确性，而且根据你的配置，还可以通过使用针对特定任务进行微调的更小、更便宜的模型来节省资金。这种方法为与多个专用LLM合作提供了更灵活、动态和高效的解决方案，帮助组织优化其人工智能基础设施。在本文中，我将演示如何使用Ollama等服务在本地运行模型，并使用Qwen作为开源模型来高效且经济地执行特定任务，从而实现这种语义路由机制。

设置

下载和配置Ollama

为了使用多个LLM实现语义路由器，我们首先下载并设置Ollama，这是一个平台，允许轻松实例化和管理各种模型。Ollama提供了一个简单的接口来拉取和本地运行模型，这对于实时测试不同的模型非常理想。

第一步是使用以下命令安装Ollama：

!curl https://ollama.ai/install.sh | sh

这个脚本会自动在你的机器上安装Ollama。安装完成后，我们可以通过运行以下命令来启动Ollama服务器：

!ollama serve > server.log 2>&1 &

此命令会在后台启动Ollama服务器，服务器默认地址为127.0.0.1:11434。该服务器支持与模型的交互，对于管理和查询多个大型语言模型（LLM）至关重要。

接下来，我们将使用Ollama的pull命令下载所需的模型。agents_model是一个专门的LLM — qwen2.5:1.5b，我们将用它来完成特定任务。而router_model则是一个稍大的模型——qwen2.5:3b，它将作为语义路由器。这个模型会分析消息历史，并决定由哪个专用代理来处理特定任务。

重要的是要下载一个支持工具使用的模型。

%%capture
agents_model = "qwen2.5:1.5b"
!ollama pull {agents_model}
%%capture
router_model = "qwen2.5:3b"
!ollama pull {router_model}

拉取模型后，我们可以通过列出所有可用模型来验证它们是否成功下载：

!ollama ls

在不同端口上提供模型服务

为了高效管理多个Ollama模型，我们需要在每个单独的服务器实例上运行每个模型，并将每个实例绑定到不同的端口。这样，我们就可以根据任务或上下文将请求路由到适当的模型。

以下代码片段展示了如何为每个Ollama服务器实例设置不同的端口：

import os
os.environ['OLLAMA_HOST'] = "127.0.0.1:11438"
!ollama serve > server.log 2>&1 &
os.environ['OLLAMA_HOST'] = "127.0.0.1:11439"
!ollama serve > server.log 2>&1 &

在这里，我们设置OLLAMA_HOST环境变量来指定每个Ollama服务器的IP地址和端口号。通过导出不同的端口，我们可以在同一台机器上运行多个Ollama实例，每个实例服务一个不同的模型。在这个例子中，第一个模型运行在127.0.0.1:11438，第二个运行在127.0.0.1:11439。同时，请记住，还有一个Ollama实例运行在127.0.0.1:11434，我们将用它作为路由器。

这种设置在需要独立部署不同模型或测试不同配置的场景中特别有用。此外，这种方法模仿了微服务架构，其中每个模型都可以作为一个独立的服务来提供，通过RESTful API调用进行访问。在在线部署场景中，每台服务器可以托管一个不同的模型，并且可以根据当前任务动态地将请求路由到适当的模型。这为部署多个大型语言模型（LLM）提供了一个可扩展且灵活的解决方案，实现了高效的负载均衡、资源管理和跨不同系统的集成。

安装所需包

最后，为本教程安装所需的Python包。

%%capture
!pip install openai==1.57.4 ollama==0.4.4 pydantic==2.10.3

工具定义

在语义路由器的上下文中，工具是指大型语言模型（LLM）可以调用的函数或外部服务，以执行特定任务。这些工具使LLM能够高效地处理专门的操作，无论是总结文本、提取主题还是获取动态信息。在此设置中使用工具的关键优势在于，它们允许LLM专注于上下文分析和路由，同时将特定任务委托给轻量级、高效的函数。

在这个示例中，我定义了三种不同的工具：

1. 总结器工具：该工具负责总结一段文本。每当用户请求总结时，例如“总结这段文字”，就会调用此工具。
2. 主题提取器工具：该工具从给定的文本中提取主题，响应诸如“这段文字的主题是什么？”之类的查询。
3. 今日代理工具：该工具简单地返回当前是星期几。当用户问“今天是星期几？”时会使用此工具。

每个工具都被定义为一个字典，类型设置为“function”，并且包含函数的名称、描述和参数。以下是工具的结构：

from openai import OpenAI
import openai
summarizer_tool = {
    "type": "function",
    "function": {
        "name": "get_summarization",
        "description": "Call this method whenever you need to perform a summarization task, for example when a user asks 'Summarize this paragraph'. This method does not accepts input parameters.",
        "parameters": {
            "type": "object",
            "properties": {},
            "required": [],
            "additionalProperties": False
        }
    }
}
topics_tool = {
    "type": "function",
    "function": {
        "name": "get_topics",
      "description": "Call this whenever you need to perform a topic extraction extraction task, for example when a user asks 'What are the topics of this paragraph?'. This method does not accepts input parameters.",
      "parameters": {
          "type": "object",
          "properties": {},
          "required": [],
          "additionalProperties": False
      }
    }
}

today_tool = {
    "type": "function",
    "function": {
        "name": "get_current_day",
      "description": "Get the current day of the week. Call this whenever you need to know what day it is today, for example when the user asks 'What day is it?'. This method does not accepts input parameters.",
      "parameters": {
          "type": "object",
          "properties": {},
          "required": [],
          "additionalProperties": False
      }
    }
}

tools = [topics_tool, summarizer_tool, today_tool]

重要的是要注意，工具并不一定需要调用另一个大型语言模型（LLM）。在这个例子中，today_tool是一个简单的函数，它返回当前的星期几，而不需要涉及LLM。这种设计提供了极大的灵活性，因为你可以根据需要定义任意多的工具，每个工具执行特定的功能。

在语义路由器的上下文中，LLM可以分析消息的上下文，确定需要哪个工具，然后调用相应的工具。这种设置不仅通过将任务卸载到专门的函数来提高性能，而且还保持了过程的高效性和成本效益，因为我们并不总是需要为每个操作调用一个昂贵的模型。

代理定义

系统的核心围绕代理构建，代理负责执行特定任务，如总结、主题提取和提供当前星期几。这些代理通过兼容OpenAI API的机制与专门模型进行交互。通过利用OpenAI接口，我们可以轻松地与每个代理通信，并以简化的方式使用它们的功能。

在这个设置中，我们定义了三个代理：

1. 总结器代理：该代理负责总结用户提供的文本。它与总结模型进行交互。
2. 主题提取器代理：该代理从用户消息中提取主题。它使用一个为主题提取任务设计的通用模型。
3. 今日代理：该代理简单地返回当前星期几。

我们首先定义客户端以与代理进行交互。通过OpenAI接口，我们与在不同端口上运行的各自服务器上的代理进行通信。

from openai import OpenAIimport OpenAI
summarizer_client = OpenAI(
    base_url = 'http://localhost:11439/v1',
    api_key='ollama',  
)
general_client = OpenAI(
    base_url = 'http://localhost:11438/v1',
    api_key='ollama', 
)

为代理构建历史记录

为了维护一个代理可以引用的对话历史记录，我们定义了一个函数 build_history_agents。该函数接受一个消息列表和系统提示，为每个代理创建一个结构化的消息历史记录。

def build_history_agents(messages: list[dict[str, str]], sys_prompt: str):
    history = [{"role": "system", "content": sys_prompt}]
    
    for message in messages:
        if message["role"] != "system":
            history.append(message)
    return history

总结器代理

get_summarization 函数负责调用总结器代理来执行总结任务。代理被指示总结用户消息，如果用户请求的不是总结，代理只需回复“Pass.”（即“跳过。”）。

def get_summarization(history: list[dict[str, str]]):
    print("calling summarization model")
    sys_prompt = "You are a summarizer agent. Your only role is to summarize user message. When user asks for other action or want to chit-chat just answer 'Pass.'"
    response = summarizer_client.chat.completions.create(
        model=agents_model,
        messages=build_history_agents(history, sys_prompt),
    )
    return response.choices[0].message.content

主题提取器代理

类似地，get_topics 函数调用主题提取器代理。该函数旨在从用户消息中提取主题，并且，如果用户请求任何其他操作，代理会回复“Pass.”（即“跳过。”）。

def get_topics(history: list[dict[str, str]]):
    print("calling topic model")
    sys_prompt = "You are a topic extractor model. Your role is to examine user message and extract the topics. When user asks for other action or want to chit-chat just answer 'Pass.'"
    response = general_client.chat.completions.create(
        model=agents_model,
        messages=build_history_agents(history, sys_prompt),
    )
    return response.choices[0].message.content

今日代理

get_current_day 函数定义了今日代理，该代理不需要任何外部模型。它只需利用 Python 的 datetime 模块返回当前是星期几。

import datetime
def get_current_day():get_current_day():
    print("calling today model")
    return datetime.datetime.now().strftime("%A")

这种设置允许灵活地定义任意数量的代理，每个代理处理一个特定任务。代理可以是执行特定操作（如总结或主题提取）的简单函数，而无需额外的LLM。这使得系统既高效又成本效益高，因为只有在需要时才会调用专门的代理。此外，你可以轻松扩展代理列表以处理其他任务，如情感分析、问题回答，甚至复杂的多步骤过程。

精确的系统提示和工具描述的重要性

在使用LLM作为语义路由器的情况下，系统提示和工具描述的精确性在确保代理的有效性和准确性方面起着关键作用。系统提示作为对模型的指令，引导它们在明确定义的范围内执行特定任务。同样，工具描述需要清晰且具体，以便LLM能够调用正确的工具并理解其预期功能。

例如，在总结器代理的情况下，系统提示明确指示模型总结用户消息，并在遇到超出其范围的请求时回复“Pass”。如果没有这种精确性，代理可能会误解请求或采取不适当的行动。

路由器

在我们的方法中，路由器的作用是根据对话的上下文确定要调用哪个专门的代理或工具。路由器配备了处理函数调用、管理消息历史以及促进系统与相关工具或模型之间通信的实用程序。

实用程序

execute_function_call实用程序负责在需要特定工具时调用正确的函数。一旦路由器的响应指示需要调用工具，该函数就会检查需要调用哪个工具并执行相应的函数。例如，如果工具调用是为了获取当前日期，则会触发get_current_day函数，而对于其他任务（如总结或主题提取），则会调用相关的代理函数。

import json
import re
def execute_function_call(response, history):
  if response.choices[0].finish_reason == "tool_calls":
    tool_call = response.choices[0].message.tool_calls[0]
    function_name = tool_call.function.name
    if function_name == "get_current_day":
      result = globals()[function_name]()
    else:
      result = globals()[function_name](history)

    function_call_result_message = {
      "role": "tool",
      "content": json.dumps({
          "result": result
      }),
      "tool_call_id": response.choices[0].message.tool_calls[0].id
    }
    response_dict = response.model_dump()
    response_dict["choices"][0]["message"]
    history = history + [response_dict["choices"][0]["message"]] + [function_call_result_message]
    return history

在该实用程序中：

• 工具调用识别：系统从响应中的tool_calls中识别函数名称，并使用globals()动态调用正确的函数，无论是获取当前日期还是调用代理进行主题提取或总结。
• 函数执行：根据函数名称，执行相应的函数（get_current_day、get_summarization或get_topics），并捕获其结果。
• 消息历史更新：将响应消息和工具调用结果添加到对话历史中，确保路由系统能够跟踪对话和执行的函数。

此外，build_router_history实用程序用于管理对话历史，通过将助手的响应追加到消息历史中。该实用程序确保历史准确反映对话的流程，包括系统采取的任何操作（例如，调用总结器或主题提取器）。

def build_router_history(messages: list[dict[str, str]], response):
    res = response.choices[0].message.content
    messages.append(
        {
            "role": "assistant",
            "content": res
        }
    )
    print(res)
    return messages

消息历史管理的重要性

管理消息历史至关重要，因为它为后续的交互提供了上下文。每次路由器接收到新消息并处理函数调用时，它都会将相关详细信息追加到历史记录中。这确保了路由器始终能够访问对话的完整上下文，从而能够就调用哪个工具或代理做出更好的决策。

例如，当用户请求总结一段文字时，系统会检查整个消息历史记录，其中可能包括之前的请求、响应甚至工具调用，以决定适当的操作。这种基于历史的方法确保了路由器保持上下文感知和高效，减少了错误和不必要的资源消耗。

通过维护全面的历史记录并使用动态函数调用，路由器变得高度适应，并且能够处理各种各样的任务，每个任务都分配给可用的最专业且最具成本效益的模型。

路由器的定义

路由器是此系统的关键组件，负责根据上下文将消息定向到适当的专门工具或代理。路由器通过分析对话历史记录并在必要时调用正确的函数来工作。以下是定义路由器及其客户端的代码：

router_client = OpenAI(
    base_url = 'http://localhost:11434/v1',
    api_key='ollama', # required, but unused
)

我们还可以定义一个实用函数来与路由器进行交互。

def call_router(messages: list[dict[str, str]], use_tools=True):
  response = router_client.chat.completions.create(
    model=router_model,
    messages=messages,
    tools=tools if use_tools else None
  )
  return response

与助手聊天：路由实战

一旦一切设置就绪，就是时候看看路由机制的实际效果了。基本思路是，路由器模型将分析对话历史和用户意图，决定是将对话转交给专门的代理（如总结器或主题提取器），还是继续由助手进行回应。

以下是我们如何设置第一个测试用例，其中我们希望路由器将用户的请求重定向到总结模型：

sys_prompt = "You are a routing model. Your role is to understand the chat history and user intent and eventualy to redirect the messages to the correct model that will handle the conversation. Your are allowed to chat with the user but only for chit-chatting, otherwise if you are asked to perform specific tasks just call other functions."
messages = [
  {
      "role": "system",
      "content": sys_prompt
  },
  {
      "role": "user",
      "content": "Hi, can you summarize this paragraph? The red glow of tail lights indicating another long drive home from work after an even longer 24-hour shift at the hospital. The shift hadn’t been horrible but the constant stream of patients entering the ER meant there was no downtime. She had some of the “regulars” in tonight with new ailments they were sure were going to kill them. It’s amazing what a couple of Tylenol and a physical exam from the doctor did to eliminate their pain, nausea, headache, or whatever other mild symptoms they had. Sometimes she wondered if all they really needed was some interaction with others and a bit of the individual attention they received from the nurses."
  }
]

在这段代码中：

• 系统提示：sys_prompt设定了路由器的行为。它解释了路由器的主要职责是理解对话并将任务分配给适当的模型，但路由器也允许进行轻松的对话。
• 用户输入：用户要求路由器总结一段文字，这是一项需要总结代理执行的特定任务。

现在，让我们调用路由器来处理这个请求：

response = call_router(messages)

分析我们获得的响应：

ChatCompletion(id='chatcmpl-4', choices=[Choice(finish_reason='tool_calls', index=0, logprobs=None, message=ChatCompletionMessage(content='', refusal=None, role='assistant', audio=None, function_call=None, tool_calls=[ChatCompletionMessageToolCall(id='call_96uqu57o', function=Function(arguments='{"text":"The red glow of tail lights indicating another long drive home from work after an even longer 24-hour shift at the hospital. The shift hadn’t been horrible but the constant stream of patients entering the ER meant there was no downtime. She had some of the \'regulars\' in tonight with new ailments they were sure were going to kill them. It\'s amazing what a couple of Tylenol and a physical exam from the doctor did to eliminate their pain, nausea, headache, or whatever other mild symptoms they had. Sometimes she wondered if all they really needed was some interaction with others and a bit of the individual attention they received from the nurses."}', name='get_summarization'), type='function', index=0)]))], created=1735233239, model='qwen2.5:3b', object='chat.completion', service_tier=None, system_fingerprint='fp_ollama', usage=CompletionUsage(completion_tokens=152, prompt_tokens=502, total_tokens=654, completion_tokens_details=None, prompt_tokens_details=None))

从finish_reason='tool_calls'可以看出，路由器实际上调用了执行总结任务的工具。

现在我们可以执行路由器告诉我们的函数，代码如下：

# Execute the function call and update the messages
messages = execute_function_call(response, messages)
# Call the router again with the updated messages, this time without using any tools (since the task is complete)
response = call_router(messages, use_tools=False)
# Build the final history of the conversation including the router's response
messages = build_router_history(messages, response)

总结模型的输出如下，太棒了！

The summary of the paragraph is:
A person reflects on their 24-hour shift at the hospital, noting that while it hasn't been terrible overall—constantly dealing with patient influxes without downtime—they often have repeat patients who come back after various minor issues (such as pain and nausea). These issues are typically resolved through medication (Tylenol) along with physical examinations from doctors. The person reflects on how the shifts remind them of how simple interactions and more personalized care can resolve many minor problems associated with being alone in the Emergency Room all day long.

现在我们再次测试路由器，这次我们想知道今天是星期几。我更新了历史记录并再次调用路由器。

messages.append({append({
      "role": "user",
      "content": "Now I want to know what day is it"
  }
)
response = call_router(messages)
print(response)

再次，路由器的表现符合预期，因为通过分析响应，我们看到它调用了正确的工具。

ChatCompletion(id='chatcmpl-650', choices=[Choice(finish_reason='tool_calls', index=0, logprobs=None, message=ChatCompletionMessage(content='', refusal=None, role='assistant', audio=None, function_call=None, tool_calls=[ChatCompletionMessageToolCall(id='call_ppy9ohoc', function=Function(arguments='{}', name='get_current_day'), type='function', index=0)]))], created=1735233249, model='qwen2.5:3b', object='chat.completion', service_tier=None, system_fingerprint='fp_ollama', usage=CompletionUsage(completion_tokens=17, prompt_tokens=757, total_tokens=774, completion_tokens_details=None, prompt_tokens_details=None))

我们可以用上面的相同代码再次执行该函数：

messages = execute_function_call(response, messages)
response = call_router(messages, use_tools=False)
messagges = build_router_history(messages, response)
# the response this time is Today is Thursday.

最后，再次测试路由器以验证最后一个工具。

messages.append({
      "role": "user",
      "content": "Now can you extract topics from the paragraph that I sent you on the first message?"
  }
)
response = call_router(messages)
messages = execute_function_call(response, messages)
response = call_router(messages, use_tools=False)
messagges = build_router_history(messages, response)

响应是：

The main topics from the paragraph are:
1. A 24-hour shift at the hospital's Emergency Room (ER)
2. The constant influx of patients during the shift
3. The treatments provided (Tylenol, physical exams)
4. Repeat patients with various minor issues
5. Reflection on personal care and interactions needed for better patient outcomes

太好了，一切都按计划进行。所以我们感谢人工智能的出色工作。

messages.append({
      "role": "user","role": "user",
      "content": "Thank you!! You are a very helpful assistant. Now, say hello to our Medium friends :)"
  }
)
response = call_router(messages)
messagges = build_router_history(messages, response)
# response: Hello to our Medium friends! How can I assist you further?

结论

本文通过使用具有函数调用功能的大型语言模型（LLM）作为语义路由器，展示了一种管理复杂对话流程和信息处理的强大方法。在这些示例中，我使用了像Qwen-2 3B这样的小型模型，但也可以采用更强大的模型。