使用CrewAI的并发查询解决系统

在人工智能时代，企业不断寻求创新方法来提升客户支持服务。其中一种方法就是利用协同工作的人工智能代理来高效解决客户查询。本文探讨了如何利用 CrewAI、OpenAI 的 GPT 模型和 Google Gemini 实现并发查询解决系统。该系统采用多个并行操作的专业代理，无缝处理客户查询，缩短响应时间并提高准确性。

• 了解人工智能代理如何通过自动回复和总结关键信息来高效处理客户查询。
• 了解 CrewAI 如何实现多代理协作以改进客户支持工作流程。
• 探索不同类型的人工智能代理（如查询解析器和汇总器）及其在客户服务自动化中的作用。
• 使用 Python的 asyncio 实现并发查询处理，以提高响应效率。
• 通过集成人工智能驱动的自动化来优化客户支持系统，从而提高准确性和可扩展性。

AI代理如何协同工作？

并发查询解决系统采用多代理框架，为每个代理分配特定的角色。该系统采用了 CrewAI，这是一个能让人工智能代理有效协作的框架。

系统的主要组成部分包括

• 查询解决代理：负责了解客户询问并提供准确答复。
• 摘要代理：总结解决过程，以便快速审查和将来参考。
• LLMs （大型语言模型）：包括 GPT-4o 和 Gemini 等模型，每个模型都有不同的配置，以平衡速度和准确性。
• 任务管理：向代理动态分配任务，确保并发查询处理。

并发查询解决系统的实现

要将人工智能代理框架从概念转化为现实，结构化的实施方法至关重要。下面，我们将概述为有效解决查询问题而设置和集成人工智能代理所涉及的关键步骤。

步骤 1：设置 API 密钥

OpenAI API 密钥使用 os 模块存储为环境变量。这样，系统就可以安全地验证 API 请求，而无需硬编码敏感凭据。

import os 
# Set the API key as an environment variable
os.environ["OPENAI_API_KEY"] = ""

系统使用 os 模块与操作系统交互。

系统将 OPENAI_API_KEY 设置为环境变量，以便对 OpenAI 的 API 请求进行验证。

步骤 2：导入所需库函数

导入必要的库，包括用于处理异步操作的 asyncio 和 crewai 组件（如Agent、Crew、Task 和 LLM）。这些对于定义和管理人工智能代理至关重要。

import asyncio
from crewai import Agent, Crew, Task, LLM, Process
import google.generativeai as genai

• asyncio：Python 用于异步编程的内置模块，支持并发执行。
• Agent：代表具有特定职责的人工智能工作者。
• Crew：管理多个代理及其交互。
• Task：定义每个代理应该做什么。
• LLM：指定使用的大型语言模型。
• Process：定义任务的执行方式，是顺序执行还是并行执行。
• google.generativeai： 用于使用谷歌生成式人工智能模型的库（本片段未使用，但可能包含在内以备将来扩展之用）。

步骤 3：初始化LLM

以不同的温度设置初始化三个不同的 LLM 实例（GPT-4o 和 GPT-4）。温度控制响应的创造性，确保人工智能生成答案的准确性和灵活性之间的平衡。

# Initialize the LLM with Gemini
llm_1 = LLM(
model="gpt-4o",
temperature=0.7)
llm_2 = LLM(
model="gpt-4",
temperature=0.2)
llm_3 = LLM(
model="gpt-4o",
temperature=0.3)系统会创建三个 LLM 实例，每个实例都有不同的配置。

参数：

• model： 指定要使用的 OpenAI 模型（gpt-4o 或 gpt-4）。
• temperature： 控制响应的随机性（0 = 确定性，1 = 更有创造力）。

这些不同的模型和温度有助于平衡准确性和创造性。

步骤 4：定义人工智能代理

每个代理都有特定的角色和预定义的目标。我们创建了两个人工智能代理：

• Query Resolver：处理客户查询并提供详细回复。
• Summary Generator：每个代理都有确定的角色、目标和背景故事，以指导其互动。

