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