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