在上一篇文章中，我介绍了语义路由器：一种使 AI 代理能够为正确的任务选择正确的 LLM 的模式任务，同时还减少了他们对大语言模型的依赖。在本后续教程中，我们将使用语义路由器项目通过选择检索信息的最佳方式来智能地处理用户查询，例如是否使用矢量数据库和/或基于工具的实时数据检索器。
与之前的教程类似，在我们的示例中，我们将使用来自 FlightAwares AeroAPI 的数据实时跟踪飞机的飞行状态。
首先，请注意，路由器基于根据意图，确保检索最相关的上下文，使这种方法独一无二。语义路由器采用 OpenAIs LLM 和结构化检索方法，并将它们结合起来，形成一个自适应、高度响应的助手，可以快速处理会话查询和特定于数据的请求。
语义路由器建议调用该工具来查询以下内容：航班时刻表和状态，同时它将有关行李政策的查询路由到提供上下文的搜索功能。
让我们逐步分解它。
在深入研究代码之前，请确保您已安装所需的库。您可以使用 pip 来执行此操作：
1
pip install openai chromadb 请求 pytz 语义路由器
让我们导入所需的模块。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
导入操作系统
从输入导入Dict，Any
从日期时间导入日期时间，时间增量
导入pytz
导入请求
从openai导入OpenAI
导入chromadb
从chromadb.utils导入embedding_functions
从semantic_router导入Route，RouteLayer
p>
从semantic_router.encoders导入OpenAIEncoder
从semantic_router.llms.openai导入get_schemas_openai
从semantic_router.llms导入OpenAILLM
接下来，为 OpenAI 和 FlightAware AeroAPI 密钥设置环境变量：
1
2
导出 OPENAI_API_KEY="your_openai_api_key"
导出 AEROAPI_KEY="your_flightaware_api_key"
这些密钥对于访问教程中使用的外部服务至关重要。
我们首先初始化 OpenAI 和 ChromaDB 的客户端。OpenAI 将为我们的查询生成嵌入，而 ChromaDB 将存储和检索行李政策等上下文数据的嵌入。
1
2
3
4
5
6
7
8
9
10
11
12
# 初始化 OpenAI 客户端
openai_client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
# 使用持久存储路径初始化 ChromaDB 客户端
chroma_client = chromadb.PersistentClient(path="./chroma_db")
# 使用 OpenAI 模型设置嵌入函数
embedding_function = embedding_functions.OpenAIEmbeddingFunction(
api_key=os.environ["OPENAI_API_KEY"],
model_name="text-embedding-3-small"
)
COLLECTION_NAME =“行李政策”
FlightAwares AeroAPI 提供实时航班数据。函数
get_flight_context检索特定航班 ID 的航班信息，包括出发、到达时间和状态。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
16
17
18
19
p>20
21
22
23
24
25
26
AEROAPI_BASE_URL = "https://aeroapi.flightaware.com/aeroapi"
AEROAPI_KEY = os.getenv("AEROAPI_KEY")
def get_flight_context(flight_id:str）->；str:
def _get_flight_data() -gt;字典：
session = requests.Session()
session.headers.update({"x-apikey": AEROAPI_KEY})
start_date = datetime.now().strftime('%Y-%m-%d')
end_date = (datetime.now() + timedelta(days=1)).strftime('%Y-%m-%d')
response = session.get(f"{AEROAPI_BASE_URL}/flights/{flight_id}?start={start_date}amp;end={end_date}")
response.raise_for_status()
返回response.json()['航班'][0]
def _utc_to_local(utc_date_str: str, local_timezone_str: str) -gt;str:
utc_datetime = datetime.strptime(utc_date_str, '%Y-%m-%dT%H:%M:%SZ').replace(tzinfo=pytz.utc)
local_timezone = pytz.timezone(local_timezone_str)
local_datetime = utc_datetime.astimezone(local_timezone)
return local_datetime.strftime('%Y-%m-%d %H:%M:%S')
flight_data = _get_flight_data()
depart_time = _utc_to_local(flight_data['scheduled_out'], Flight_data['origin']['timezone'])
arrival_time = _utc_to_local(flight_data['scheduled_in'], Flight_data['destination']['timezone'])
返回 (
f"航班 {flight_id}从 {flight_data['origin']['city']} 飞往 {flight_data['destination']['city']} “
f”于 {depart_time} 出发，并于 {arrival_time} 抵达{flight_data['status']} 的状态。"
)
此函数从 AeroAPI 获取航班数据，并将 UTC 时间转换为出发和到达机场的当地时区，作为大语言模型提供有关航班时刻表的实时信息的上下文。
接下来，我们定义一个查询行李政策信息的方法。这些信息存储在 ChromaDB（一个矢量数据库）中，我们可以根据用户输入使用嵌入来查询它。
1
2
3
4
5
6
def get_baggage_context(query: str) -gt;str:
collection = chroma_client.get_collection(name=COLLECTION_NAME, embedding_function=embedding_function)
结果 = collection.query(query_texts=[query], n_results=3)
if results 和 results['documents']:
return " ".join(results['documents'][0])
return "未找到相关行李信息."
在这里，我们通过搜索 ChromaDB 集合，根据用户查询获取相关行李政策信息。
现在，我们将使用 OpenAIGPT-40-mini 用于生成包含上下文（航班状态或行李政策）的响应。
1
2
3
4
5
6
7
8
9
def get_llm_response(查询: str, 上下文: str) -gt;str:
response = openai_client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "您是一位有用的航空公司助理。根据提供的上下文回答用户查询。"},
{"role": "user", "content": f"查询: {query}\nContext: {context}"},
],
)
返回响应。选择[0].message.content
语言模型接收用户查询和上下文（即航班状态或行李政策）并生成响应。
让我们为行李建立索引ChromaDB 中的策略规则，以便我们可以在需要时查询它们。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
16
17
18
19
p>def index_baggage_policy():
baggage_rules = [
“阿联酋航空提供宽松的行李政策，该政策根据航线、票价类型和舱位而有所不同。”,
“经济舱乘客可携带一件随身行李，重量不超过 7 公斤，尺寸为 55 x 38 x 20 厘米。”,
# ...更多规则
]
如果 COLLECTION_NAME 不在 [chroma_client.list_collections() 中 col 的 col.name] 中：
collection = chroma_client.create_collection(
name=COLLECTION_NAME,
embedding_function=embedding_function,
metadata={"hnsw:space": "cosine"}
)
对于 idx，enumerate(baggage_rules) 中的规则：
collection.add(documents=[rule], ids=[f"baggage_rule_{idx}"])
print(f"在 ChromaDB 中存储了 {len(baggage_rules)} 行李规则。")
其他：
collection = chroma_client.get_collection(name=COLLECTION_NAME)
返回集合
为了增强用户体验，我们设置了一个路由器，可以智能地确定查询是否与航班、行李或其他对话任务，例如笑话或诗歌。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def setup_router():
encoder = OpenAIEncoder()
flight_info = Route(
name="flight_info",
utterances=["我的航班状态如何？", "我的航班什么时候起飞？"],
function_schemas=get_schemas_openai([get_flight_context])
)
baggage_policy = Route(
name="baggage_policy",
utterances=["行李限额是多少？", "我可以携带多少件行李？"],
function_schemas=get_schemas_openai([get_baggage_context])
)
聊天 = 路线(
name="chat",
utterances=["写一首诗","给我讲个笑话"]
)
llm = OpenAILLM()
返回RouteLayer（编码器，路线= [flight_info，baggage_policy，chat]，llm = llm）
这一步是最关键的。它设置一个语义路由器，根据意图智能地将用户查询路由到适当的功能。它定义了航班信息、行李政策和一般对话的路线。每条路线都将特定的话语链接到函数，使用 OpenAIEncoder 来理解查询上下文。然后，路由器确定查询是否需要来自 ChromaDB 的航班数据和行李详细信息，或者会话响应，以确保系统内正确的处理程序进行准确、高效的处理。
请注意，我们将三个路线映射到可能的用户查询。第一个路由通过将函数和参数映射到
get_flight_context函数来映射到 FlightAware API。第二条路线有与行李政策相关的话语，它指向负责从矢量数据库检索数据的
get_baggage_context函数。第三条路线没有与之相关的功能；它充当包罗万象的路由。
最后，我们通过路由器处理用户查询并提供适当的响应。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
def process_query(query: str, router_layer: RouteLayer):
response = router_layer(query)
context = "未找到相关上下文。"
if response.function_call:
用于在response.function_call中调用：
if call["function_name"] == "get_flight_context":
context = get_flight_context(**call["arguments"])
elif call["function_name"] == "get_baggage_context":
context = get_baggage_context(**call["arguments]"])
llm_response = get_llm_response(query, context)
print(f"查询：{query}")
print(f"上下文：{上下文}")
print(f"LLM 响应：{llm_response}\n")
这是将所有内容联系在一起的主要功能。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def main():
index_baggage_policy()
router_layer = setup_router()
查询= [
“EK524 航班的状态如何？”、
“随身行李的尺寸限制是多少？”、
“写一首关于猫的诗。”
]
对于查询中的查询：
process_query(query, router_layer)
if __name__ == "__main__":
main()
以下是 GitHub Gist 中提供的完整代码。