使用LangGraph構建你的第一個AI Agent-附完整代碼
AI Agent 指的是有能力主動思考和行動的智能體,能夠以類似人類的方式工作,通過大模型來“理解”用戶需求,主動“規劃”以達成目標,使用各種“工具”來完成任務,并最終“行動”執行這些任務。
從模型到 Agent
在 Agent 出現之前,我們將 AI 模型建為獨立且互不關聯的組件——一個模型用于理解文本,一個模型用于生成代碼,還有一個模型用于處理圖像。
這種碎片化的方法
- 迫使用戶手動管理工作流程
- 導致在不同系統之間切換時上下文信息消失
- 需要為每個流程步驟集成 ai 模型
Agent 的出現就是為了解決這一碎片問題。
與處理孤立任務的傳統模型不同,Agent 能夠管理各種功能,同時保持對整個任務的全面理解,它通過賦予大型語言模型 (LLMs) 訪問工具和知識來擴展其能力,使模型能夠連貫執行一系列操作。
使用 LangGraph 構建 AI 代理
現在讓我們使用 LangGraph來構建一個文本分析 agent。
LangGraph 是 LangChain 生態系統中的一個開源框架,專門用于構建基于大型語言模型(LLM)的復雜應用程序。它通過 圖結構(Graph) 將應用邏輯組織為有向圖,提供靈活的流程控制和狀態管理能力,適用于開發多智能體(Multi-Agent)、多步驟、動態交互的智能系統。
該 Agent 將根據文章內容,進行歸納分類,提取重要元素并給出摘要信息。
開發環境準備
Python:3.11+
1. 導入必要的庫
pip install langgraph langchain langchain-openai python-dotenv
2. import 準備
import os
from typing import TypedDict, List
from langgraph.graph import StateGraph, END
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage
- StateGraph 管理代理組件之間的信息流
- PromptTemplate 創建一致的指令
- ChatOpenAI 連接到 OpenAI 的字符模型,為 agent 提供思維支持
3. 定義 State 類,記錄 Agent 數據
定義一個名為 'State' 的 TypedDict,用于結構化個工具方法的狀態數據
from typing import TypedDict, List
# 定義一個名為 'State' 的TypedDict,用于結構化表示狀態數據
class State(TypedDict):
text: str # 存儲原始輸入文本
classification: str # 分類結果(例如類別標簽)
entities: List[str] # 存儲提取的實體列表(如命名實體)
summary: str # 存儲文本的摘要版本
4. 創建 llm 對象
llm = ChatOpenAI(model="qwen-plus", temperature=0, api_key="",
base_url="")
temperature 參數很重要
- temperature=0:專注、確定性的響應
- temperature=1:更加多樣化、更具創造力的輸出
- temperature=2:天馬行空、有時語無倫次的想法
api_key 是訪問 AI 服務(如 OpenAI、DeepSeek 等)的身份憑證,用于:
- 身份驗證:確保請求來自授權用戶。
- 權限控制:限制調用接口的范圍(如僅允許特定模型或功能)。
- 計費與監控:跟蹤 API 調用次數、費用和使用模式。
base_url 是訪問 AI 服務的連接,不同 AI 服務商的 base_url 不一樣。
5. 添加 agent 功能
現在,我們將為 Agent 構建專用工具方法,每個工具分別處理特定的任務類型。
分類功能
首先,是我們的分類功能:
def classification_node(state: State):
"""
將文本分類到預定義的類別中。
參數:
state (State): 包含待分類文本的當前狀態字典
返回:
dict: 包含"分類"鍵的字典,值為分類結果
分類類別:
- 新聞: 事實性報道當前事件
- 博客: 個人或非正式的網絡寫作
- 研究: 學術或科學內容
- 其他: 不符合上述類別的內容
"""
# 創建提示模板,要求模型將文本分類到指定類別
prompt = PromptTemplate(
input_variables=["text"],
template="將以下文本分類到以下類別之一:新聞、博客、研究、其他。\n\n文本:{text}\n\n類別:"
)
# 使用狀態中的文本格式化提示并包裝為HumanMessage
message = HumanMessage(content=prompt.format(text=state["text"]))
# 調用語言模型進行分類
classification = llm.invoke([message]).content.strip()
# 返回分類結果字典
return {"classification": classification}
此函數使用提示模板向我們的 AI 模型發出清晰的指令。該函數獲取當前狀態(包含我們正在分析的文本)并返回其分類。
實體提取功能
接下來是我們的實體提取功能:
def entity_extraction_node(state: State):
# 從文本中識別并提取命名實體(按人物、組織、地點分類)
# 創建實體提取提示模板,要求返回逗號分隔的實體列表
prompt = PromptTemplate(
input_variables=["text"],
template="從以下文本中提取所有實體(人物、組織、地點)。以逗號分隔列表形式返回結果。\n\n文本:{text}\n\n實體:"
)
# 使用文本格式化提示并包裝為HumanMessage
message = HumanMessage(content=prompt.format(text=state["text"]))
# 調用模型獲取響應,清理空白并拆分為列表
entities = llm.invoke([message]).content.strip().split(", ")
# 返回包含實體列表的字典
return {"entities": entities}
此函數處理文檔并返回關鍵實體列表,例如重要名稱、組織和地點。
摘要功能
最后,我們的摘要功能
def summarize_node(state: State):
# 創建摘要提示模板,要求用一句話總結輸入文本
# 使用模板創建鏈式處理流程(提示模板 → 語言模型)
summarization_prompt = PromptTemplate.from_template(
"""用一句話總結以下文本。\n\n文本:{text}\n\n摘要:"""
)
# 構建處理鏈:提示模板 → 語言模型
chain = summarization_prompt | llm
# 執行處理流程,傳入文本進行摘要生成
response = chain.invoke({"text": state["text"]})
# 返回包含摘要結果的字典
return {"summary": response.content}
此功能將文檔提煉為要點的簡明摘要。
這些技能相結合,使我們的代理能夠理解內容類型、識別關鍵信息并創建易于理解的摘要——每個功能都遵循相同的模式:獲取當前狀態、進行處理,并將有用信息返回給下一個功能
7. 創建狀態圖對象
# 創建狀態圖對象
workflow = StateGraph(State)
# 添加處理節點到流程圖
workflow.add_node("classification_node", classification_node) # 分類節點
workflow.add_node("entity_extraction", entity_extraction_node) # 實體提取節點
workflow.add_node("summarization", summarize_node) # 摘要生成節點
# 設置流程入口點并定義執行路徑
workflow.set_entry_point("classification_node") # 設置初始執行節點
workflow.add_edge("classification_node", "entity_extraction") # 分類 → 實體提取
workflow.add_edge("entity_extraction", "summarization") # 實體提取 → 摘要生成
workflow.add_edge("summarization", END) # 摘要生成 → 流程結束
# 編譯工作流為可執行應用
app = workflow.compile()
OK,到之類我們就已經構建了一個完整的用于文章歸納總結摘要的 Agent,它能夠按照協調的順序完成從分類到實體提取再到摘要的整個過程,使其能夠理解文本類型、識別重要實體、創建摘要,然后完成整個流程。
8. 實戰一下
現在讓我們用示例文本測試我們的代理:
# 測試用例:關于Anthropic的MCP技術說明文本
sample_text = """
Anthropic的MCP(Model Context Protocol)是一個開源的強大工具,允許應用程序與各種系統的API進行無縫交互。
"""
# 創建初始狀態對象
state_input = {"text": sample_text}
# 執行完整工作流處理
result = app.invoke(state_input)
# 輸出處理結果
print("分類結果:", result["classification"])
print("\n實體列表:", result["entities"])
print("\n摘要內容:", result["summary"])
運行結果
分類結果: 研究
理由:該文本提到了Anthropic的MCP(Model Context Protocol),并描述了其功能和特性。這種內容通常與技術研究、創新工具或方法相關,因此分類為“研究”較為合適。如果該文本出現在具體的產品新聞或個人觀點分享的語境下,也可能屬于其他類別,但從當前表述來看,“研究”是最貼切的分類。
實體列表: ['Anthropic', 'MCP', 'Model Context Protocol']
摘要內容: Anthropic的MCP是一個開源工具,可實現應用程序與系統API的無縫交互。
最后
隨著大模型能力的不斷突破,未來 AI Agent 將深度融入社會生產與生活,成為數字化轉型的核心驅動力。開發者與企業需關注場景適配性、數據安全和人機協作,以最大化其價值。