在我們之前的教學中，我們建立了一個能夠透過網路回答問題的人工智慧代理。然而，當我們為長時間運行的任務建立代理時，有兩個重要的概念需要考慮：持久性和串流。持久性讓你可以在任何時刻保存代理的狀態，這樣在未來的互動中可以從該狀態繼續。這對於長時間運行的應用程式來說非常重要。另一方面，串流則讓你能夠即時發送代理在任何時刻的行動信號，提供透明度和對其行動的控制。在本教程中，我們將透過加入這些強大的功能來增強我們的代理。
設置代理
讓我們開始重新創建我們的代理。我們將載入必要的環境變數，安裝並導入所需的庫，設置 Tavily 搜索工具，定義代理狀態，最後建立代理。
os.environ[‘TAVILY_API_KEY’] = “<TAVILY_API_KEY>”
os.environ[‘GROQ_API_KEY’] = “<GROQ_API_KEY>”
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage
from langchain_groq import ChatGroq
from langchain_community.tools.tavily_search import TavilySearchResults
tool = TavilySearchResults(max_results=2)
class AgentState(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
class Agent:
def __init__(self, model, tools, 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()
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
加入持久性
為了加入持久性，我們將使用 LangGraph 的檢查點功能。檢查點會在每個節點之後和之間保存代理的狀態。在本教程中，我們將使用 SqliteSaver，一個簡單的檢查點，利用 SQLite 這個內建的資料庫。雖然我們將使用內存資料庫以簡化操作，但你也可以輕鬆地將其連接到外部資料庫，或使用其他檢查點如 Redis 或 Postgres 來獲得更穩定的持久性。
import sqlite3
sqlite_conn = sqlite3.connect(“checkpoints.sqlite”,check_same_thread=False)
memory = SqliteSaver(sqlite_conn)
接下來，我們將修改我們的代理以接受檢查點：
def __init__(self, model, tools, checkpointer, system=””):
# 其他部分保持不變
self.graph = graph.compile(checkpointer=checkpointer)
# 其他部分保持不變
現在，我們可以創建一個啟用持久性的代理：
你可以進行多次調用（無論是一起還是按順序）。 \
只有在確定你想要的信息時，才可以查找信息。 \
如果你需要在問後續問題之前查找一些信息，你是可以這樣做的！
“””
model = ChatGroq(model=”Llama-3.3-70b-Specdec”)
bot = Agent(model, [tool], system=prompt, checkpointer=memory)
加入串流
串流對於即時更新是必不可少的。我們將專注於兩種類型的串流：
1. 串流消息：發送中間消息，如 AI 決策和工具結果。
2. 串流標記：從 LLM 的回應中串流單個標記。讓我們先開始串流消息。我們將創建一條人類消息，並使用串流方法即時觀察代理的行動。
thread = “configurable”: “thread_id”: “1”
for event in bot.graph.stream(“messages”: messages, thread):
for v in event.values():
print(v[‘messages’])
最終輸出：德克薩斯州的當前天氣是晴天，氣溫為 19.4°C (66.9°F)，風速為 4.3 mph (6.8 kph)…..
當你運行這個時，你會看到一系列結果。首先是一條 AI 消息指示代理調用 Tavily，接著是一條工具消息顯示搜索結果，最後是一條 AI 消息回答問題。
理解線程 ID
線程 ID 是線程配置中的一個重要部分。它允許代理與不同的用戶或上下文保持獨立的對話。通過為每個對話分配唯一的線程 ID，代理可以同時跟踪多個互動，而不會混淆。
例如，讓我們繼續對話，問：“洛杉磯的天氣怎麼樣？”使用相同的線程 ID：
thread = “configurable”: “thread_id”: “1”
for event in bot.graph.stream(“messages”: messages, thread):
for v in event.values():
print(v)
最終輸出：洛杉磯的當前天氣是晴天，氣溫為 17.2°C (63.0°F)，風速為 2.2 mph (3.6 kph) ….
代理推斷我們在詢問天氣，這得益於持久性。為了驗證，讓我們問：“哪個更暖和？”：
thread = “configurable”: “thread_id”: “1”
for event in bot.graph.stream(“messages”: messages, thread):
for v in event.values():
print(v)
最終輸出：德克薩斯州比洛杉磯暖和。德克薩斯州的當前氣溫為 19.4°C (66.9°F)，而洛杉磯的當前氣溫為 17.2°C (63.0°F)
代理正確比較了德克薩斯州和洛杉磯的天氣。為了測試持久性是否能保持對話的獨立性，讓我們用不同的線程 ID 提問相同的問題：
thread = “configurable”: “thread_id”: “2”
for event in bot.graph.stream(“messages”: messages, thread):
for v in event.values():
print(v)
輸出：我需要更多信息來回答這個問題。你能提供更多上下文或具體說明你在比較哪兩樣東西嗎？
這次，代理感到困惑，因為它無法訪問之前對話的歷史。
串流標記
要串流標記，我們將使用 astream_events 方法，這是一個異步方法。我們還將切換到異步檢查點。
async with AsyncSqliteSaver.from_conn_string(“:memory:”) as checkpointer:
abot = Agent(model, [tool], system=prompt, checkpointer=checkpointer)
messages = [HumanMessage(content=”舊金山的天氣如何？”)]
thread = “configurable”: “thread_id”: “4”
async for event in abot.graph.astream_events(“messages”: messages, thread, version=”v1″):
kind = event[“event”]
if kind == “on_chat_model_stream”:
content = event[“data”][“chunk”].content
if content:
# 在 OpenAI 的上下文中，空內容意味著
# 模型正在請求調用一個工具。
# 所以我們只打印非空內容
print(content, end=”|”)
這將即時串流標記，讓你能夠實時查看代理的思考過程。
結論
通過加入持久性和串流，我們顯著增強了我們的人工智慧代理的能力。持久性使代理能夠在互動中保持上下文，而串流則提供了對其行動的即時洞察。這些功能對於構建生產就緒的應用程式尤其重要，特別是那些涉及多個用戶或人機互動的應用。
在下一個教程中，我們將深入探討人機互動，其中持久性在促進人類與人工智慧代理之間的無縫協作中扮演著重要角色。敬請期待！
參考資料：
(DeepLearning.ai) https://learn.deeplearning.ai/courses/ai-agents-in-langgraph
此外，別忘了在 Twitter 上關注我們，並加入我們的 Telegram 頻道和 LinkedIn 群組。別忘了加入我們的 75k+ ML SubReddit。
🚨 介紹 IntellAgent：一個開源的多代理框架，用於評估複雜的對話 AI 系統（推廣）
本文由 AI 台灣 運用 AI 技術編撰，內容僅供參考，請自行核實相關資訊。
歡迎加入我們的 AI TAIWAN 台灣人工智慧中心 FB 社團，
隨時掌握最新 AI 動態與實用資訊！
