只需8步，手把手教你用LangGraph創建AI智能體
AI領域正從基礎的RAG系統向更智能的AI智能體進化，后者能處理更復雜的任務并適應新信息。LangGraph作為LangChain庫的擴展，助力開發者構建具有狀態管理和循環計算能力的先進AI系統。本文教大家如何使用LangGraph開發一個太陽能節能計算的智能體。
1 LangGraph概述
LangGraph是LangChain的高級庫，為大型語言模型（LLM）帶來循環計算能力。它超越了LangChain的線性工作流，通過循環支持復雜的任務處理。
- 狀態：維護計算過程中的上下文，實現基于累積數據的動態決策。
- 節點：代表計算步驟，執行特定任務，可定制以適應不同工作流。
- 邊：連接節點，定義計算流程，支持條件邏輯，實現復雜工作流。
LangGraph簡化了AI開發，自動管理狀態，保持上下文，使AI能智能響應變化。它讓開發者專注于創新，而非技術細節，同時確保應用程序的高性能和可靠性。
2 逐步指南
了解LangGraph后，我們通過實例來實際應用：構建一個AI智能體，用于計算太陽能板節能潛力，并在銷售網站上與潛在客戶互動，提供個性化節能估算。這個智能體有助于教育客戶太陽能的經濟效益，并篩選出值得跟進的潛在客戶。
步驟1：導入必要的庫
我們先導入構建AI助手所需的Python庫和模塊。
from langchain_core.tools import tool
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import Runnable
from langchain_aws import ChatBedrock
import boto3
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import AnyMessage, add_messages
from langchain_core.messages import ToolMessage
from langchain_core.runnables import RunnableLambda
from langgraph.prebuilt import ToolNode
from langgraph.prebuilt import tools_condition這些庫的導入為我們利用LangChain、LangGraph和AWS服務來構建AI助手打下了堅實的基礎。
步驟2：定義計算太陽能節省的工具
接下來，定義一個工具來根據用戶每月的電費計算使用太陽能板可能節省的費用。
@tool
def compute_savings(monthly_cost: float) -> float:
"""
根據用戶每月電費計算切換到太陽能時潛在節省的工具。
參數：
monthly_cost (float)：用戶當前的每月電費。
返回：
dict：包含以下內容的字典：
- 'number_of_panels'：估計所需的太陽能板數量。
- 'installation_cost'：估計的安裝成本。
- 'net_savings_10_years'：安裝成本后的10年凈節省。
"""
def calculate_solar_savings(monthly_cost):
# 計算的假設
cost_per_kWh = 0.28
cost_per_watt = 1.50
sunlight_hours_per_day = 3.5
panel_wattage = 350
system_lifetime_years = 10
# 每月用電量（千瓦時）
monthly_consumption_kWh = monthly_cost / cost_per_kWh
# 所需系統大小（千瓦）
daily_energy_production = monthly_consumption_kWh / 30
system_size_kW = daily_energy_production / sunlight_hours_per_day
# 太陽能板數量和安裝成本
number_of_panels = system_size_kW * 1000 / panel_wattage
installation_cost = system_size_kW * 1000 * cost_per_watt
# 年度和凈節省
annual_savings = monthly_cost * 12
total_savings_10_years = annual_savings * system_lifetime_years
net_savings = total_savings_10_years - installation_cost
return {
"number_of_panels": round(number_of_panels),
"installation_cost": round(installation_cost, 2),
"net_savings_10_years": round(net_savings, 2)
}
# 返回計算的太陽能節省
return calculate_solar_savings(monthly_cost)這個函數基于用戶電費數據，提供太陽能板系統的詳細節省估算，包括所需板數量、安裝成本和未來十年的凈節省。目前，我們使用了一些平均值來進行簡化計算。未來，我們可以從用戶那里直接獲取更精確的數據，以提供更個性化的估算。
步驟3：設置狀態管理和錯誤處理
有效的狀態管理和錯誤處理對于構建健壯的AI系統非常重要。在這里，我們定義了工具來管理錯誤并維護對話的狀態。
def handle_tool_error(state) -> dict:
"""
處理工具執行期間發生的錯誤的函數。
參數： state (dict)：AI智能體的當前狀態，包括消息和工具調用詳情。
返回：
dict：包含每個遇到問題的工具的錯誤消息的字典。
"""
# 從當前狀態中檢索錯誤
error = state.get("error")
# 從狀態的消息歷史中獲取最后一個消息的工具調用
tool_calls = state["messages"][-1].tool_calls
# 返回包含錯誤詳情的ToolMessages列表，與每個工具調用ID關聯
return {
"messages": [
ToolMessage(
cnotallow=f"錯誤：{repr(error)}\n請修正你的錯誤。", # 為用戶格式化錯誤消息
tool_call_id=tc["id"], # 將錯誤消息與相應的工具調用ID關聯
)
for tc in tool_calls # 遍歷每個工具調用以產生單獨的錯誤消息
]
}
def create_tool_node_with_fallback(tools: list) -> dict:
"""
創建具有后備錯誤處理的工具節點的函數。
參數：
tools (list)：要包含在節點中的工具列表。
返回：
dict：如果發生錯誤，使用后備行為的工具節點。
"""
# 使用提供的工具創建ToolNode，并附加后備機制
# 如果發生錯誤，將調用handle_tool_error函數來管理錯誤
return ToolNode(tools).with_fallbacks(
[RunnableLambda(handle_tool_error)], # 使用lambda函數包裝錯誤處理器
exception_key="error" # 指定這個后備是用于處理錯誤的
)這些函數確保在工具執行期間遇到的任何錯誤都能得到優雅地處理，為用戶提供有用的反饋。
步驟4：定義狀態和助手類
在此步驟，我們設定AI智能體如何維護對話狀態并響應用戶輸入及工具輸出。
用Python的TypedDict創建State類來規范消息結構，包括用戶和系統的消息。
class State(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]然后，構建助手類來驅動AI智能體，管理對話。助手類調用工具，處理結果，并在需要時重新向用戶詢問。它通過循環調用Runnable直至獲得有效輸出，確保對話流暢。
class Assistant:
def __init__(self, runnable: Runnable):
# 使用定義與工具交互過程的可運行對象進行初始化
self.runnable = runnable
def
__call__(self, state: State):
while True:
# 使用當前狀態（消息和上下文）調用可運行對象
result = self.runnable.invoke(state)
# 如果工具未能返回有效輸出，重新提示用戶澄清或重試
if not result.tool_calls and (
not result.content
or isinstance(result.content, list)
and not result.content[0].get("text")
):
# 添加請求有效響應的消息
messages = state["messages"] + [("user", "請給出一個真實的輸出。")]
state = {**state, "messages": messages}
else:
# 當獲得有效輸出時跳出循環
break
# 在處理完可運行對象后返回最終狀態
return {"messages": result}這個機制確保了對話的連貫性和助手的恰當響應。
步驟5：使用AWS Bedrock配置LLM
在這一步，我們通過AWS Bedrock設置大型語言模型（LLM），增強AI助手的語言處理能力。需要先配置AWS憑證，以便訪問Bedrock服務。
def get_bedrock_client(region):
return boto3.client("bedrock-runtime", region_name=region)
def create_bedrock_llm(client):
return ChatBedrock(model_id='anthropic.claude-3-sonnet-20240229-v1:0', client=client, model_kwargs={'temperature': 0}, region_name='us-east-1')
llm = create_bedrock_llm(get_bedrock_client(reginotallow='us-east-1'))這確保了助手能準確理解和回應用戶。
步驟6：定義助手的工作流程
設置好LLM和工具后，現在定義AI助手的工作流程，主要涉及創建對話模板和指定工具的使用。
工作流程的第一部分創建一個模板來引導助手與用戶溝通，明確要問的問題和如何根據回答調用工具。
primary_assistant_prompt = ChatPromptTemplate.from_messages(
[
(
"system",
'''你是太陽能板比利時的樂于助人的客服助手。
你應該從他們那里獲取以下信息：
- 每月電費
如果你不能清晰地識別這些信息，要求他們澄清！不要試圖胡亂猜測。
在你能夠清晰地識別所有信息后，調用相關工具。
''',
),
("placeholder", "{messages}"),
]
)確定助手將使用的工具，如compute_savings，并將其綁定到工作流程中，以便在對話中適時調用。
# 定義助手將使用的工具
part_1_tools = [
compute_savings
]
# 將工具綁定到助手的工作流程
part_1_assistant_runnable = primary_assistant_prompt | llm.bind_tools(part_1_tools)這確保了助手能根據用戶輸入靈活響應，實現順暢的對話體驗。
步驟7：構建圖結構
利用LangGraph，我們為AI助手搭建圖結構，控制其處理用戶輸入、觸發工具和階段轉換的流程。
- 節點：代表操作步驟，如助手互動和工具執行。
- 邊：定義步驟間的流程，如從助手到工具，再返回助手。
AI智能體旨在計算太陽能板潛在節能節省。
builder = StateGraph(State)
builder.add_node("assistant", Assistant(part_1_assistant_runnable))
builder.add_node("tools", create_tool_node_with_fallback(part_1_tools))
邊定義了流程如何在節點之間移動。在這里，助手開始對話，一旦收集到所需輸入，就過渡到工具，并在工具執行后返回助手。
builder.add_edge(START, "assistant") # 從助手開始
builder.add_conditional_edges("assistant", tools_condition) # 輸入后移動到工具
builder.add_edge("tools", "assistant") # 工具執行后返回助手
我們使用MemorySaver確保圖在不同步驟之間保留對話狀態。這允許助手記住用戶的輸入，確保多步驟交互中的連續性。
memory = MemorySaver()
graph = builder.compile(checkpointer=memory)步驟8：運行助手
最后，你可以通過啟動圖并開始對話來運行助手。
# import shutil
import uuid
# 讓我們創建一個用戶可能與助手進行的示例對話
tutorial_questions = [
'嘿',
'你能計算我的節能嗎',
“我的每月成本是100美元，我能節省多少”
]
thread_id = str(uuid.uuid4())
config = {
"configurable": {
"thread_id": thread_id,
}
}
_printed = set()
for question in tutorial_questions:
events = graph.stream(
{"messages": ("user", question)}, config, stream_mode="values"
)
for event in events:
_print_event(event, _printed)通過這些步驟，現在已經使用LangGraph成功打造了一個能根據用戶輸入計算太陽能節能的AI助手。這凸顯了LangGraph在處理復雜任務和解決實際問題中的優勢。
本文轉載自 ??AI科技論談??，作者： AI科技論談
