在我們之前的教學中，我們建立了一個能夠透過網路搜尋回答問題的人工智慧代理，並增加了持久性來維持狀態。然而，在許多情況下，你可能希望讓人類參與監控和批准代理的行動。這可以輕鬆地透過 LangGraph 來實現。讓我們來看看這是怎麼運作的。
設置代理
我們將從上次課程的內容繼續。首先，設置環境變數，進行必要的導入，並配置檢查點。
os.environ[‘TAVILY_API_KEY’] = “<TAVILY_API_KEY>”
os.environ[‘GROQ_API_KEY’] = “<GROQ_API_KEY>”
from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage, AIMessage
from langchain_groq import ChatGroq
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3
sqlite_conn = sqlite3.connect(“checkpoints.sqlite”,check_same_thread=False)
memory = SqliteSaver(sqlite_conn)
# 初始化搜尋工具
tool = TavilySearchResults(max_results=2)
定義代理
def __init__(self, model, tools, checkpointer, system=””):
self.system = system
graph = StateGraph(AgentState)
graph.add_node(“llm”, self.call_openai)
graph.add_node(“action”, self.take_action)
graph.add_conditional_edges(“llm”, self.exists_action, {True: “action”, False: END})
graph.add_edge(“action”, “llm”)
graph.set_entry_point(“llm”)
self.graph = graph.compile(checkpointer=checkpointer)
self.tools = {t.name: t for t in tools}
self.model = model.bind_tools(tools)
def call_openai(self, state: AgentState):
messages = state[‘messages’]
if self.system:
messages = [SystemMessage(content=self.system)] + messages
message = self.model.invoke(messages)
return {‘messages’: [message]}
def exists_action(self, state: AgentState):
result = state[‘messages’][-1]
return len(result.tool_calls) > 0
def take_action(self, state: AgentState):
tool_calls = state[‘messages’][-1].tool_calls
results = []
for t in tool_calls:
print(f”Calling: {t}”)
result = self.tools[t[‘name’]].invoke(t[‘args’])
results.append(ToolMessage(tool_call_id=t[‘id’], name=t[‘name’], content=str(result)))
print(“Back to the model!”)
return {‘messages’: results}
設置代理狀態
我們現在稍微修改一下代理狀態的配置。之前，消息列表是用 operator.add 來註解的，將新消息附加到現有的數組中。對於人類參與的互動，有時我們也希望用相同的 ID 替換現有的消息，而不是附加它們。
def reduce_messages(left: list[AnyMessage], right: list[AnyMessage]) -> list[AnyMessage]:
# 為沒有 ID 的消息分配 ID
for message in right:
if not message.id:
message.id = str(uuid4())
# 將新消息與現有消息合併
merged = left.copy()
for message in right:
for i, existing in enumerate(merged):
if existing.id == message.id:
merged[i] = message
break
else:
merged.append(message)
return merged
class AgentState(TypedDict):
messages: Annotated[list[AnyMessage], reduce_messages]
加入人類參與
在編譯圖形時，我們引入了一個額外的修改。interrupt_before=[“action”] 參數在調用 action 節點之前添加了一個中斷，確保在執行工具之前需要手動批准。
def __init__(self, model, tools, checkpointer, system=””):
# 其他部分與之前相同
self.graph = graph.compile(checkpointer=checkpointer, interrupt_before=[“action”])
# 其他部分保持不變
運行代理
現在，我們將使用與之前相同的提示、模型和檢查點來初始化系統。當我們調用代理時，我們傳遞線程配置和線程 ID。
你可以進行多次調用（無論是一起還是依次）。 \
只有在確定你想要的資訊時才查找。 \
如果你需要在提出後續問題之前查找一些資訊，你可以這樣做！
“””
model = ChatGroq(model=”Llama-3.3-70b-Specdec”)
abot = Agent(model, [tool], system=prompt, checkpointer=memory)
messages = [HumanMessage(content=”舊金山的天氣怎麼樣？”)]
thread = {“configurable”: {“thread_id”: “1”}}
for event in abot.graph.stream({“messages”: messages}, thread):
for v in event.values():
print(v)
回應會被串流回來，並且在 AI 消息後過程會停止，這表示進行了一次工具調用。然而，interrupt_before 參數防止了立即執行。我們也可以獲取此線程的圖形當前狀態，看看它包含了什麼，並且它還包含了下一個要調用的節點（這裡是 ‘action’）。
abot.graph.get_state(thread).next
要繼續，我們再次使用相同的線程配置調用串流，並傳遞 None 作為輸入。這會串流回結果，包括工具消息和最終的 AI 消息。由於 action 節點和 LLM 節點之間沒有添加中斷，因此執行會無縫繼續。
for v in event.values():
print(v)
互動人類批准
我們可以實現一個簡單的循環，提示用戶在繼續執行之前進行批准。使用新的線程 ID 進行新的執行。如果用戶選擇不繼續，代理將停止。
thread = {“configurable”: {“thread_id”: “2”}}
for event in abot.graph.stream({“messages”: messages}, thread):
for v in event.values():
print(v)
while abot.graph.get_state(thread).next:
print(“\n”, abot.graph.get_state(thread), “\n”)
_input = input(“繼續嗎？ (y/n): “)
if _input.lower() != “y”:
print(“中止”)
break
for event in abot.graph.stream(None, thread):
for v in event.values():
print(v)
太好了！現在你知道如何讓人類參與其中。現在，試著實驗不同的中斷，看看代理的行為。
參考資料：DeepLearning.ai (https://learn.deeplearning.ai/courses/ai-agents-in-langgraph/lesson/6/human-in-the-loop)
本文由 AI 台灣 運用 AI 技術編撰，內容僅供參考，請自行核實相關資訊。
歡迎加入我們的 AI TAIWAN 台灣人工智慧中心 FB 社團，
隨時掌握最新 AI 動態與實用資訊！
