category
作者：Andre Boaventura 发布于 2025 年 3 月 6 日 于
Amazon Bedrock, Amazon Machine Learning, 基础模型, 生成式AI, 技术教程
智能体（Agent）正在革命性地改变生成式 AI 的格局，充当着大型语言模型（LLM）与现实世界应用之间的桥梁。这些智能、自主的系统即将成为各行业采用 AI 的基石，预示着人机协作和问题解决的新时代。通过利用 LLM 的能力并将它们与专业工具和 API 结合，智能体能够处理传统 AI 系统先前无法解决的复杂多步骤任务。本文演示的多智能体城市信息系统（Multi-Agent City Information System）例证了基于智能体的架构创造复杂、适应性强且功能强大的 AI 应用潜力。
展望未来，智能体将在以下方面发挥极其重要的作用：
- 利用更深层次、具有情境感知的信息改善决策
- 自动化跨多个领域的复杂工作流程，从客户服务到科学研究
- 实现更自然、直观的人机交互
- 通过整合多样化数据源和专业知识来产生新想法
- 通过提供更透明、可解释的 AI 系统来解决伦理问题
构建和部署像本文中这样的多智能体系统是迈向释放生成式 AI 全部潜力的关键一步。随着这些系统的发展，它们将改变行业、拓展可能性并为人工智能打开新的大门。
解决方案概述
在本文中，我们探讨了如何利用 Amazon Bedrock 上的 LangGraph 和 Mistral 模型来创建一个强大的多智能体系统，该系统能通过协作问题处理来解决复杂的工作流。这种集成方式使得创建能够协作解决复杂问题的 AI 智能体成为可能，模拟人类的推理和协作过程。
其成果是一个可以为指定城市提供有关活动、天气、活动和推荐等全面详情的系统，展示了如何在 Amazon Web Services (AWS) 上构建和部署有状态的多智能体应用以应对现实世界的挑战。
LangGraph 通过提供一种组织良好的方法来定义和管理智能体之间的信息流，对实现我们的解决方案至关重要。它内置支持状态管理和检查点功能，确保了流程的连续性。该框架还允许轻松可视化智能体工作流，提高清晰度和可理解性。它能轻松与 LLM 和 Amazon Bedrock 集成，提供一个多功能且强大的解决方案。此外，其对条件路由的支持允许基于中间结果动态调整工作流，在处理不同场景时提供灵活性。
我们提出的多智能体架构具有以下几个关键优势：
- 模块化 – 每个智能体专注于特定任务，使系统更易于维护和扩展
- 灵活性 – 可以快速添加、移除或修改智能体，而不会影响整个系统
- 复杂工作流处理 – 通过将任务分配给多个智能体，系统能够管理高级复杂的工作流
- 专业化 – 每个智能体针对其特定任务进行了优化，从而改善了延迟、准确性并提高了系统整体效率
- 安全性 – 通过确保每个智能体只能访问其任务所需的工具来增强安全性，降低了未经授权访问敏感数据或其他智能体任务的潜在风险
我们的多智能体系统如何工作
在本节中，我们将探讨我们的多智能体城市信息系统如何工作，该系统基于 GitHub 上 Mistral on AWS examples for Bedrock & SageMaker 仓库中提供的多智能体 LangGraph Mistral Jupyter notebook。
该智能体工作流接收城市名称作为输入并提供详细信息，展示了在处理不同场景时的适应性：
- 活动（Events） – 它在本地数据库和在线来源中搜索该城市即将举办的活动。每当本地数据库信息不可用时，它会触发使用 Tavily API 的在线搜索。这确保了无论信息是存储在本地还是需要从网络检索，用户都能获得最新的活动信息。
- 天气（Weather） – 系统使用 OpenWeatherMap API 获取当前天气数据，为查询地点提供准确及时的天气信息。基于天气情况，系统还提供相应的着装和活动建议，为每个城市提供相关推荐。
- 餐厅（Restaurants） – 推荐通过检索增强生成（RAG）系统提供。该方法结合预存储信息和实时生成来提供相关且最新的餐饮建议。
该系统处理不同信息水平的能力通过其自适应方法得以体现，这意味着无论不同城市的数据可用性如何变化，用户都能获得尽可能全面和最新的信息。例如：
- 某些城市在本地数据库数据不可用时，可能需要使用搜索工具来查找活动信息。
- 其他城市可能已有数据存储在本地数据库中，无需在线搜索即可快速获取活动信息。
-
如果特定城市的餐厅推荐不可用，系统仍能基于可用的事件和天气数据提供有价值的见解。
数据来源
多智能体城市信息系统可以利用两个数据源。
- 本地活动数据库（Local events database）
这是一个 SQLite 数据库，由 JSON 文件填充了城市活动数据，提供对本地活动信息的快速访问，包括社区活动、文化活动以及全市范围的活动等。该数据库由 events_database_tool() 使用，用于高效查询和检索城市活动详情，包括地点、日期和活动类型。
- 餐厅 RAG 系统（Restaurant RAG system）
对于餐厅推荐，generate_restaurants_dataset() 函数生成合成数据，创建一个专门为我们推荐系统定制的数据集。create_restaurant_vector_store() 函数处理这些数据，使用 Amazon Titan Text Embeddings 生成嵌入向量，并使用 Facebook AI Similarity Search (FAISS) 构建向量存储。虽然这种方法适用于原型开发，但对于更具可扩展性和企业级的解决方案，我们建议使用 Amazon Bedrock Knowledge Bases。
构建多智能体架构
我们多智能体城市信息系统的核心在于一组专门设计的函数和工具，用于从各种来源收集、处理和合成信息。它们构成了我们系统的支柱，使其能够提供有关城市的全面且最新的信息。在本节中，我们将深入探讨驱动我们系统的关键组件：使用 Mistral 模型的 generate_text() 函数，以及用于本地数据库查询、在线搜索、天气信息和餐厅推荐的专用数据检索函数和工具。这些函数和工具共同创建了一个强大且多功能的系统，能够为用户提供有价值的洞察。
文本生成函数（Text generation function）
该函数是我们智能体的核心，使它们能够在需要时使用 Mistral 模型生成文本。它利用了支持文本生成、流式传输和外部函数调用（工具）的 Amazon Bedrock Converse API。该函数工作流程如下：
1.使用 Amazon Bedrock Converse API 将用户消息发送给 Mistral 模型。
2.调用适当的工具并将结果整合到对话中。
3.持续对话直到生成最终响应。
def generate_text(bedrock_client, model_id, tool_config, input_text):
......
while True:
response = bedrock_client.converse(**kwargs)
output_message = response['output']['message']
messages.append(output_message) # Add assistant's response to messages
stop_reason = response.get('stopReason')
if stop_reason == 'tool_use' and tool_config:
tool_use = output_message['content'][0]['toolUse']
tool_use_id = tool_use['toolUseId']
tool_name = tool_use['name']
tool_input = tool_use['input']
try:
if tool_name == 'get_upcoming_events':
tool_result = local_info_database_tool(tool_input['city'])
json_result = json.dumps({"events": tool_result})
elif tool_name == 'get_city_weather':
tool_result = weather_tool(tool_input['city'])
json_result = json.dumps({"weather": tool_result})
elif tool_name == 'search_and_summarize_events':
tool_result = search_tool(tool_input['city'])
json_result = json.dumps({"events": tool_result})
else:
raise ValueError(f"Unknown tool: {tool_name}")
tool_response = {
"toolUseId": tool_use_id,
"content": [{"json": json.loads(json_result)}]
}
......
messages.append({
"role": "user",
"content": [{"toolResult": tool_response}]
})
# Update kwargs with new messages
kwargs["messages"] = messages
else:
break
return output_message, tool_result
(具体 Python 代码实现位于原文相应位置)
本地数据库查询工具（Local database query tool）
events_database_tool() 函数通过连接到 SQLite 数据库、执行查询以获取指定城市的即将举办活动，并将结果格式化为字符串返回，来查询本地活动信息。它由 events_database_agent() 函数使用。
def events_database_tool(city: str) -> str:
conn = sqlite3.connect(db_path)
query = """
SELECT event_name, event_date, description
FROM local_events
WHERE city = ?
ORDER BY event_date
LIMIT 3
"""
df = pd.read_sql_query(query, conn, params=(city,))
conn.close()
print(df)
if not df.empty:
events = df.apply(
lambda row: (
f"{row['event_name']} on {row['event_date']}: {row['description']}"
),
axis=1
).tolist()
return "\n".join(events)
else:
return f"No upcoming events found for {city}."(具体 Python 代码实现位于原文相应位置)
天气工具（Weather tool）
weather_tool() 函数通过调用 OpenWeatherMap API 获取指定城市的当前天气数据。它由 weather_agent() 函数使用。
def weather_tool(city: str) -> str:
weather = OpenWeatherMapAPIWrapper()
tool_result = weather.run("Tampa")
return tool_result(具体 Python 代码实现位于原文相应位置)
在线搜索工具（Online search tool）
当本地活动信息不可用时，search_tool() 函数使用 Tavily API 执行在线搜索，查找指定城市的即将举办活动并返回摘要。它由 search_agent() 函数使用。
def search_tool(city: str) -> str:
client = TavilyClient(api_key=os.environ['TAVILY_API_KEY'])
query = f"What are the upcoming events in {city}?"
response = client.search(query, search_depth="advanced")
results_content = "\n\n".join([result['content'] for result in response['results']])
return results_content (具体 Python 代码实现位于原文相应位置)
餐厅推荐函数（Restaurant recommendation function）
query_restaurants_RAG() 函数使用 RAG 系统提供餐厅推荐：它在向量数据库中对相关餐厅信息执行相似性搜索，过滤出指定城市的高评分餐厅，并使用 Amazon Bedrock 和 Mistral 模型基于检索到的信息生成顶级餐厅摘要。它由 query_restaurants_agent() 函数使用。
(具体 Python 代码实现位于原文相应位置)
关于这些函数和工具的详细实现、环境设置和用例，请参考 Multi-Agent LangGraph Mistral Jupyter notebook。
https://github.com/aws-samples/mistral-on-aws/blob/main/notebooks/mistr…
使用 LangGraph 实现 AI 智能体
我们的多智能体系统由几个专门的智能体组成。在此架构中，每个智能体在 LangGraph 中表示为一个 Node（节点），该节点又与之前定义的工具和函数进行交互。
工作流遵循以下步骤：
- 活动数据库智能体 (events_database_agent) – 使用 events_database_tool() 查询本地 SQLite 数据库以查找本地活动信息。
- 在线搜索智能体 (search_agent) – 当数据库中本地活动信息不可用时，该智能体使用 search_tool() 通过在线搜索查找指定城市的即将举办活动。
- 天气智能体 (weather_agent) – 使用 weather_tool() 获取指定城市的当前天气数据。
- 餐厅推荐智能体 (query_restaurants_agent) – 使用 query_restaurants_RAG() 函数为指定城市提供餐厅推荐。
- 分析智能体 (analysis_agent) – 汇总来自其他智能体的信息以提供全面的建议。
以下是 weather_agent 的核心代码结构示例（省略细节）：
def weather_agent(state: State) -> State:
......
tool_config = {
"tools": [
{
"toolSpec": {
"name": "get_city_weather",
"description": "Get current weather information for a specific city",
"inputSchema": {
"json": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "The name of the city to look up weather for"
}
},
"required": ["city"]
}
}
}
}
]
}
input_text = f"Get current weather for {state.city}"
output_message, tool_result = generate_text(bedrock_client, DEFAULT_MODEL, tool_config, input_text)
if tool_result:
state.weather_info = {"city": state.city, "weather": tool_result}
else:
state.weather_info = {"city": state.city, "weather": "Weather information not available."}
print(f"Weather info set to: {state.weather_info}")
return state编排智能体协作
在多智能体城市信息系统中，build_graph() 函数负责使用 LangGraph 编排智能体协作，定义工作流（节点、路由、条件）。
该函数的行为概述：
1.初始化工作流 –
创建 StateGraph 对象 (workflow)，使用表示智能体间传递数据和上下文的 State 对象初始化。
workflow = StateGraph(State)
2.添加节点（智能体）–
将每个智能体函数注册为工作流中的一个节点。
workflow.add_node("活动数据库智能体", events_database_agent)
workflow.add_node("在线搜索智能体", search_agent)
workflow.add_node("天气智能体", weather_agent)
workflow.add_node("餐厅推荐智能体", query_restaurants_agent)
workflow.add_node("分析智能体", analysis_agent)3.设置入口点和条件路由 –
入口点设为“活动数据库智能体”。定义条件路由函数 route_events()，它检查 events_database_agent 的结果：若本地数据库找不到活动，则路由到“在线搜索智能体”，否则路由到“天气智能体”。
workflow.set_entry_point("活动数据库智能体")
workflow.add_conditional_edges("活动数据库智能体", route_events, {"在线搜索智能体": "在线搜索智能体", "天气智能体": "天气智能体"})4.添加智能体间边 –
定义节点之间的确定流向：搜索智能体 -> 天气智能体 -> 餐厅推荐智能体 -> 分析智能体 -> 结束 (END)。
workflow.add_edge("在线搜索智能体", "天气智能体")
workflow.add_edge("天气智能体", "餐厅推荐智能体")
workflow.add_edge("餐厅推荐智能体", "分析智能体")
workflow.add_edge("分析智能体", END)5.初始化状态持久化内存 – 使用 MemorySaver 对象 (checkpointer) 确保智能体执行过程中的状态得以保留。
checkpointer = MemorySaver()6.编译工作流和可视化 – 编译工作流并关联状态存储器 (checkpointer)。可视化工作流图（图代码省略）。
app = workflow.compile(checkpointer=checkpointer)# 可视化代码...
结果与分析
为了展示我们多智能体城市信息系统的多功能性，我们为三个不同的城市运行它：坦帕（Tampa）、费城（Philadelphia）和纽约（New York）。每个示例都展示了系统功能的不同方面。
运行代码的核心部分：
if __name__ == "__main__":
cities = ["Tampa", "Philadelphia", "New York"]
for city in cities:
print(f"\nStarting script execution for city: {city}")
main(city)三个用例示例的结果总结：
1.示例 1 (坦帕 - Tampa)：
- 活动：本地数据库未找到事件 → 触发在线搜索（Tavily API）→ 成功获取多个事件信息。
- 天气：工具获取成功 → 中雨、28°C、87%湿度。
- 活动与着装：根据天气和活动生成相应建议 → 室内/外活动建议 + 轻便透气衣物和雨具推荐。
- 餐厅：RAG 系统提供推荐。
2.示例 2 (费城 - Philadelphia)：
- 活动：本地数据库中找到事件 → 如文化活动和节日。
- 天气：工具获取成功。
- 活动与着装：生成基于本地事件和天气的建议。
- 餐厅：RAG 系统提供推荐。
3.示例 3 (纽约 - New York)：
- 活动：本地数据库中找到事件 → 如百老汇演出、景点。
- 天气：工具获取成功。
- 活动与着装：生成基于多样化本地事件和天气的建议 → 建议适合城市环境的着装。
- 餐厅：RAG 系统失败 → 因为早期创建的合成数据集未包含纽约餐厅数据 → 系统基于其他信息仍能提供部分见解。
这些示例展示了系统适应不同场景（本地数据有无、在线搜索、RAG 成功/失败）的能力。详细输出可参考笔记本的“结果与分析”部分。
结论
在我们开发的多智能体城市信息系统中，智能体将各种数据源和 API 整合在一个灵活、模块化的框架内，提供有关不同城市的活动、天气、建议活动、着装推荐和餐饮选择等有价值的信息。利用 Amazon Bedrock 和 LangGraph，我们创建了一个复杂的基于智能体的工作流，能够无缝适应不同可用信息水平，根据需要在本地方案和在线数据源之间切换。这些智能体自主地收集、处理和整合数据形成可行见解，编排和自动化业务逻辑以简化流程并提供实时洞察。因此，这种多智能体方法使得创建健壮、可扩展且智能的智能体系统成为可能，这些系统正在突破生成式 AI 的能力边界。
想要深入探索？请访问 GitHub 仓库以查看使用 LangGraph 实现 Mistral 模型的多智能体协作与编排（Implementation of Multi-Agent Collaboration and Orchestration using LangGraph for Mistral Models）的实际代码并亲自尝试该解决方案：
您将找到设置和运行多智能体系统的分步说明，以及与数据源、智能体交互、路由数据和可视化工作流相关的代码。
- 登录 发表评论
- 3 次浏览
最新内容
- 8 minutes 4 seconds ago
- 17 hours ago
- 18 hours 9 minutes ago
- 20 hours 14 minutes ago
- 20 hours 27 minutes ago
- 20 hours 34 minutes ago
- 1 week 1 day ago
- 1 month ago
- 1 month 1 week ago
- 1 month 1 week ago