代理人正在改變生成式人工智慧的格局，成為大型語言模型 (LLMs) 與現實應用之間的橋樑。這些智能、自主的系統有望成為各行各業採用人工智慧的基石，開創人類與人工智慧合作和解決問題的新時代。通過利用 LLMs 的力量，並將其與專門的工具和應用程式介面 (APIs) 結合，代理人可以處理複雜的多步驟任務，這些任務在傳統人工智慧系統中是無法完成的。本文中展示的多代理城市資訊系統就是代理人架構潛力的範例，能夠創造出複雜、適應性強且高度能力的人工智慧應用。
展望未來，代理人在以下方面將扮演非常重要的角色：
改善決策，提供更深入且具上下文意識的信息
自動化各個領域的複雜工作流程，從客戶服務到科學研究
促進更自然和直觀的人類與人工智慧的互動
通過整合多樣的數據來源和專業知識來產生新想法
通過提供更透明和可解釋的人工智慧系統來解決倫理問題
建立和部署像本文所述的多代理系統是釋放生成式人工智慧全部潛力的一步。隨著這些系統的發展，它們將改變行業，擴大可能性，並為人工智慧開啟新的大門。
解決方案概述
在本文中，我們探討如何在 Amazon Bedrock 上使用 LangGraph 和 Mistral 模型來創建一個強大的多代理系統，能夠通過協作解決問題來處理複雜的工作流程。這種整合使得創建可以協同工作的人工智慧代理人成為可能，以解決複雜的問題，模仿人類的推理和合作。
結果是一個系統，提供有關事件、天氣、活動和特定城市的建議的全面詳細信息，展示了如何在 Amazon Web Services (AWS) 上構建和部署有狀態的多代理應用，以應對現實挑戰。
LangGraph 對我們的解決方案至關重要，提供了一種有組織的方法來定義和管理代理之間的信息流。它內建支持狀態管理和檢查點，確保流程的連續性。這個框架還允許簡單地可視化代理工作流程，增強清晰度和理解。它與 LLMs 和 Amazon Bedrock 輕鬆整合，提供了一個多功能且強大的解決方案。此外，它對條件路由的支持允許根據中間結果動態調整工作流程，提供靈活性以處理不同情境。
我們提出的多代理架構提供了幾個關鍵優勢：
模組化 – 每個代理專注於特定任務，使系統更易於維護和擴展
靈活性 – 可以快速添加、移除或修改代理，而不影響整個系統
複雜工作流程處理 – 系統可以通過將任務分配給多個代理來管理先進和複雜的工作流程
專業化 – 每個代理針對其特定任務進行優化，提高延遲、準確性和整體系統效率
安全性 – 系統通過確保每個代理僅能訪問其任務所需的工具來增強安全性，減少未經授權訪問敏感數據或其他代理任務的潛在風險
我們的多代理系統如何運作
在本節中，我們探討我們的多代理城市資訊系統是如何運作的，基於可在 GitHub 的 Mistral on AWS 範例中找到的多代理 LangGraph Mistral Jupyter 筆記本。
這個代理工作流程以城市名稱作為輸入，提供詳細信息，展示了在處理不同情境時的適應性：
事件 – 它在本地數據庫和在線來源中搜索即將舉行的城市活動。每當本地數據庫信息不可用時，它會使用 Tavily API 觸發在線搜索。這確保用戶無論是本地存儲還是需要從網絡檢索，都能獲得最新的事件信息
天氣 – 系統使用 OpenWeatherMap API 獲取當前天氣數據，提供準確及時的天氣信息。根據天氣，系統還提供根據條件量身定制的服裝和活動建議，為每個城市提供相關建議
餐廳 – 通過檢索增強生成 (RAG) 系統提供建議。這種方法將預先存儲的信息與實時生成相結合，以提供相關且最新的用餐建議
系統在處理不同信息層級的能力通過其適應性方法得以展示，這意味著用戶無論不同城市的信息可用性如何，都能獲得最全面和最新的信息。例如：
某些城市可能需要在本地數據庫數據不可用時使用搜索工具來獲取事件信息
其他城市可能在本地數據庫中有數據，提供快速訪問事件信息，而無需在線搜索
在某些城市無法提供餐廳建議的情況下，系統仍然可以根據可用的事件和天氣數據提供有價值的見解
以下圖表是解決方案的參考架構：
數據來源
多代理城市資訊系統可以利用兩個數據來源。
本地事件數據庫
這個 SQLite 數據庫從 JSON 文件中填充城市事件數據，提供快速訪問從社區活動到文化活動和全市活動的本地事件信息。這個數據庫由 events_database_tool() 使用，以高效查詢和檢索城市事件的詳細信息，包括位置、日期和事件類型。
餐廳 RAG 系統
對於餐廳建議，generate_restaurants_dataset() 函數生成合成數據，創建特別針對我們的推薦系統的自定義數據集。create_restaurant_vector_store() 函數處理這些數據，使用 Amazon Titan 文本嵌入生成嵌入，並使用 Facebook AI 相似性搜索 (FAISS) 建立向量存儲。雖然這種方法適合原型設計，但對於更具可擴展性和企業級的解決方案，我們建議使用 Amazon Bedrock 知識庫。
建立多代理架構
我們的多代理城市資訊系統的核心是一組專門的函數和工具，旨在從各種來源收集、處理和綜合信息。它們構成了我們系統的支柱，使其能夠提供有關城市的全面和最新信息。在本節中，我們探討驅動我們系統的關鍵組件：generate_text() 函數，使用 Mistral 模型，以及用於本地數據庫查詢、在線搜索、天氣信息和餐廳建議的專門數據檢索函數。這些函數和工具共同創造了一個強大且多功能的系統，能夠為用戶提供有價值的見解。
文本生成函數
這個函數作為我們代理的核心，允許它們根據需要使用 Mistral 模型生成文本。它使用 Amazon Bedrock Converse API，支持文本生成、流式傳輸和外部函數調用（工具）。
該函數的工作原理如下：
將用戶消息發送到 Mistral 模型，使用 Amazon Bedrock Converse API
調用適當的工具並將結果納入對話中
繼續對話，直到生成最終回應
以下是實現：
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) # 將助手的回應添加到消息中
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”未知工具: {tool_name}”)
tool_response = {
“toolUseId”: tool_use_id,
“content”: [{“json”: json.loads(json_result)}]
}
……
messages.append({
“role”: “user”,
“content”: [{“toolResult”: tool_response}]
})
# 使用新消息更新 kwargs
kwargs[“messages”] = messages
else:
break
return output_message, tool_result
本地數據庫查詢工具
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’]} 在 {row[‘event_date’]}： {row[‘description’]}”
),
axis=1
).tolist()
return “\n”.join(events)
else:
return f”{city} 沒有即將舉行的事件。”
天氣工具
weather_tool() 通過調用 OpenWeatherMap API 獲取指定城市的當前天氣數據。它由 weather_agent() 函數使用。以下是代碼：
def weather_tool(city: str) -> str:
weather = OpenWeatherMapAPIWrapper()
tool_result = weather.run(“Tampa”)
return tool_result
在線搜索工具
當本地事件信息不可用時，search_tool() 使用 Tavily API 執行在線搜索，以查找指定城市的即將舉行的事件並返回摘要。它由 search_agent() 函數使用。以下是代碼：
def search_tool(city: str) -> str:
client = TavilyClient(api_key=os.environ[‘TAVILY_API_KEY’])
query = f”{city} 有哪些即將舉行的事件？”
response = client.search(query, search_depth=”advanced”)
results_content = “\n\n”.join([result[‘content’] for result in response[‘results’]])
return results_content
餐廳推薦函數
query_restaurants_RAG() 函數使用 RAG 系統提供餐廳建議，通過在向量數據庫中執行相似性搜索來獲取相關餐廳信息，過濾指定城市中的高評價餐廳，並使用 Amazon Bedrock 和 Mistral 模型生成基於檢索信息的頂級餐廳摘要。它由 query_restaurants_agent() 函數使用。
有關這些函數和工具的詳細實現、環境設置和用例，請參閱多代理 LangGraph Mistral Jupyter 筆記本。
使用 LangGraph 實現人工智慧代理
我們的多代理系統由幾個專門的代理組成。這個架構中的每個代理都由 LangGraph 中的節點表示，這些節點與之前定義的工具和函數進行互動。以下圖表顯示了工作流程：
工作流程遵循以下步驟：
事件數據庫代理 (events_database_agent) – 使用 events_database_tool() 查詢本地 SQLite 數據庫並查找本地事件信息
在線搜索代理 (search_agent) – 每當數據庫中沒有本地事件信息時，該代理使用 search_tool() 在線查找指定城市的即將舉行的事件
天氣代理 (weather_agent) – 使用 weather_tool() 獲取指定城市的當前天氣數據
餐廳推薦代理 (query_restaurants_agent) – 使用 query_restaurants_RAG() 函數為指定城市提供餐廳建議
分析代理 (analysis_agent) – 聚合來自其他代理的信息，以提供全面的建議
以下是我們創建天氣代理的示例：
def weather_agent(state: State) -> State:
……
tool_config = {
“tools”: [
{
“toolSpec”: {
“name”: “get_city_weather”,
“description”: “獲取特定城市的當前天氣信息”,
“inputSchema”: {
“json”: {
“type”: “object”,
“properties”: {
“city”: {
“type”: “string”,
“description”: “要查詢天氣的城市名稱”
}
},
“required”: [“city”]
}
}
}
}
]
}
input_text = f”獲取 {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”: “天氣信息不可用。”}
print(f”天氣信息設置為: {state.weather_info}”)
return state
協調代理合作
在多代理城市資訊系統中，有幾個關鍵原語協調代理合作。build_graph() 函數在 LangGraph 中定義工作流程，利用節點、路由和條件。工作流程是動態的，根據事件搜索結果進行條件路由，並包含記憶持久性以在不同執行之間存儲狀態。以下是該函數行為的概述：
初始化工作流程 – 該函數首先創建一個名為 workflow 的 StateGraph 對象，並用 State 初始化。在 LangGraph 中，State 代表在代理執行任務時通過工作流程傳遞的數據或上下文。在我們的示例中，狀態包括來自先前代理的結果（例如事件數據、搜索結果和天氣信息）、輸入參數（例如城市名稱）和代理可能需要處理的其他相關信息：
# 定義圖形
def build_graph():
workflow = StateGraph(State)
…
添加節點（代理） – 每個代理與特定函數相關聯，例如檢索事件數據、執行在線搜索、獲取天氣信息、推薦餐廳或分析收集的信息：
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)
設置入口點和條件路由 – 工作流程的入口點設置為事件數據庫代理，這意味著工作流程的執行從這個代理開始。此外，該函數使用 add_conditional_edges 方法定義條件路由。route_events() 函數根據事件數據庫代理的結果決定下一步：
workflow.set_entry_point(“事件數據庫代理”)
def route_events(state):
print(f”路由事件。當前狀態: {state}”)
print(f”事件內容: ‘{state.events_result}'”)
if f”{state.city} 沒有即將舉行的事件。” in state.events_result:
print(“本地數據庫中未找到事件。路由到在線搜索代理。”)
return “在線搜索代理”
else:
print(“在本地數據庫中找到事件。路由到天氣代理。”)
return “天氣代理”
workflow.add_conditional_edges(
“事件數據庫代理”,
route_events,
{
“在線搜索代理”: “在線搜索代理”,
“天氣代理”: “天氣代理”
}
)
在代理之間添加邊 – 這些邊定義了代理在工作流程中互動的順序。代理將按照特定順序進行：從在線搜索代理到天氣代理，再從天氣代理到餐廳推薦代理，然後到分析代理，最後到達結束：
workflow.add_edge(“在線搜索代理”, “天氣代理”)
workflow.add_edge(“天氣代理”, “餐廳推薦代理”)
workflow.add_edge(“餐廳推薦代理”, “分析代理”)
workflow.add_edge(“分析代理”, END)
初始化狀態持久性的記憶 – MemorySaver 類用於確保工作流程的狀態在運行之間保持。這在多代理系統中特別有用，因為系統的狀態需要在代理互動時保持：
# 初始化記憶以在圖形運行之間持久化狀態
checkpointer = MemorySaver()
編譯工作流程並可視化圖形 – 工作流程被編譯，並包含記憶保存對象 (checkpointer)，以確保狀態在執行之間持久化。然後，它輸出工作流程的圖形表示：
# 編譯工作流程
app = workflow.compile(checkpointer=checkpointer)
# 可視化圖形
display(
Image(
app.get_graph().draw_mermaid_png(
draw_method=MermaidDrawMethod.API
)
)
)
以下圖表說明了這些步驟：
結果與分析
為了展示我們的多代理城市資訊系統的多功能性，我們對三個不同的城市進行了運行：坦帕 (Tampa)、費城 (Philadelphia) 和紐約 (New York)。每個示例展示了系統功能的不同方面。
使用的函數 main() 協調整個過程：
調用 build_graph() 函數，實現代理工作流程
使用指定城市初始化狀態
通過工作流程流式傳輸事件
檢索並顯示最終分析和建議
要運行代碼，請執行以下操作：
if __name__ == “__main__”:
cities = [“Tampa”, “Philadelphia”, “New York”]
for city in cities:
print(f”\n開始執行城市: {city}”)
main(city)
三個示例用例
在示例 1 (坦帕) 中，以下圖表顯示了代理工作流程如何對用戶的問題“坦帕有什麼活動，我應該穿什麼？”產生輸出。
系統產生了以下結果：
事件 – 在本地數據庫中未找到，觸發搜索工具，調用 Tavily API 查找幾個即將舉行的事件
天氣 – 從天氣工具檢索。當前條件包括中等降雨、28°C 和 87% 的濕度
活動 – 系統根據事件和天氣建議了各種室內和室外活動
服裝建議 – 考慮到溫暖、潮濕和多雨的條件，系統建議輕便透氣的衣物和雨具
餐廳 – 通過 RAG 系統提供建議
在示例 2 (費城) 中，代理工作流程在本地數據庫中識別了文化活動和節慶等事件。它從 OpenWeatherMap API 獲取天氣數據，然後根據當地事件和天氣條件建議活動。服裝建議根據天氣預報提出，餐廳建議通過 RAG 系統提供。
在示例 3 (紐約) 中，工作流程在本地數據庫中識別了百老匯表演和城市景點等事件。它從 OpenWeatherMap API 獲取天氣數據，並根據當地事件和天氣條件建議活動。服裝建議根據紐約的天氣和城市環境量身定制。然而，由於之前創建的合成數據集中未包含該城市的餐廳，RAG 系統無法提供紐約的餐廳建議。
這些示例展示了系統適應不同情境的能力。要獲得這些示例的詳細輸出，請參閱多代理 LangGraph Mistral Jupyter 筆記本的結果與分析部分。
結論
在我們開發的多代理城市資訊系統中，代理在靈活的模組化框架內整合各種數據來源和 API，以提供有關不同城市的事件、天氣、活動、服裝建議和用餐選擇的有價值信息。利用 Amazon Bedrock 和 LangGraph，我們創建了一個複雜的基於代理的工作流程，能夠無縫適應可用信息的不同層級，根據需要在本地和在線數據來源之間切換。這些代理自主收集、處理和整合數據，形成可行的見解，協調和自動化業務邏輯，以簡化流程並提供實時見解。因此，這種多代理方法使得創建強大、可擴展和智能的代理系統成為可能，推動生成式人工智慧的邊界。
想深入了解嗎？探索在 GitHub 上使用 LangGraph 和 Mistral 模型實現多代理協作和協調的實作，觀察代碼的運行並親自嘗試解決方案。您將找到有關設置和運行多代理系統的逐步說明，以及與數據來源、代理、路由數據和可視化工作流程互動的代碼。
關於作者
Andre Boaventura 是 AWS 的首席人工智慧/機器學習解決方案架構師，專注於生成式人工智慧和可擴展的機器學習解決方案。在高科技軟體行業擁有超過 25 年的經驗，他在使用 AWS 服務（如 Amazon Bedrock、Amazon SageMaker 和 Amazon Q）設計和部署人工智慧應用方面擁有深厚的專業知識。Andre 與全球系統整合商 (GSIs) 和各行各業的客戶密切合作，架構和實施尖端的人工智慧/機器學習解決方案，以推動商業價值。在工作之餘，Andre 喜歡與他的兒子一起練習巴西柔術（經常被青少年壓制或窒息）、在女兒的舞蹈比賽中為她加油（儘管他不懂芭蕾術語——他仍然熱情地鼓掌），以及與妻子共度“高品質時間”——通常是在購物中心，假裝對衣服和鞋子感興趣，同時暗自考慮一個新愛好。
本文由 AI 台灣 運用 AI 技術編撰，內容僅供參考，請自行核實相關資訊。
歡迎加入我們的 AI TAIWAN 台灣人工智慧中心 FB 社團，
隨時掌握最新 AI 動態與實用資訊！
