建動態(tài)擴(kuò)展的AI智能體)
在實(shí)際 AI 應(yīng)用開發(fā)中構(gòu)建一個能理解用戶意圖、調(diào)用工具并完成復(fù)雜任務(wù)的智能體Agent是核心挑戰(zhàn)。傳統(tǒng)的 LangChain Agent 框架雖然提供了基礎(chǔ)范式但在工具擴(kuò)展性、協(xié)議標(biāo)準(zhǔn)化和技能復(fù)用性上仍存在瓶頸。當(dāng)我們將 LangChain Agent 與模型上下文協(xié)議Model Context Protocol, MCP以及標(biāo)準(zhǔn)化的技能Skills體系相結(jié)合時Agent 的能力邊界和工程效率將得到顯著躍升。這種集成不僅讓 Agent 能動態(tài)接入海量外部工具和數(shù)據(jù)源還能實(shí)現(xiàn)技能的模塊化開發(fā)與跨項(xiàng)目復(fù)用為基于 Claude、GPT 等大模型構(gòu)建更強(qiáng)大、更穩(wěn)定的 AI 應(yīng)用提供了堅(jiān)實(shí)的技術(shù)底座。本文旨在深入解析 LangChain Agent 接入 MCP 與 Skills 的技術(shù)原理并通過一個從環(huán)境搭建到生產(chǎn)級實(shí)踐的全流程示例展示如何利用這套技術(shù)棧全方位提升開發(fā)效率與應(yīng)用能力。我們將從核心概念入手逐步完成一個能查詢數(shù)據(jù)庫、調(diào)用 API 并處理文件的智能體構(gòu)建并深入探討其中的配置細(xì)節(jié)、常見陷阱及性能優(yōu)化策略。1. 理解 MCP 與 SkillsAgent 能力擴(kuò)展的基石在深入代碼之前必須厘清 MCP 和 Skills 這兩個核心概念它們共同構(gòu)成了現(xiàn)代 Agent 擴(kuò)展能力的協(xié)議層和模塊層。1.1 模型上下文協(xié)議MCP是什么模型上下文協(xié)議是一種開放協(xié)議用于在大語言模型LLM與外部工具、數(shù)據(jù)源之間建立標(biāo)準(zhǔn)化的通信橋梁。你可以將其理解為 LLM 世界的“USB 協(xié)議”或“驅(qū)動程序接口標(biāo)準(zhǔn)”。在沒有 MCP 之前每個工具都需要為不同的 Agent 框架如 LangChain、LlamaIndex編寫特定的適配器代碼導(dǎo)致重復(fù)勞動和兼容性問題。MCP 的核心價值在于解耦與標(biāo)準(zhǔn)化對模型/Agent 而言它只需實(shí)現(xiàn) MCP 客戶端就能接入任何遵循 MCP 協(xié)議的服務(wù)端Server所提供的工具無需關(guān)心工具的具體實(shí)現(xiàn)。對工具開發(fā)者而言只需將工具包裝成一個 MCP 服務(wù)端就能被所有支持 MCP 的 Agent 框架使用極大地?cái)U(kuò)展了工具的受眾。一個典型的 MCP 服務(wù)端會通過標(biāo)準(zhǔn)接口向客戶端“公布”自己提供了哪些工具Tools、數(shù)據(jù)源Resources以及提示詞模板Prompts。例如一個數(shù)據(jù)庫 MCP 服務(wù)端可能公布一個“執(zhí)行 SQL 查詢”的工具一個天氣 API 的 MCP 服務(wù)端可能公布一個“獲取城市天氣”的工具。1.2 Skills可復(fù)用的能力模塊Skills技能是比工具Tools更高一層的抽象。一個 Skill 通常是為了完成一個特定領(lǐng)域任務(wù)而打包的一組工具、提示詞、工作流程甚至小模型。如果說工具是“螺絲刀”那么技能就是“組裝家具的完整工具箱和說明書”。在 LangChain 的生態(tài)中Skills 強(qiáng)調(diào)可復(fù)用性和組合性。例如“數(shù)據(jù)分析”技能可能包含數(shù)據(jù)加載、清洗、可視化和報告生成等多個工具和預(yù)設(shè)提示詞。通過將 Skills 與 MCP 結(jié)合我們可以實(shí)現(xiàn)動態(tài)發(fā)現(xiàn)與加載Agent 在運(yùn)行時可以通過 MCP 發(fā)現(xiàn)并加載遠(yuǎn)端服務(wù)器上的 Skills。版本管理與共享Skills 可以像軟件包一樣進(jìn)行版本管理并在團(tuán)隊(duì)或社區(qū)內(nèi)共享。熱插拔無需重啟 Agent 服務(wù)即可動態(tài)添加或移除 Skills實(shí)現(xiàn)能力的靈活伸縮。1.3 LangChain Agent 的工作范式LangChain Agent 的核心思想是“推理-執(zhí)行”循環(huán)。Agent 內(nèi)部有一個 LLM 作為“大腦”它根據(jù)用戶輸入和當(dāng)前上下文決定下一步是直接回答還是調(diào)用某個工具。調(diào)用工具后工具的執(zhí)行結(jié)果會返回給 LLMLLM 再據(jù)此決定后續(xù)動作直到任務(wù)完成或達(dá)到終止條件。傳統(tǒng)的 LangChain Agent 在工具管理上相對靜態(tài)通常需要在代碼中顯式定義并傳入一個工具列表。而接入 MCP 后Agent 的工具列表可以動態(tài)地從多個 MCP 服務(wù)端獲取實(shí)現(xiàn)了工具管理的“云原生”化。2. 環(huán)境準(zhǔn)備與核心依賴配置為了構(gòu)建一個接入 MCP 與 Skills 的 LangChain Agent我們需要搭建一個包含客戶端、服務(wù)端和技能庫的完整開發(fā)環(huán)境。2.1 基礎(chǔ)環(huán)境與 Python 包管理建議使用 Python 3.10 或更高版本并使用虛擬環(huán)境隔離依賴。# 創(chuàng)建并激活虛擬環(huán)境以 conda 為例 conda create -n langchain-mcp-agent python3.10 conda activate langchain-mcp-agent # 使用 pip 安裝核心依賴 pip install langchain langchain-community langchain-core2.2 安裝 MCP 相關(guān) SDKMCP 的實(shí)現(xiàn)通常包含客戶端庫和服務(wù)端開發(fā)庫。我們將使用mcp這個 Python SDK它提供了開發(fā) MCP 組件所需的核心功能。# 安裝 MCP SDK pip install mcp # 安裝 LangChain 與 MCP 的集成庫如果官方或社區(qū)有提供 # 例如一個可能的集成包請根據(jù)實(shí)際生態(tài)調(diào)整 pip install langchain-mcp-integration注意MCP 生態(tài)仍在快速發(fā)展中具體的集成庫名稱可能變化。關(guān)鍵在于找到或?qū)崿F(xiàn)一個能將 MCP 服務(wù)端提供的工具轉(zhuǎn)換為 LangChainTool對象的適配器。2.3 安裝示例 Skills 與工具服務(wù)端為了進(jìn)行演示我們需要一些實(shí)際的 MCP 服務(wù)端來提供工具。這里以兩個常見的服務(wù)端為例文件系統(tǒng)服務(wù)端提供讀取、寫入、列出文件等工具。SQLite 數(shù)據(jù)庫服務(wù)端提供執(zhí)行 SQL 查詢的工具。我們可以從社區(qū)尋找或自己實(shí)現(xiàn)這些服務(wù)端。假設(shè)我們使用一個名為mcp-server-filesystem和mcp-server-sqlite的包。# 安裝示例 MCP 服務(wù)端假設(shè)的包名請?zhí)鎿Q為實(shí)際可用的包 pip install mcp-server-filesystem mcp-server-sqlite2.4 配置大模型訪問本文以 Anthropic 的 Claude 模型為例你需要準(zhǔn)備相應(yīng)的 API 密鑰。其他模型如 OpenAI GPT 的配置邏輯類似。# 安裝 Claude SDK pip install anthropic在項(xiàng)目根目錄創(chuàng)建.env文件來管理敏感配置# .env ANTHROPIC_API_KEYyour_anthropic_api_key_here在代碼中通過python-dotenv加載# config.py import os from dotenv import load_dotenv load_dotenv() ANTHROPIC_API_KEY os.getenv(ANTHROPIC_API_KEY) if not ANTHROPIC_API_KEY: raise ValueError(請?jiān)?.env 文件中設(shè)置 ANTHROPIC_API_KEY)3. 構(gòu)建一個動態(tài)工具發(fā)現(xiàn)的 LangChain Agent本節(jié)將分步構(gòu)建一個核心 Agent它能夠從本地運(yùn)行的多個 MCP 服務(wù)端動態(tài)發(fā)現(xiàn)工具并利用 Claude 模型進(jìn)行推理和調(diào)用。3.1 啟動并連接 MCP 服務(wù)端首先我們需要在后臺啟動 MCP 服務(wù)端進(jìn)程。在實(shí)際部署中這些服務(wù)端可能以獨(dú)立進(jìn)程、容器或遠(yuǎn)程服務(wù)的形式存在。這里我們在同一臺機(jī)器上以子進(jìn)程方式啟動它們。# mcp_servers.py import subprocess import time import signal import sys class MCPServerManager: def __init__(self): self.servers [] def start_file_server(self): 啟動文件系統(tǒng) MCP 服務(wù)端 # 假設(shè)服務(wù)端通過命令 mcp-server-filesystem 啟動監(jiān)聽 8001 端口 cmd [mcp-server-filesystem, --root, ./data, --port, 8001] proc subprocess.Popen(cmd, stdoutsubprocess.PIPE, stderrsubprocess.PIPE) self.servers.append((filesystem, proc, 8001)) time.sleep(2) # 等待服務(wù)端啟動 print(f文件系統(tǒng) MCP 服務(wù)端已啟動 (PID: {proc.pid})) return proc def start_sqlite_server(self, db_path./data/example.db): 啟動 SQLite MCP 服務(wù)端 # 假設(shè)服務(wù)端通過命令 mcp-server-sqlite 啟動監(jiān)聽 8002 端口 cmd [mcp-server-sqlite, --db, db_path, --port, 8002] proc subprocess.Popen(cmd, stdoutsubprocess.PIPE, stderrsubprocess.PIPE) self.servers.append((sqlite, proc, 8002)) time.sleep(2) print(fSQLite MCP 服務(wù)端已啟動 (PID: {proc.pid})) return proc def stop_all(self): 停止所有 MCP 服務(wù)端 for name, proc, _ in self.servers: print(f正在停止 {name} 服務(wù)端...) proc.terminate() proc.wait() self.servers.clear() # 使用上下文管理器確保資源清理 if __name__ __main__: manager MCPServerManager() try: manager.start_file_server() manager.start_sqlite_server() print(所有 MCP 服務(wù)端已就緒按 CtrlC 停止...) while True: time.sleep(1) except KeyboardInterrupt: manager.stop_all()3.2 實(shí)現(xiàn) MCP 客戶端并轉(zhuǎn)換為 LangChain Tools這是最關(guān)鍵的一步我們需要編寫一個 MCP 客戶端連接到服務(wù)端獲取其提供的工具列表并將每個工具包裝成 LangChain 能識別的Tool對象。# mcp_client.py import asyncio from typing import List, Optional from langchain.tools import BaseTool from langchain_core.tools import Tool from mcp import ClientSession, StdioServerParameters from mcp.client import stdio class MCPToolFetcher: def __init__(self, server_name: str, server_params: StdioServerParameters): self.server_name server_name self.server_params server_params self.tools: List[Tool] [] async def connect_and_fetch_tools(self): 連接到 MCP 服務(wù)端并獲取工具列表 # 創(chuàng)建與 MCP 服務(wù)端的會話 async with stdio.stdio_client(self.server_params) as (read, write): async with ClientSession(read, write) as session: # 初始化連接 await session.initialize() # 列出服務(wù)端提供的所有工具 response await session.list_tools() mcp_tools response.tools for mcp_tool in mcp_tools: # 為每個 MCP 工具創(chuàng)建一個 LangChain Tool 包裝器 langchain_tool Tool( namef{self.server_name}_{mcp_tool.name}, funcself._create_tool_func(session, mcp_tool), descriptionmcp_tool.description, ) self.tools.append(langchain_tool) return self.tools def _create_tool_func(self, session, mcp_tool): 創(chuàng)建一個能調(diào)用特定 MCP 工具的同步函數(shù) # 注意LangChain Tool 的 func 是同步的但 MCP 調(diào)用是異步的。 # 我們需要在同步函數(shù)中運(yùn)行異步代碼。這里使用 asyncio.run 簡化處理 # 在生產(chǎn)環(huán)境中需要考慮更優(yōu)的異步集成方式。 async def async_tool_func(**kwargs): # 調(diào)用 MCP 工具的 execute 方法 result await session.call_tool(mcp_tool.name, argumentskwargs) # 返回工具執(zhí)行結(jié)果的文本內(nèi)容 return \n.join([c.text for c in result.content if c.type text]) def sync_wrapper(**kwargs): # 在新的事件循環(huán)中運(yùn)行異步函數(shù)適用于簡單腳本 # 注意在已有事件循環(huán)的環(huán)境中如 FastAPI需要使用其他方式 return asyncio.run(async_tool_func(**kwargs)) return sync_wrapper # 工具獲取工具函數(shù) def get_all_mcp_tools() - List[Tool]: 獲取所有已配置 MCP 服務(wù)端的工具 all_tools [] # 定義服務(wù)端連接參數(shù) servers [ (filesystem, StdioServerParameters(commandmcp-server-filesystem, args[--root, ./data])), (sqlite, StdioServerParameters(commandmcp-server-sqlite, args[--db, ./data/example.db])), ] async def fetch_all(): for name, params in servers: fetcher MCPToolFetcher(name, params) tools await fetcher.connect_and_fetch_tools() all_tools.extend(tools) # 運(yùn)行異步函數(shù)獲取所有工具 asyncio.run(fetch_all()) return all_tools3.3 創(chuàng)建 LangChain Agent 并集成動態(tài)工具現(xiàn)在我們可以使用獲取到的動態(tài)工具列表來初始化一個 LangChain Agent。這里使用 ReAct 代理類型它適合多步驟的工具調(diào)用場景。# agent_builder.py from langchain.agents import AgentExecutor, create_react_agent from langchain_core.prompts import PromptTemplate from langchain_anthropic import ChatAnthropic from config import ANTHROPIC_API_KEY from mcp_client import get_all_mcp_tools def build_mcp_agent(): # 1. 初始化 Claude 模型 llm ChatAnthropic( modelclaude-3-haiku-20240307, # 可根據(jù)需要選擇 sonnet, opus 等型號 temperature0, api_keyANTHROPIC_API_KEY ) # 2. 動態(tài)獲取所有 MCP 工具 print(正在從 MCP 服務(wù)端發(fā)現(xiàn)工具...) tools get_all_mcp_tools() print(f已發(fā)現(xiàn) {len(tools)} 個工具: {[t.name for t in tools]}) # 3. 定義 ReAct 代理的提示詞模板 # 提示詞需要指導(dǎo)模型如何思考和使用工具 prompt PromptTemplate.from_template( 你是一個有幫助的AI助手可以訪問以下工具 {tools} 請使用以下格式回答 問題用戶提出的問題 思考你需要思考如何一步步解決問題。你可以使用工具也可以直接回答。 行動要使用的工具名稱必須是以下工具之一[{tool_names}] 行動輸入工具的輸入必須是一個有效的JSON字符串 觀察工具返回的結(jié)果 ... (這個 思考/行動/行動輸入/觀察 循環(huán)可以重復(fù)多次) 思考我現(xiàn)在知道了最終答案 最終答案對原始問題的最終回答 開始 問題{input} 思考{agent_scratchpad} ) # 4. 創(chuàng)建 ReAct 代理 agent create_react_agent(llm, tools, prompt) # 5. 創(chuàng)建代理執(zhí)行器控制最大迭代次數(shù)以避免無限循環(huán) agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, # 開啟詳細(xì)日志便于調(diào)試 handle_parsing_errorsTrue, # 處理模型輸出解析錯誤 max_iterations10, # 限制最大思考步驟 early_stopping_methodgenerate, # 當(dāng)模型決定不再使用工具時停止 ) return agent_executor4. 運(yùn)行驗(yàn)證與結(jié)果分析構(gòu)建好 Agent 后我們需要準(zhǔn)備測試數(shù)據(jù)運(yùn)行幾個典型任務(wù)來驗(yàn)證其能力。4.1 準(zhǔn)備測試環(huán)境與數(shù)據(jù)首先創(chuàng)建必要的目錄和測試數(shù)據(jù)。# 創(chuàng)建數(shù)據(jù)目錄和示例文件 mkdir -p ./data echo 項(xiàng)目報告草案\n主要內(nèi)容...\n待辦整理圖表 ./data/report.txt echo 會議記錄\n日期2024-05-27\n議題Agent架構(gòu)評審 ./data/meeting.txt # 創(chuàng)建并初始化一個 SQLite 示例數(shù)據(jù)庫 sqlite3 ./data/example.db EOF CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, email TEXT, department TEXT); INSERT INTO users (name, email, department) VALUES (張三, zhangsanexample.com, 工程部), (李四, lisiexample.com, 產(chǎn)品部), (王五, wangwuexample.com, 市場部); EOF4.2 執(zhí)行綜合任務(wù)測試現(xiàn)在編寫一個測試腳本讓 Agent 執(zhí)行一個結(jié)合了文件操作和數(shù)據(jù)庫查詢的復(fù)雜任務(wù)。# run_agent.py from agent_builder import build_mcp_agent import asyncio async def main(): # 構(gòu)建 Agent agent build_mcp_agent() # 測試任務(wù) 1簡單的文件讀取 print(\n 測試任務(wù) 1讀取文件 ) result1 await agent.ainvoke({input: 請讀取 data 目錄下 report.txt 文件的內(nèi)容并總結(jié)其要點(diǎn)。}) print(f任務(wù)1結(jié)果: {result1[output]}) # 測試任務(wù) 2數(shù)據(jù)庫查詢 print(\n 測試任務(wù) 2查詢數(shù)據(jù)庫 ) result2 await agent.ainvoke({input: 查詢 example.db 數(shù)據(jù)庫中 users 表里所有在工程部的員工姓名和郵箱。}) print(f任務(wù)2結(jié)果: {result2[output]}) # 測試任務(wù) 3多步驟組合任務(wù) print(\n 測試任務(wù) 3組合任務(wù) ) result3 await agent.ainvoke({ input: 1. 首先請列出 data 目錄下所有的 .txt 文件。 2. 然后讀取 meeting.txt 文件提取會議日期。 3. 最后去 example.db 數(shù)據(jù)庫的 users 表里查一下產(chǎn)品部有哪些人把結(jié)果和會議日期一起整理成一個簡短的摘要。 }) print(f任務(wù)3結(jié)果: {result3[output]}) if __name__ __main__: asyncio.run(main())運(yùn)行此腳本你應(yīng)該能看到類似以下的輸出具體內(nèi)容因模型隨機(jī)性略有不同正在從 MCP 服務(wù)端發(fā)現(xiàn)工具... 已發(fā)現(xiàn) 4 個工具: [filesystem_read_file, filesystem_list_directory, sqlite_execute_query, sqlite_list_tables] 測試任務(wù) 1讀取文件 進(jìn)入新的 Agent 執(zhí)行鏈... 思考用戶要求讀取 report.txt 文件并總結(jié)要點(diǎn)。我需要使用文件讀取工具。 行動filesystem_read_file 行動輸入{path: ./data/report.txt} 觀察項(xiàng)目報告草案\n主要內(nèi)容...\n待辦整理圖表 思考我已讀取文件內(nèi)容?，F(xiàn)在需要總結(jié)要點(diǎn)。內(nèi)容顯示這是一個項(xiàng)目報告草案包含主要內(nèi)容和待辦事項(xiàng)整理圖表。我可以直接總結(jié)。 最終答案該文件是一個項(xiàng)目報告草案主要內(nèi)容已列出當(dāng)前待辦事項(xiàng)是整理圖表。 任務(wù)1結(jié)果該文件是一個項(xiàng)目報告草案主要內(nèi)容已列出當(dāng)前待辦事項(xiàng)是整理圖表。 測試任務(wù) 2查詢數(shù)據(jù)庫 進(jìn)入新的 Agent 執(zhí)行鏈... 思考用戶需要查詢工程部員工。我需要使用 SQLite 查詢工具。 行動sqlite_execute_query 行動輸入{query: SELECT name, email FROM users WHERE department 工程部} 觀察[{name: 張三, email: zhangsanexample.com}] 思考查詢返回了結(jié)果。我可以直接給出答案。 最終答案工程部的員工是張三郵箱是 zhangsanexample.com。 任務(wù)2結(jié)果工程部的員工是張三郵箱是 zhangsanexample.com。從輸出中我們可以看到 Agent 成功完成了以下工作動態(tài)工具發(fā)現(xiàn)啟動時從兩個 MCP 服務(wù)端獲取了 4 個工具。正確工具選擇針對“讀文件”任務(wù)選擇了filesystem_read_file針對“查數(shù)據(jù)庫”任務(wù)選擇了sqlite_execute_query。參數(shù)構(gòu)造能夠根據(jù)任務(wù)描述正確構(gòu)造工具所需的輸入?yún)?shù)如文件路徑、SQL 語句。結(jié)果解析與總結(jié)能夠理解工具返回的原始數(shù)據(jù)文本行、JSON 數(shù)組并將其組織成自然語言回答。4.3 關(guān)鍵配置參數(shù)解析在構(gòu)建 Agent 時有幾個關(guān)鍵參數(shù)直接影響其行為和性能參數(shù)所在位置含義與影響推薦值/建議max_iterationsAgentExecutor代理最大推理-執(zhí)行循環(huán)次數(shù)。防止任務(wù)過于復(fù)雜導(dǎo)致無限循環(huán)。簡單任務(wù) 5-10復(fù)雜任務(wù) 15-20。需結(jié)合max_tokens考慮。handle_parsing_errorsAgentExecutor是否處理模型輸出格式解析錯誤。開啟后解析失敗會嘗試讓模型重試。建議始終設(shè)為True提高魯棒性。verboseAgentExecutor是否打印詳細(xì)的思考鏈Chain of Thought日志。開發(fā)調(diào)試時設(shè)為True生產(chǎn)環(huán)境設(shè)為False。temperatureChatAnthropic模型生成文本的隨機(jī)性。值越高輸出越多樣、越有創(chuàng)造性。Agent 工具調(diào)用場景建議設(shè)為0或0.1以保證工具選擇和參數(shù)生成的穩(wěn)定性。modelChatAnthropic使用的 Claude 模型版本。claude-3-haiku速度最快成本最低claude-3-sonnet平衡claude-3-opus能力最強(qiáng)但最慢最貴。根據(jù)任務(wù)復(fù)雜度選擇。5. 常見問題排查與性能優(yōu)化將 LangChain Agent 與 MCP、Skills 集成到生產(chǎn)環(huán)境時會遇到一系列工程化挑戰(zhàn)。以下是典型問題的排查路徑和優(yōu)化建議。5.1 連接與工具發(fā)現(xiàn)失敗現(xiàn)象Agent 啟動時報錯無法連接到 MCP 服務(wù)端或工具列表為空。可能原因檢查方式解決方案MCP 服務(wù)端未啟動檢查對應(yīng)端口如 8001, 8002是否在監(jiān)聽 (netstat -an | grep 8001)。查看MCPServerManager日志是否有啟動錯誤。確保啟動命令正確依賴已安裝。檢查服務(wù)端二進(jìn)制文件路徑。命令或參數(shù)錯誤檢查StdioServerParameters中的command和args是否與服務(wù)端程序匹配。使用絕對路徑指定命令或確保命令在系統(tǒng) PATH 中。參考服務(wù)端文檔確認(rèn)參數(shù)格式。權(quán)限問題檢查服務(wù)端是否有權(quán)限訪問指定目錄如./data或數(shù)據(jù)庫文件。調(diào)整目錄權(quán)限或使用服務(wù)端可訪問的路徑。協(xié)議版本不兼容查看 MCP 客戶端和服務(wù)端的版本。檢查初始化握手階段的錯誤信息。確?？蛻舳撕头?wù)端使用的mcpSDK 版本兼容?？蓢L試升級到最新穩(wěn)定版。5.2 工具調(diào)用錯誤或超時現(xiàn)象Agent 選擇了正確的工具但調(diào)用失敗或長時間無響應(yīng)。問題現(xiàn)象常見原因檢查方式處理建議工具參數(shù)格式錯誤模型生成的 JSON 參數(shù)不符合工具要求。查看verbose日志中的“行動輸入”字段。手動用相同參數(shù)測試工具。在提示詞中更清晰地描述工具所需的參數(shù)格式。使用 Pydantic 模型對工具輸入進(jìn)行校驗(yàn)。工具執(zhí)行內(nèi)部錯誤MCP 服務(wù)端在處理請求時崩潰或返回錯誤。查看 MCP 服務(wù)端進(jìn)程的標(biāo)準(zhǔn)錯誤輸出。檢查服務(wù)端日志。確保輸入數(shù)據(jù)如 SQL 語法、文件路徑有效。在工具包裝函數(shù)中添加更詳細(xì)的錯誤捕獲和日志。網(wǎng)絡(luò)或進(jìn)程通信超時服務(wù)端響應(yīng)慢或進(jìn)程僵死。在工具調(diào)用代碼處添加超時設(shè)置。監(jiān)控服務(wù)端資源占用CPU/內(nèi)存。為異步調(diào)用設(shè)置asyncio.wait_for超時。優(yōu)化服務(wù)端性能或?qū)臅r工具單獨(dú)設(shè)置更長的超時。異步上下文沖突在已有事件循環(huán)如 FastAPI中同步調(diào)用工具導(dǎo)致錯誤。觀察是否報錯RuntimeError: This event loop is already running。避免在同步函數(shù)中直接使用asyncio.run。改用asyncio.create_task或在主異步上下文中調(diào)用工具。重構(gòu)代碼使整個 Agent 調(diào)用鏈保持異步。5.3 Agent 邏輯錯誤與優(yōu)化現(xiàn)象Agent 陷入循環(huán)、選擇錯誤工具、或生成無關(guān)內(nèi)容。問題根因分析優(yōu)化策略工具選擇不準(zhǔn)1. 工具描述 (description) 不夠清晰。2. 提示詞未充分指導(dǎo)模型如何選擇工具。3. 工具過多模型混淆。1.優(yōu)化工具描述用自然語言清晰說明工具功能、輸入輸出示例。例如將sqlite_execute_query描述改為“執(zhí)行一條 SQL SELECT 查詢語句并返回結(jié)果集。輸入應(yīng)為包含query鍵的 JSON 對象?！?.改進(jìn)提示詞在PromptTemplate中加入工具選擇范例。3.工具分組/路由對工具進(jìn)行分類先讓 Agent 選擇大類再選擇具體工具。無效迭代過多模型在已經(jīng)獲得答案的情況下仍繼續(xù)嘗試使用工具。1.調(diào)整max_iterations根據(jù)任務(wù)復(fù)雜度設(shè)置合理上限。2.使用更好的停止條件AgentExecutor的early_stopping_method設(shè)為generate讓模型自己決定何時停止。3.優(yōu)化思考鏈在提示詞中強(qiáng)調(diào)“當(dāng)你認(rèn)為已有足夠信息回答問題時可以直接給出最終答案”。處理復(fù)雜任務(wù)能力弱單一 ReAct 代理難以規(guī)劃冗長或多分支任務(wù)。1.升級 Agent 類型使用Plan-and-Execute或OpenAI Functions代理它們更擅長規(guī)劃。2.引入 LangGraph對于有狀態(tài)、多分支的工作流使用 LangGraph 來顯式定義狀態(tài)圖和節(jié)點(diǎn)邏輯。3.任務(wù)分解在上層設(shè)計(jì)一個“主控”Agent負(fù)責(zé)將復(fù)雜任務(wù)拆解為子任務(wù)再分發(fā)給負(fù)責(zé)具體工具的“子”Agent。5.4 生產(chǎn)環(huán)境部署建議在開發(fā)環(huán)境跑通后部署到生產(chǎn)環(huán)境還需考慮以下方面MCP 服務(wù)端部署不應(yīng)以簡單的子進(jìn)程方式運(yùn)行。建議將每個 MCP 服務(wù)端部署為獨(dú)立的容器Docker或系統(tǒng)服務(wù)systemd并配置健康檢查、資源限制和自動重啟。連接管理與池化頻繁創(chuàng)建銷毀到 MCP 服務(wù)端的連接開銷大。應(yīng)實(shí)現(xiàn)連接池讓多個 Agent 實(shí)例共享到同一服務(wù)端的穩(wěn)定連接。安全性工具權(quán)限嚴(yán)格限制每個 MCP 服務(wù)端的權(quán)限。例如文件系統(tǒng)服務(wù)端只允許訪問特定沙箱目錄數(shù)據(jù)庫服務(wù)端使用只讀或最小權(quán)限賬戶。輸入驗(yàn)證與清理對所有從模型傳遞給工具的參數(shù)進(jìn)行嚴(yán)格的驗(yàn)證和清理防止 SQL 注入、路徑遍歷等攻擊。API 密鑰管理使用安全的秘密管理服務(wù)如 Vault, AWS Secrets Manager存儲和輪換 API 密鑰切勿硬編碼在代碼或配置文件中?？捎^測性結(jié)構(gòu)化日志記錄每個工具調(diào)用的詳細(xì)信息工具名、輸入、輸出、耗時、狀態(tài)。鏈路追蹤為每個用戶會話分配唯一 ID并貫穿所有的 Agent 思考、工具調(diào)用步驟便于問題排查。監(jiān)控指標(biāo)監(jiān)控 Agent 的請求量、響應(yīng)時間、工具調(diào)用成功率、迭代次數(shù)分布等。性能與成本緩存對頻繁且結(jié)果不變的工具調(diào)用如讀取靜態(tài)配置添加緩存層。模型選擇根據(jù)任務(wù)類型選擇合適的模型。簡單的工具調(diào)用可用Haiku復(fù)雜規(guī)劃可用Sonnet或Opus。限制 Token 消耗設(shè)置max_tokens上限防止因異常導(dǎo)致生成過長內(nèi)容而產(chǎn)生高費(fèi)用。6. 擴(kuò)展方向構(gòu)建自定義 Skills 與高級工作流掌握了基礎(chǔ)集成后你可以向兩個方向深入一是創(chuàng)建自己的 Skills二是構(gòu)建更復(fù)雜的智能工作流。6.1 開發(fā)自定義 MCP 服務(wù)端與 Skill一個 Skill 本質(zhì)上是一個或多個相關(guān)工具的集合并可能附帶一些預(yù)設(shè)提示詞。開發(fā)自定義 Skill 的步驟如下定義工具接口明確 Skill 要提供哪些功能。實(shí)現(xiàn) MCP 服務(wù)端使用mcpSDK 實(shí)現(xiàn)這些功能并遵循 MCP 協(xié)議暴露它們。打包與分發(fā)將服務(wù)端代碼和配置打包如 Docker 鏡像、Python 包方便部署和共享。以下是一個簡單的“天氣查詢” Skill 的服務(wù)端示例# weather_mcp_server.py from mcp.server import Server, NotificationOptions from mcp.server.models import InitializationOptions import mcp.server.stdio import asyncio from some_weather_lib import get_weather # 假設(shè)有一個獲取天氣的庫 async def main(): # 創(chuàng)建 MCP 服務(wù)端 server Server(weather-skills) server.list_tools() async def handle_list_tools(): # 公布此服務(wù)端提供的工具 return [ { name: get_current_weather, description: 獲取指定城市的當(dāng)前天氣情況。, inputSchema: { type: object, properties: { city: {type: string, description: 城市名稱例如北京} }, required: [city] } } ] server.call_tool() async def handle_call_tool(name: str, arguments: dict): if name get_current_weather: city arguments.get(city) if not city: raise ValueError(缺少參數(shù) city) # 調(diào)用實(shí)際的外部天氣 API weather_info await get_weather(city) return [ { type: text, text: f{city}的天氣{weather_info[condition]}溫度{weather_info[temp]}°C。 } ] else: raise ValueError(f未知工具: {name}) # 通過標(biāo)準(zhǔn)輸入輸出運(yùn)行服務(wù)端 async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions( server_nameweather-skills, server_version0.1.0 ) ) if __name__ __main__: asyncio.run(main())6.2 使用 LangGraph 編排復(fù)雜 Agent 工作流當(dāng)任務(wù)涉及多個 Agent 協(xié)作、狀態(tài)持久化或條件分支時LangChain 的基礎(chǔ) AgentExecutor 會顯得力不從心。此時LangGraph 是更強(qiáng)大的選擇。它允許你將工作流定義為圖Graph其中節(jié)點(diǎn)可以是 Agent、工具或任何函數(shù)邊定義了執(zhí)行流程。例如一個“數(shù)據(jù)報告生成”工作流可能包含以下節(jié)點(diǎn)規(guī)劃節(jié)點(diǎn)分析用戶請求拆解為“取數(shù)據(jù)”、“分析”、“生成圖表”、“撰寫報告”等子任務(wù)。數(shù)據(jù)查詢節(jié)點(diǎn)調(diào)用 SQL MCP 工具獲取數(shù)據(jù)。分析節(jié)點(diǎn)調(diào)用 Python 計(jì)算工具或另一個 LLM 進(jìn)行數(shù)據(jù)分析。圖表生成節(jié)點(diǎn)調(diào)用圖表生成 API。報告匯編節(jié)點(diǎn)將數(shù)據(jù)、分析結(jié)果、圖表整合成最終報告。使用 LangGraph你可以清晰地定義這些節(jié)點(diǎn)的執(zhí)行順序和條件分支例如如果數(shù)據(jù)為空則跳過分析節(jié)點(diǎn)并持久化整個工作流的狀態(tài)實(shí)現(xiàn)更穩(wěn)健和可調(diào)試的復(fù)雜 Agent 系統(tǒng)。通過將 LangChain Agent、MCP 協(xié)議和 Skills 模塊相結(jié)合我們構(gòu)建的智能體不再是一個封閉、僵化的系統(tǒng)而是一個能夠動態(tài)擴(kuò)展、靈活組合的開放平臺。這種架構(gòu)使得集成新工具、復(fù)用已有能力、以及構(gòu)建復(fù)雜工作流變得前所未有的高效。從簡單的文件查詢到結(jié)合數(shù)據(jù)庫、API 和自定義邏輯的復(fù)雜任務(wù)Agent 都能通過統(tǒng)一的協(xié)議層進(jìn)行調(diào)度和執(zhí)行。在向生產(chǎn)環(huán)境邁進(jìn)時務(wù)必關(guān)注安全性、可靠性和可觀測性通過連接池、權(quán)限控制、結(jié)構(gòu)化日志和監(jiān)控指標(biāo)來保障系統(tǒng)的穩(wěn)定運(yùn)行。下一步你可以嘗試開發(fā)自己的專屬 Skills或者利用 LangGraph 來設(shè)計(jì)更精巧的多智能體協(xié)作流程從而解鎖 AI 應(yīng)用開發(fā)的更大潛力。 LangChain Agent集成MCP與Skills：構(gòu)建動態(tài)擴(kuò)展的AI智能體
在實(shí)際 AI 應(yīng)用開發(fā)中構(gòu)建一個能理解用戶意圖、調(diào)用工具并完成復(fù)雜任務(wù)的智能體Agent是核心挑戰(zhàn)。傳統(tǒng)的 LangChain Agent 框架雖然提供了基礎(chǔ)范式但在工具擴(kuò)展性、協(xié)議標(biāo)準(zhǔn)化和技能復(fù)用性上仍存在瓶頸。當(dāng)我們將 LangChain Agent 與模型上下文協(xié)議Model Context Protocol, MCP以及標(biāo)準(zhǔn)化的技能Skills體系相結(jié)合時Agent 的能力邊界和工程效率將得到顯著躍升。這種集成不僅讓 Agent 能動態(tài)接入海量外部工具和數(shù)據(jù)源還能實(shí)現(xiàn)技能的模塊化開發(fā)與跨項(xiàng)目復(fù)用為基于 Claude、GPT 等大模型構(gòu)建更強(qiáng)大、更穩(wěn)定的 AI 應(yīng)用提供了堅(jiān)實(shí)的技術(shù)底座。本文旨在深入解析 LangChain Agent 接入 MCP 與 Skills 的技術(shù)原理并通過一個從環(huán)境搭建到生產(chǎn)級實(shí)踐的全流程示例展示如何利用這套技術(shù)棧全方位提升開發(fā)效率與應(yīng)用能力。我們將從核心概念入手逐步完成一個能查詢數(shù)據(jù)庫、調(diào)用 API 并處理文件的智能體構(gòu)建并深入探討其中的配置細(xì)節(jié)、常見陷阱及性能優(yōu)化策略。1. 理解 MCP 與 SkillsAgent 能力擴(kuò)展的基石在深入代碼之前必須厘清 MCP 和 Skills 這兩個核心概念它們共同構(gòu)成了現(xiàn)代 Agent 擴(kuò)展能力的協(xié)議層和模塊層。1.1 模型上下文協(xié)議MCP是什么模型上下文協(xié)議是一種開放協(xié)議用于在大語言模型LLM與外部工具、數(shù)據(jù)源之間建立標(biāo)準(zhǔn)化的通信橋梁。你可以將其理解為 LLM 世界的“USB 協(xié)議”或“驅(qū)動程序接口標(biāo)準(zhǔn)”。在沒有 MCP 之前每個工具都需要為不同的 Agent 框架如 LangChain、LlamaIndex編寫特定的適配器代碼導(dǎo)致重復(fù)勞動和兼容性問題。MCP 的核心價值在于解耦與標(biāo)準(zhǔn)化對模型/Agent 而言它只需實(shí)現(xiàn) MCP 客戶端就能接入任何遵循 MCP 協(xié)議的服務(wù)端Server所提供的工具無需關(guān)心工具的具體實(shí)現(xiàn)。對工具開發(fā)者而言只需將工具包裝成一個 MCP 服務(wù)端就能被所有支持 MCP 的 Agent 框架使用極大地?cái)U(kuò)展了工具的受眾。一個典型的 MCP 服務(wù)端會通過標(biāo)準(zhǔn)接口向客戶端“公布”自己提供了哪些工具Tools、數(shù)據(jù)源Resources以及提示詞模板Prompts。例如一個數(shù)據(jù)庫 MCP 服務(wù)端可能公布一個“執(zhí)行 SQL 查詢”的工具一個天氣 API 的 MCP 服務(wù)端可能公布一個“獲取城市天氣”的工具。1.2 Skills可復(fù)用的能力模塊Skills技能是比工具Tools更高一層的抽象。一個 Skill 通常是為了完成一個特定領(lǐng)域任務(wù)而打包的一組工具、提示詞、工作流程甚至小模型。如果說工具是“螺絲刀”那么技能就是“組裝家具的完整工具箱和說明書”。在 LangChain 的生態(tài)中Skills 強(qiáng)調(diào)可復(fù)用性和組合性。例如“數(shù)據(jù)分析”技能可能包含數(shù)據(jù)加載、清洗、可視化和報告生成等多個工具和預(yù)設(shè)提示詞。通過將 Skills 與 MCP 結(jié)合我們可以實(shí)現(xiàn)動態(tài)發(fā)現(xiàn)與加載Agent 在運(yùn)行時可以通過 MCP 發(fā)現(xiàn)并加載遠(yuǎn)端服務(wù)器上的 Skills。版本管理與共享Skills 可以像軟件包一樣進(jìn)行版本管理并在團(tuán)隊(duì)或社區(qū)內(nèi)共享。熱插拔無需重啟 Agent 服務(wù)即可動態(tài)添加或移除 Skills實(shí)現(xiàn)能力的靈活伸縮。1.3 LangChain Agent 的工作范式LangChain Agent 的核心思想是“推理-執(zhí)行”循環(huán)。Agent 內(nèi)部有一個 LLM 作為“大腦”它根據(jù)用戶輸入和當(dāng)前上下文決定下一步是直接回答還是調(diào)用某個工具。調(diào)用工具后工具的執(zhí)行結(jié)果會返回給 LLMLLM 再據(jù)此決定后續(xù)動作直到任務(wù)完成或達(dá)到終止條件。傳統(tǒng)的 LangChain Agent 在工具管理上相對靜態(tài)通常需要在代碼中顯式定義并傳入一個工具列表。而接入 MCP 后Agent 的工具列表可以動態(tài)地從多個 MCP 服務(wù)端獲取實(shí)現(xiàn)了工具管理的“云原生”化。2. 環(huán)境準(zhǔn)備與核心依賴配置為了構(gòu)建一個接入 MCP 與 Skills 的 LangChain Agent我們需要搭建一個包含客戶端、服務(wù)端和技能庫的完整開發(fā)環(huán)境。2.1 基礎(chǔ)環(huán)境與 Python 包管理建議使用 Python 3.10 或更高版本并使用虛擬環(huán)境隔離依賴。# 創(chuàng)建并激活虛擬環(huán)境以 conda 為例 conda create -n langchain-mcp-agent python3.10 conda activate langchain-mcp-agent # 使用 pip 安裝核心依賴 pip install langchain langchain-community langchain-core2.2 安裝 MCP 相關(guān) SDKMCP 的實(shí)現(xiàn)通常包含客戶端庫和服務(wù)端開發(fā)庫。我們將使用mcp這個 Python SDK它提供了開發(fā) MCP 組件所需的核心功能。# 安裝 MCP SDK pip install mcp # 安裝 LangChain 與 MCP 的集成庫如果官方或社區(qū)有提供 # 例如一個可能的集成包請根據(jù)實(shí)際生態(tài)調(diào)整 pip install langchain-mcp-integration注意MCP 生態(tài)仍在快速發(fā)展中具體的集成庫名稱可能變化。關(guān)鍵在于找到或?qū)崿F(xiàn)一個能將 MCP 服務(wù)端提供的工具轉(zhuǎn)換為 LangChainTool對象的適配器。2.3 安裝示例 Skills 與工具服務(wù)端為了進(jìn)行演示我們需要一些實(shí)際的 MCP 服務(wù)端來提供工具。這里以兩個常見的服務(wù)端為例文件系統(tǒng)服務(wù)端提供讀取、寫入、列出文件等工具。SQLite 數(shù)據(jù)庫服務(wù)端提供執(zhí)行 SQL 查詢的工具。我們可以從社區(qū)尋找或自己實(shí)現(xiàn)這些服務(wù)端。假設(shè)我們使用一個名為mcp-server-filesystem和mcp-server-sqlite的包。# 安裝示例 MCP 服務(wù)端假設(shè)的包名請?zhí)鎿Q為實(shí)際可用的包 pip install mcp-server-filesystem mcp-server-sqlite2.4 配置大模型訪問本文以 Anthropic 的 Claude 模型為例你需要準(zhǔn)備相應(yīng)的 API 密鑰。其他模型如 OpenAI GPT 的配置邏輯類似。# 安裝 Claude SDK pip install anthropic在項(xiàng)目根目錄創(chuàng)建.env文件來管理敏感配置# .env ANTHROPIC_API_KEYyour_anthropic_api_key_here在代碼中通過python-dotenv加載# config.py import os from dotenv import load_dotenv load_dotenv() ANTHROPIC_API_KEY os.getenv(ANTHROPIC_API_KEY) if not ANTHROPIC_API_KEY: raise ValueError(請?jiān)?.env 文件中設(shè)置 ANTHROPIC_API_KEY)3. 構(gòu)建一個動態(tài)工具發(fā)現(xiàn)的 LangChain Agent本節(jié)將分步構(gòu)建一個核心 Agent它能夠從本地運(yùn)行的多個 MCP 服務(wù)端動態(tài)發(fā)現(xiàn)工具并利用 Claude 模型進(jìn)行推理和調(diào)用。3.1 啟動并連接 MCP 服務(wù)端首先我們需要在后臺啟動 MCP 服務(wù)端進(jìn)程。在實(shí)際部署中這些服務(wù)端可能以獨(dú)立進(jìn)程、容器或遠(yuǎn)程服務(wù)的形式存在。這里我們在同一臺機(jī)器上以子進(jìn)程方式啟動它們。# mcp_servers.py import subprocess import time import signal import sys class MCPServerManager: def __init__(self): self.servers [] def start_file_server(self): 啟動文件系統(tǒng) MCP 服務(wù)端 # 假設(shè)服務(wù)端通過命令 mcp-server-filesystem 啟動監(jiān)聽 8001 端口 cmd [mcp-server-filesystem, --root, ./data, --port, 8001] proc subprocess.Popen(cmd, stdoutsubprocess.PIPE, stderrsubprocess.PIPE) self.servers.append((filesystem, proc, 8001)) time.sleep(2) # 等待服務(wù)端啟動 print(f文件系統(tǒng) MCP 服務(wù)端已啟動 (PID: {proc.pid})) return proc def start_sqlite_server(self, db_path./data/example.db): 啟動 SQLite MCP 服務(wù)端 # 假設(shè)服務(wù)端通過命令 mcp-server-sqlite 啟動監(jiān)聽 8002 端口 cmd [mcp-server-sqlite, --db, db_path, --port, 8002] proc subprocess.Popen(cmd, stdoutsubprocess.PIPE, stderrsubprocess.PIPE) self.servers.append((sqlite, proc, 8002)) time.sleep(2) print(fSQLite MCP 服務(wù)端已啟動 (PID: {proc.pid})) return proc def stop_all(self): 停止所有 MCP 服務(wù)端 for name, proc, _ in self.servers: print(f正在停止 {name} 服務(wù)端...) proc.terminate() proc.wait() self.servers.clear() # 使用上下文管理器確保資源清理 if __name__ __main__: manager MCPServerManager() try: manager.start_file_server() manager.start_sqlite_server() print(所有 MCP 服務(wù)端已就緒按 CtrlC 停止...) while True: time.sleep(1) except KeyboardInterrupt: manager.stop_all()3.2 實(shí)現(xiàn) MCP 客戶端并轉(zhuǎn)換為 LangChain Tools這是最關(guān)鍵的一步我們需要編寫一個 MCP 客戶端連接到服務(wù)端獲取其提供的工具列表并將每個工具包裝成 LangChain 能識別的Tool對象。# mcp_client.py import asyncio from typing import List, Optional from langchain.tools import BaseTool from langchain_core.tools import Tool from mcp import ClientSession, StdioServerParameters from mcp.client import stdio class MCPToolFetcher: def __init__(self, server_name: str, server_params: StdioServerParameters): self.server_name server_name self.server_params server_params self.tools: List[Tool] [] async def connect_and_fetch_tools(self): 連接到 MCP 服務(wù)端并獲取工具列表 # 創(chuàng)建與 MCP 服務(wù)端的會話 async with stdio.stdio_client(self.server_params) as (read, write): async with ClientSession(read, write) as session: # 初始化連接 await session.initialize() # 列出服務(wù)端提供的所有工具 response await session.list_tools() mcp_tools response.tools for mcp_tool in mcp_tools: # 為每個 MCP 工具創(chuàng)建一個 LangChain Tool 包裝器 langchain_tool Tool( namef{self.server_name}_{mcp_tool.name}, funcself._create_tool_func(session, mcp_tool), descriptionmcp_tool.description, ) self.tools.append(langchain_tool) return self.tools def _create_tool_func(self, session, mcp_tool): 創(chuàng)建一個能調(diào)用特定 MCP 工具的同步函數(shù) # 注意LangChain Tool 的 func 是同步的但 MCP 調(diào)用是異步的。 # 我們需要在同步函數(shù)中運(yùn)行異步代碼。這里使用 asyncio.run 簡化處理 # 在生產(chǎn)環(huán)境中需要考慮更優(yōu)的異步集成方式。 async def async_tool_func(**kwargs): # 調(diào)用 MCP 工具的 execute 方法 result await session.call_tool(mcp_tool.name, argumentskwargs) # 返回工具執(zhí)行結(jié)果的文本內(nèi)容 return \n.join([c.text for c in result.content if c.type text]) def sync_wrapper(**kwargs): # 在新的事件循環(huán)中運(yùn)行異步函數(shù)適用于簡單腳本 # 注意在已有事件循環(huán)的環(huán)境中如 FastAPI需要使用其他方式 return asyncio.run(async_tool_func(**kwargs)) return sync_wrapper # 工具獲取工具函數(shù) def get_all_mcp_tools() - List[Tool]: 獲取所有已配置 MCP 服務(wù)端的工具 all_tools [] # 定義服務(wù)端連接參數(shù) servers [ (filesystem, StdioServerParameters(commandmcp-server-filesystem, args[--root, ./data])), (sqlite, StdioServerParameters(commandmcp-server-sqlite, args[--db, ./data/example.db])), ] async def fetch_all(): for name, params in servers: fetcher MCPToolFetcher(name, params) tools await fetcher.connect_and_fetch_tools() all_tools.extend(tools) # 運(yùn)行異步函數(shù)獲取所有工具 asyncio.run(fetch_all()) return all_tools3.3 創(chuàng)建 LangChain Agent 并集成動態(tài)工具現(xiàn)在我們可以使用獲取到的動態(tài)工具列表來初始化一個 LangChain Agent。這里使用 ReAct 代理類型它適合多步驟的工具調(diào)用場景。# agent_builder.py from langchain.agents import AgentExecutor, create_react_agent from langchain_core.prompts import PromptTemplate from langchain_anthropic import ChatAnthropic from config import ANTHROPIC_API_KEY from mcp_client import get_all_mcp_tools def build_mcp_agent(): # 1. 初始化 Claude 模型 llm ChatAnthropic( modelclaude-3-haiku-20240307, # 可根據(jù)需要選擇 sonnet, opus 等型號 temperature0, api_keyANTHROPIC_API_KEY ) # 2. 動態(tài)獲取所有 MCP 工具 print(正在從 MCP 服務(wù)端發(fā)現(xiàn)工具...) tools get_all_mcp_tools() print(f已發(fā)現(xiàn) {len(tools)} 個工具: {[t.name for t in tools]}) # 3. 定義 ReAct 代理的提示詞模板 # 提示詞需要指導(dǎo)模型如何思考和使用工具 prompt PromptTemplate.from_template( 你是一個有幫助的AI助手可以訪問以下工具 {tools} 請使用以下格式回答 問題用戶提出的問題 思考你需要思考如何一步步解決問題。你可以使用工具也可以直接回答。 行動要使用的工具名稱必須是以下工具之一[{tool_names}] 行動輸入工具的輸入必須是一個有效的JSON字符串 觀察工具返回的結(jié)果 ... (這個 思考/行動/行動輸入/觀察 循環(huán)可以重復(fù)多次) 思考我現(xiàn)在知道了最終答案 最終答案對原始問題的最終回答 開始 問題{input} 思考{agent_scratchpad} ) # 4. 創(chuàng)建 ReAct 代理 agent create_react_agent(llm, tools, prompt) # 5. 創(chuàng)建代理執(zhí)行器控制最大迭代次數(shù)以避免無限循環(huán) agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, # 開啟詳細(xì)日志便于調(diào)試 handle_parsing_errorsTrue, # 處理模型輸出解析錯誤 max_iterations10, # 限制最大思考步驟 early_stopping_methodgenerate, # 當(dāng)模型決定不再使用工具時停止 ) return agent_executor4. 運(yùn)行驗(yàn)證與結(jié)果分析構(gòu)建好 Agent 后我們需要準(zhǔn)備測試數(shù)據(jù)運(yùn)行幾個典型任務(wù)來驗(yàn)證其能力。4.1 準(zhǔn)備測試環(huán)境與數(shù)據(jù)首先創(chuàng)建必要的目錄和測試數(shù)據(jù)。# 創(chuàng)建數(shù)據(jù)目錄和示例文件 mkdir -p ./data echo 項(xiàng)目報告草案\n主要內(nèi)容...\n待辦整理圖表 ./data/report.txt echo 會議記錄\n日期2024-05-27\n議題Agent架構(gòu)評審 ./data/meeting.txt # 創(chuàng)建并初始化一個 SQLite 示例數(shù)據(jù)庫 sqlite3 ./data/example.db EOF CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, email TEXT, department TEXT); INSERT INTO users (name, email, department) VALUES (張三, zhangsanexample.com, 工程部), (李四, lisiexample.com, 產(chǎn)品部), (王五, wangwuexample.com, 市場部); EOF4.2 執(zhí)行綜合任務(wù)測試現(xiàn)在編寫一個測試腳本讓 Agent 執(zhí)行一個結(jié)合了文件操作和數(shù)據(jù)庫查詢的復(fù)雜任務(wù)。# run_agent.py from agent_builder import build_mcp_agent import asyncio async def main(): # 構(gòu)建 Agent agent build_mcp_agent() # 測試任務(wù) 1簡單的文件讀取 print(\n 測試任務(wù) 1讀取文件 ) result1 await agent.ainvoke({input: 請讀取 data 目錄下 report.txt 文件的內(nèi)容并總結(jié)其要點(diǎn)。}) print(f任務(wù)1結(jié)果: {result1[output]}) # 測試任務(wù) 2數(shù)據(jù)庫查詢 print(\n 測試任務(wù) 2查詢數(shù)據(jù)庫 ) result2 await agent.ainvoke({input: 查詢 example.db 數(shù)據(jù)庫中 users 表里所有在工程部的員工姓名和郵箱。}) print(f任務(wù)2結(jié)果: {result2[output]}) # 測試任務(wù) 3多步驟組合任務(wù) print(\n 測試任務(wù) 3組合任務(wù) ) result3 await agent.ainvoke({ input: 1. 首先請列出 data 目錄下所有的 .txt 文件。 2. 然后讀取 meeting.txt 文件提取會議日期。 3. 最后去 example.db 數(shù)據(jù)庫的 users 表里查一下產(chǎn)品部有哪些人把結(jié)果和會議日期一起整理成一個簡短的摘要。 }) print(f任務(wù)3結(jié)果: {result3[output]}) if __name__ __main__: asyncio.run(main())運(yùn)行此腳本你應(yīng)該能看到類似以下的輸出具體內(nèi)容因模型隨機(jī)性略有不同正在從 MCP 服務(wù)端發(fā)現(xiàn)工具... 已發(fā)現(xiàn) 4 個工具: [filesystem_read_file, filesystem_list_directory, sqlite_execute_query, sqlite_list_tables] 測試任務(wù) 1讀取文件 進(jìn)入新的 Agent 執(zhí)行鏈... 思考用戶要求讀取 report.txt 文件并總結(jié)要點(diǎn)。我需要使用文件讀取工具。 行動filesystem_read_file 行動輸入{path: ./data/report.txt} 觀察項(xiàng)目報告草案\n主要內(nèi)容...\n待辦整理圖表 思考我已讀取文件內(nèi)容?，F(xiàn)在需要總結(jié)要點(diǎn)。內(nèi)容顯示這是一個項(xiàng)目報告草案包含主要內(nèi)容和待辦事項(xiàng)整理圖表。我可以直接總結(jié)。 最終答案該文件是一個項(xiàng)目報告草案主要內(nèi)容已列出當(dāng)前待辦事項(xiàng)是整理圖表。 任務(wù)1結(jié)果該文件是一個項(xiàng)目報告草案主要內(nèi)容已列出當(dāng)前待辦事項(xiàng)是整理圖表。 測試任務(wù) 2查詢數(shù)據(jù)庫 進(jìn)入新的 Agent 執(zhí)行鏈... 思考用戶需要查詢工程部員工。我需要使用 SQLite 查詢工具。 行動sqlite_execute_query 行動輸入{query: SELECT name, email FROM users WHERE department 工程部} 觀察[{name: 張三, email: zhangsanexample.com}] 思考查詢返回了結(jié)果。我可以直接給出答案。 最終答案工程部的員工是張三郵箱是 zhangsanexample.com。 任務(wù)2結(jié)果工程部的員工是張三郵箱是 zhangsanexample.com。從輸出中我們可以看到 Agent 成功完成了以下工作動態(tài)工具發(fā)現(xiàn)啟動時從兩個 MCP 服務(wù)端獲取了 4 個工具。正確工具選擇針對“讀文件”任務(wù)選擇了filesystem_read_file針對“查數(shù)據(jù)庫”任務(wù)選擇了sqlite_execute_query。參數(shù)構(gòu)造能夠根據(jù)任務(wù)描述正確構(gòu)造工具所需的輸入?yún)?shù)如文件路徑、SQL 語句。結(jié)果解析與總結(jié)能夠理解工具返回的原始數(shù)據(jù)文本行、JSON 數(shù)組并將其組織成自然語言回答。4.3 關(guān)鍵配置參數(shù)解析在構(gòu)建 Agent 時有幾個關(guān)鍵參數(shù)直接影響其行為和性能參數(shù)所在位置含義與影響推薦值/建議max_iterationsAgentExecutor代理最大推理-執(zhí)行循環(huán)次數(shù)。防止任務(wù)過于復(fù)雜導(dǎo)致無限循環(huán)。簡單任務(wù) 5-10復(fù)雜任務(wù) 15-20。需結(jié)合max_tokens考慮。handle_parsing_errorsAgentExecutor是否處理模型輸出格式解析錯誤。開啟后解析失敗會嘗試讓模型重試。建議始終設(shè)為True提高魯棒性。verboseAgentExecutor是否打印詳細(xì)的思考鏈Chain of Thought日志。開發(fā)調(diào)試時設(shè)為True生產(chǎn)環(huán)境設(shè)為False。temperatureChatAnthropic模型生成文本的隨機(jī)性。值越高輸出越多樣、越有創(chuàng)造性。Agent 工具調(diào)用場景建議設(shè)為0或0.1以保證工具選擇和參數(shù)生成的穩(wěn)定性。modelChatAnthropic使用的 Claude 模型版本。claude-3-haiku速度最快成本最低claude-3-sonnet平衡claude-3-opus能力最強(qiáng)但最慢最貴。根據(jù)任務(wù)復(fù)雜度選擇。5. 常見問題排查與性能優(yōu)化將 LangChain Agent 與 MCP、Skills 集成到生產(chǎn)環(huán)境時會遇到一系列工程化挑戰(zhàn)。以下是典型問題的排查路徑和優(yōu)化建議。5.1 連接與工具發(fā)現(xiàn)失敗現(xiàn)象Agent 啟動時報錯無法連接到 MCP 服務(wù)端或工具列表為空。可能原因檢查方式解決方案MCP 服務(wù)端未啟動檢查對應(yīng)端口如 8001, 8002是否在監(jiān)聽 (netstat -an | grep 8001)。查看MCPServerManager日志是否有啟動錯誤。確保啟動命令正確依賴已安裝。檢查服務(wù)端二進(jìn)制文件路徑。命令或參數(shù)錯誤檢查StdioServerParameters中的command和args是否與服務(wù)端程序匹配。使用絕對路徑指定命令或確保命令在系統(tǒng) PATH 中。參考服務(wù)端文檔確認(rèn)參數(shù)格式。權(quán)限問題檢查服務(wù)端是否有權(quán)限訪問指定目錄如./data或數(shù)據(jù)庫文件。調(diào)整目錄權(quán)限或使用服務(wù)端可訪問的路徑。協(xié)議版本不兼容查看 MCP 客戶端和服務(wù)端的版本。檢查初始化握手階段的錯誤信息。確?？蛻舳撕头?wù)端使用的mcpSDK 版本兼容?？蓢L試升級到最新穩(wěn)定版。5.2 工具調(diào)用錯誤或超時現(xiàn)象Agent 選擇了正確的工具但調(diào)用失敗或長時間無響應(yīng)。問題現(xiàn)象常見原因檢查方式處理建議工具參數(shù)格式錯誤模型生成的 JSON 參數(shù)不符合工具要求。查看verbose日志中的“行動輸入”字段。手動用相同參數(shù)測試工具。在提示詞中更清晰地描述工具所需的參數(shù)格式。使用 Pydantic 模型對工具輸入進(jìn)行校驗(yàn)。工具執(zhí)行內(nèi)部錯誤MCP 服務(wù)端在處理請求時崩潰或返回錯誤。查看 MCP 服務(wù)端進(jìn)程的標(biāo)準(zhǔn)錯誤輸出。檢查服務(wù)端日志。確保輸入數(shù)據(jù)如 SQL 語法、文件路徑有效。在工具包裝函數(shù)中添加更詳細(xì)的錯誤捕獲和日志。網(wǎng)絡(luò)或進(jìn)程通信超時服務(wù)端響應(yīng)慢或進(jìn)程僵死。在工具調(diào)用代碼處添加超時設(shè)置。監(jiān)控服務(wù)端資源占用CPU/內(nèi)存。為異步調(diào)用設(shè)置asyncio.wait_for超時。優(yōu)化服務(wù)端性能或?qū)臅r工具單獨(dú)設(shè)置更長的超時。異步上下文沖突在已有事件循環(huán)如 FastAPI中同步調(diào)用工具導(dǎo)致錯誤。觀察是否報錯RuntimeError: This event loop is already running。避免在同步函數(shù)中直接使用asyncio.run。改用asyncio.create_task或在主異步上下文中調(diào)用工具。重構(gòu)代碼使整個 Agent 調(diào)用鏈保持異步。5.3 Agent 邏輯錯誤與優(yōu)化現(xiàn)象Agent 陷入循環(huán)、選擇錯誤工具、或生成無關(guān)內(nèi)容。問題根因分析優(yōu)化策略工具選擇不準(zhǔn)1. 工具描述 (description) 不夠清晰。2. 提示詞未充分指導(dǎo)模型如何選擇工具。3. 工具過多模型混淆。1.優(yōu)化工具描述用自然語言清晰說明工具功能、輸入輸出示例。例如將sqlite_execute_query描述改為“執(zhí)行一條 SQL SELECT 查詢語句并返回結(jié)果集。輸入應(yīng)為包含query鍵的 JSON 對象?！?.改進(jìn)提示詞在PromptTemplate中加入工具選擇范例。3.工具分組/路由對工具進(jìn)行分類先讓 Agent 選擇大類再選擇具體工具。無效迭代過多模型在已經(jīng)獲得答案的情況下仍繼續(xù)嘗試使用工具。1.調(diào)整max_iterations根據(jù)任務(wù)復(fù)雜度設(shè)置合理上限。2.使用更好的停止條件AgentExecutor的early_stopping_method設(shè)為generate讓模型自己決定何時停止。3.優(yōu)化思考鏈在提示詞中強(qiáng)調(diào)“當(dāng)你認(rèn)為已有足夠信息回答問題時可以直接給出最終答案”。處理復(fù)雜任務(wù)能力弱單一 ReAct 代理難以規(guī)劃冗長或多分支任務(wù)。1.升級 Agent 類型使用Plan-and-Execute或OpenAI Functions代理它們更擅長規(guī)劃。2.引入 LangGraph對于有狀態(tài)、多分支的工作流使用 LangGraph 來顯式定義狀態(tài)圖和節(jié)點(diǎn)邏輯。3.任務(wù)分解在上層設(shè)計(jì)一個“主控”Agent負(fù)責(zé)將復(fù)雜任務(wù)拆解為子任務(wù)再分發(fā)給負(fù)責(zé)具體工具的“子”Agent。5.4 生產(chǎn)環(huán)境部署建議在開發(fā)環(huán)境跑通后部署到生產(chǎn)環(huán)境還需考慮以下方面MCP 服務(wù)端部署不應(yīng)以簡單的子進(jìn)程方式運(yùn)行。建議將每個 MCP 服務(wù)端部署為獨(dú)立的容器Docker或系統(tǒng)服務(wù)systemd并配置健康檢查、資源限制和自動重啟。連接管理與池化頻繁創(chuàng)建銷毀到 MCP 服務(wù)端的連接開銷大。應(yīng)實(shí)現(xiàn)連接池讓多個 Agent 實(shí)例共享到同一服務(wù)端的穩(wěn)定連接。安全性工具權(quán)限嚴(yán)格限制每個 MCP 服務(wù)端的權(quán)限。例如文件系統(tǒng)服務(wù)端只允許訪問特定沙箱目錄數(shù)據(jù)庫服務(wù)端使用只讀或最小權(quán)限賬戶。輸入驗(yàn)證與清理對所有從模型傳遞給工具的參數(shù)進(jìn)行嚴(yán)格的驗(yàn)證和清理防止 SQL 注入、路徑遍歷等攻擊。API 密鑰管理使用安全的秘密管理服務(wù)如 Vault, AWS Secrets Manager存儲和輪換 API 密鑰切勿硬編碼在代碼或配置文件中?？捎^測性結(jié)構(gòu)化日志記錄每個工具調(diào)用的詳細(xì)信息工具名、輸入、輸出、耗時、狀態(tài)。鏈路追蹤為每個用戶會話分配唯一 ID并貫穿所有的 Agent 思考、工具調(diào)用步驟便于問題排查。監(jiān)控指標(biāo)監(jiān)控 Agent 的請求量、響應(yīng)時間、工具調(diào)用成功率、迭代次數(shù)分布等。性能與成本緩存對頻繁且結(jié)果不變的工具調(diào)用如讀取靜態(tài)配置添加緩存層。模型選擇根據(jù)任務(wù)類型選擇合適的模型。簡單的工具調(diào)用可用Haiku復(fù)雜規(guī)劃可用Sonnet或Opus。限制 Token 消耗設(shè)置max_tokens上限防止因異常導(dǎo)致生成過長內(nèi)容而產(chǎn)生高費(fèi)用。6. 擴(kuò)展方向構(gòu)建自定義 Skills 與高級工作流掌握了基礎(chǔ)集成后你可以向兩個方向深入一是創(chuàng)建自己的 Skills二是構(gòu)建更復(fù)雜的智能工作流。6.1 開發(fā)自定義 MCP 服務(wù)端與 Skill一個 Skill 本質(zhì)上是一個或多個相關(guān)工具的集合并可能附帶一些預(yù)設(shè)提示詞。開發(fā)自定義 Skill 的步驟如下定義工具接口明確 Skill 要提供哪些功能。實(shí)現(xiàn) MCP 服務(wù)端使用mcpSDK 實(shí)現(xiàn)這些功能并遵循 MCP 協(xié)議暴露它們。打包與分發(fā)將服務(wù)端代碼和配置打包如 Docker 鏡像、Python 包方便部署和共享。以下是一個簡單的“天氣查詢” Skill 的服務(wù)端示例# weather_mcp_server.py from mcp.server import Server, NotificationOptions from mcp.server.models import InitializationOptions import mcp.server.stdio import asyncio from some_weather_lib import get_weather # 假設(shè)有一個獲取天氣的庫 async def main(): # 創(chuàng)建 MCP 服務(wù)端 server Server(weather-skills) server.list_tools() async def handle_list_tools(): # 公布此服務(wù)端提供的工具 return [ { name: get_current_weather, description: 獲取指定城市的當(dāng)前天氣情況。, inputSchema: { type: object, properties: { city: {type: string, description: 城市名稱例如北京} }, required: [city] } } ] server.call_tool() async def handle_call_tool(name: str, arguments: dict): if name get_current_weather: city arguments.get(city) if not city: raise ValueError(缺少參數(shù) city) # 調(diào)用實(shí)際的外部天氣 API weather_info await get_weather(city) return [ { type: text, text: f{city}的天氣{weather_info[condition]}溫度{weather_info[temp]}°C。 } ] else: raise ValueError(f未知工具: {name}) # 通過標(biāo)準(zhǔn)輸入輸出運(yùn)行服務(wù)端 async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions( server_nameweather-skills, server_version0.1.0 ) ) if __name__ __main__: asyncio.run(main())6.2 使用 LangGraph 編排復(fù)雜 Agent 工作流當(dāng)任務(wù)涉及多個 Agent 協(xié)作、狀態(tài)持久化或條件分支時LangChain 的基礎(chǔ) AgentExecutor 會顯得力不從心。此時LangGraph 是更強(qiáng)大的選擇。它允許你將工作流定義為圖Graph其中節(jié)點(diǎn)可以是 Agent、工具或任何函數(shù)邊定義了執(zhí)行流程。例如一個“數(shù)據(jù)報告生成”工作流可能包含以下節(jié)點(diǎn)規(guī)劃節(jié)點(diǎn)分析用戶請求拆解為“取數(shù)據(jù)”、“分析”、“生成圖表”、“撰寫報告”等子任務(wù)。數(shù)據(jù)查詢節(jié)點(diǎn)調(diào)用 SQL MCP 工具獲取數(shù)據(jù)。分析節(jié)點(diǎn)調(diào)用 Python 計(jì)算工具或另一個 LLM 進(jìn)行數(shù)據(jù)分析。圖表生成節(jié)點(diǎn)調(diào)用圖表生成 API。報告匯編節(jié)點(diǎn)將數(shù)據(jù)、分析結(jié)果、圖表整合成最終報告。使用 LangGraph你可以清晰地定義這些節(jié)點(diǎn)的執(zhí)行順序和條件分支例如如果數(shù)據(jù)為空則跳過分析節(jié)點(diǎn)并持久化整個工作流的狀態(tài)實(shí)現(xiàn)更穩(wěn)健和可調(diào)試的復(fù)雜 Agent 系統(tǒng)。通過將 LangChain Agent、MCP 協(xié)議和 Skills 模塊相結(jié)合我們構(gòu)建的智能體不再是一個封閉、僵化的系統(tǒng)而是一個能夠動態(tài)擴(kuò)展、靈活組合的開放平臺。這種架構(gòu)使得集成新工具、復(fù)用已有能力、以及構(gòu)建復(fù)雜工作流變得前所未有的高效。從簡單的文件查詢到結(jié)合數(shù)據(jù)庫、API 和自定義邏輯的復(fù)雜任務(wù)Agent 都能通過統(tǒng)一的協(xié)議層進(jìn)行調(diào)度和執(zhí)行。在向生產(chǎn)環(huán)境邁進(jìn)時務(wù)必關(guān)注安全性、可靠性和可觀測性通過連接池、權(quán)限控制、結(jié)構(gòu)化日志和監(jiān)控指標(biāo)來保障系統(tǒng)的穩(wěn)定運(yùn)行。下一步你可以嘗試開發(fā)自己的專屬 Skills或者利用 LangGraph 來設(shè)計(jì)更精巧的多智能體協(xié)作流程從而解鎖 AI 應(yīng)用開發(fā)的更大潛力。