Query Resolver代理

query_resolution_agent = Agent(
llm=llm_1,
role="Query Resolver",
backstory="An AI agent that resolves customer queries efficiently and professionally.",
goal="Resolve customer queries accurately and provide helpful solutions.",
verbose=True
)

让我们看看这个代码块中发生了什么

• Agent Creation：query_resolution_agent 是一个人工智能助手，负责解决客户的询问。
• Model Selection：它使用 llm_1，配置为 GPT-4o，温度为 0.7。这种平衡可以做出既有创意又准确的回复。
• Role：系统指定该代理为查询解决者。
• Backstory：开发人员对代理进行编程，使其充当专业的客户服务助理，确保高效、专业的回复。
• Goal：为用户查询提供准确的解决方案。
• Verbose Mode：verbose=True 可确保记录详细日志，帮助开发人员调试和跟踪其性能。

Summary代理

summary_agent = Agent(
llm=llm_2,
role="Summary Generator",
backstory="An AI agent that summarizes the resolution of customer queries.",
goal="Provide a concise summary of the query resolution process.",
verbose=True
)

代码块包括：

• Agent Creation：summary_agent 用于汇总查询结果。
• Model Selection：使用 temperature 为 0.2 的 llm_2 (GPT-4)，使其响应更加确定和精确。
• Role：该代理充当摘要生成器。
• Backstory：它能简明扼要地总结查询决议，以便快速参考。
• Goal：它能简明扼要地总结客户查询的解决情况。
• Verbose Mode：verbose=True 可确保在需要时提供调试信息。

步骤 5：定义任务

系统会动态分配任务，以确保并行查询处理。

本节定义了并行查询解决系统中分配给人工智能代理的任务。

resolution_task = Task(
description="Resolve the customer query: {query}.",
expected_output="A detailed resolution for the customer query.",
agent=query_resolution_agent
)
summary_task = Task(
description="Summarize the resolution of the customer query: {query}.",
expected_output="A concise summary of the query resolution.",
agent=summary_agent
)

代码说明

定义任务：

• resolution_task： 该任务指示查询解决代理分析和解决客户查询。
• summary_task： 该任务指示摘要代理生成解决过程的简要摘要。

动态查询处理：

• 执行任务时，系统会用实际客户查询替换 {query}。
• 这样，系统就可以动态处理任何客户查询。

预期输出：

• resolution_task 希望得到对查询的详细回复。
• summary_task 生成查询解决的简明摘要。

代理分配：

• 分配 query_resolution_agent 处理解决任务。
• summary_agent 用于处理摘要任务。

重要性

• 任务专业化：每个人工智能代理都有特定的工作，从而确保效率和清晰度。
• 可扩展性：您可以添加更多任务和代理，以处理不同类型的客户支持交互。
• 并行处理：任务可同时执行，减少客户等待时间。

步骤 6：使用人工智能代理执行查询

创建一个异步函数来处理查询。Crew 类组织代理和任务，按顺序执行，以确保正确的查询解析和汇总。

async def execute_query(query: str):
crew = Crew(
agents=[query_resolution_agent, summary_agent],
tasks=[resolution_task, summary_task],
process=Process.sequential,
verbose=True
)
result = await crew.kickoff_async(inputs={"query": query})
return result

该函数定义了一个执行查询的异步流程。它创建一个 Crew 实例，其中包括

• agents：参与流程的人工智能代理（查询解析器和摘要生成器）。
• tasks：分配给代理的任务（查询解析和摘要生成）。
• process=Process.sequential：确保任务按顺序执行。
• verbose=True：启用详细日志，以便更好地跟踪。

该函数使用 await 异步执行人工智能代理并返回结果。

步骤 7：同时处理多个查询

使用 asyncio.gather()，可以同时处理多个查询。这样，人工智能代理就可以并行处理不同的客户问题，从而缩短响应时间。

async def handle_two_queries(query_1: str, query_2: str):
# Run both queries concurrently
results = await asyncio.gather(
execute_query(query_1),
execute_query(query_2)
)
return results

asyncio.gather() 同时处理两个查询，大大缩短了响应时间。执行完成后，该函数将返回两个查询的结果

步骤 8：定义示例查询

开发人员定义示例查询来测试系统，涵盖常见的客户支持问题，如登录失败和支付处理错误。

query_1 = "I am unable to log in to my account. It says 'Invalid credentials', but I am sure I am using the correct username and password."
query_2 = "The payment gateway is not working. Also, a weird error message is displayed. My card has been charged, but the transaction is not going through."

这些是测试系统的示例查询。

查询 1 涉及登录问题，查询 2 则与支付网关错误有关。

步骤 9：设置事件循环

系统会初始化一个事件循环，以处理异步操作。如果找不到现有的循环，系统会创建一个新的循环来管理 AI 任务的执行。

try:
loop = asyncio.get_event_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

本节确保事件循环可用于运行异步任务。

如果系统检测到没有事件循环（发生 RuntimeError），则会创建一个新循环并将其设置为活动循环。

步骤 10： 在 Jupyter Notebook/Google Colab 中处理事件循环

由于 Jupyter 和 Colab 都有预先存在的事件循环，nest_asyncio.apply()可用于防止冲突，确保异步查询的顺利执行。

# Check if the event loop is already running
if loop.is_running():
# If the loop is running, use `nest_asyncio` to allow re-entrant event loops
import nest_asyncio
nest_asyncio.apply()

Jupyter Notebooks 和 Google Colab 预先存在事件循环，在运行异步函数时可能会导致错误。

nest_asyncio.apply()允许嵌套事件循环，从而解决了兼容性问题。

步骤 11：执行查询并打印结果

事件循环运行handle_two_queries()，并发处理查询。系统会打印人工智能生成的最终响应，显示查询结果和摘要。

# Run the async function
results = loop.run_until_complete(handle_two_queries(query_1, query_2))
# Print the results
for i, result in enumerate(results):
print(f"Result for Query {i+1}:")
print(result)
print("\n---\n")

loop.run_until_complete() 开始执行 handle_two_queries()，同时处理两个查询。

系统会打印结果，显示人工智能为每个查询生成的决议。

并发查询解析系统的优势

下面，我们将了解并发查询解析系统如何通过同时处理多个查询来提高效率，从而加快响应时间并改善用户体验。

• 更快的响应时间：并行执行可同时解决多个查询。
• 提高准确性：利用多个 LLM 可确保创造性和事实正确性之间的平衡。
• 可扩展性：系统可处理大量查询，无需人工干预。
• 更好的客户体验：自动摘要可提供查询解决方案的快速概览。

并发查询解决系统的应用

下面我们将探讨并发查询解决系统的各种应用，包括客户支持自动化、聊天机器人中的实时查询处理以及大规模服务请求的高效处理。

• 客户支持自动化：使人工智能驱动的聊天机器人能够同时解决多个客户查询，从而缩短响应时间。
• 实时查询处理：通过并行处理大量查询，提高即时支持系统的效率。
• 电子商务协助：简化在线购物平台中的产品查询、订单跟踪和支付问题解决方案。
• IT 服务台管理：通过同时诊断和解决多个技术问题，为 IT 服务台提供支持。
• 医疗保健与远程医疗：协助同时管理病人咨询、预约安排和医疗建议。

小结

并发查询解决系统展示了人工智能驱动的多代理协作如何彻底改变客户支持。通过利用 CrewAI、OpenAI 的 GPT 模型和 Google Gemini，企业可以自动处理查询，提高效率和用户满意度。这种方法为未来更先进的人工智能驱动服务解决方案铺平了道路。

• 人工智能代理可简化客户支持，缩短响应时间。
• CrewAI 使专业代理能够有效地协同工作。
• 使用 asyncio 可同时处理多个查询。
• 不同的 LLM 配置可兼顾准确性和创造性。
• 系统可管理大量查询，无需人工干预。
• 自动摘要提供快速、清晰的查询解决方案。