構建強大AI Agent的關鍵 = Pydantic AI + MCP + Advanced Web Scraping
引言
在文中，我將展示一個超快速教程，教你如何使用 Pydantic AI、MCP 和高級網頁抓取技術，構建一個強大的智能聊天機器人，適用于商業或個人用途。。
MCP 逐漸被接受，因為它是一個開放標準。我制作了一些非常酷的視頻，你一定會喜歡。在 AI 項目開發中，AI 模型的集成非常復雜，現有框架如 LangGraph 和 Pydantic AI Tools 存在一些問題。LangGraph 和 Pydantic AI 的代碼抽象度高，且過于商業化。
MCP 的優勢在于，開放標準便于服務提供商開發 API，避免開發者重復造輪子，并允許他們使用現有的 MCP 服務來增強智能代理的功能。
最近，DeepSeek 發布了一個新的開源模型 DeepSeek-V3，這是一個擁有 6710 億參數的專家混合模型（MoE），其中每個 token 激活 370 億參數。訓練僅耗時兩個月，成本不到 600 萬美元，足以與 ChatGPT 競爭，并震撼全球 AI 社區。
DeepSeek-V3 的完整訓練僅需 278.8 萬 GPU 小時。假設 H800 GPU 的租賃價格為每 GPU 小時 2 美元，DeepSeek-V3 的總訓練成本僅為 557.6 萬美元。
Pydantic AI 是一個我在多個視頻中介紹過的框架，因此這里不再詳細描述。但其應用非常易于理解。
接下來，我將通過一個實時聊天機器人演示，向你展示我的意思。
演示
我將提出兩個問題：“搜索有關人工智能的最新新聞”和“創建一個柱狀圖，展示全球五大城市的人口”。你也可以提出任何你想問的問題。
聊天機器人使用模型控制協議（MCP），其中不同工具協同工作以完成任務。Python 工具由 FastMCP 驅動，可以安全地運行 Python 代碼，并使用 PythonREPL 類創建可視化內容。它包含兩個工具：python_repl(code: str) 用于運行代碼，data_visualization(code: str) 用于生成 Matplotlib 圖表。
網頁搜索工具使用 Exa API 獲取搜索結果，配置 FastMCP 作為 API 客戶端，默認返回五個搜索結果，并以 Markdown 格式返回答案。
AI 代理連接到兩個 MCP 服務器，必要時進行重試，并執行多步驟任務，如使用 exa_search.py 查找氣候變化數據或使用 python_tools.py 創建柱狀圖。
免責聲明：本文僅用于研究為目的。我們不鼓勵任何人抓取網站，特別是那些可能有條款和條件禁止此類行為的網頁。
DeepSeek V3 是如何訓練的？
DeepSeek 能夠在兩年內以僅 557 萬美元的成本開發出高性能 AI 模型，這得益于以下創新技術：
精確激活部分“腦細胞”
DeepSeek-V3 采用了“混合專家架構”設計。簡單來說，它只在需要時激活部分“腦細胞”，而不是全部，從而大大降低了計算資源消耗。訓練模型僅使用了 2048 個 NVIDIA H800 GPU。
數據處理與節能創新
DeepSeek 開發了內部工具來生成高質量訓練數據，并使用“蒸餾技術”進一步壓縮計算資源。訓練過程中使用了 FP8 技術，這種低精度數據格式可以顯著降低顯存需求，同時提高效率。FP8 技術將顯存需求降至傳統 FP16 技術的一半，同時保持相同的計算性能。
圖片
開始編碼
我們通過以下方式初始化代碼：
?MCPServerStdio：以子進程方式運行服務器，并通過 stdio 傳輸進行連接。?FastMCP：使構建 MCP 服務器變得簡單直觀。創建工具、公開資源并使用簡潔的 Python 代碼定義提示。?exa_py：原名 Metaphor，是一個基于大型語言模型的搜索引擎 ??，允許用戶使用完整句子和自然語言進行搜索，還能模擬人們在網絡上分享和討論鏈接的方式來查詢內容。
app.py
import os
import asyncio
from pydantic_ai importAgent
from pydantic_ai.mcp importMCPServerStdioexa_search.py
from mcp.server.fastmcp importFastMCP
from dotenv import load_dotenv
import os
from exa_py importExapython_tools.py
from mcp.server.fastmcp importFastMCP
import io
import base64
import matplotlib.pyplot as plt
import sys
from io importStringIO
import traceback我們定義了一個由 FastMCP 驅動的 Python 工具，用于執行任意 Python 代碼并使用 Matplotlib 生成數據可視化。它包括一個 PythonREPL 類，在受限環境中運行 Python 代碼，重定向標準輸出 sys.stdout 以捕獲執行結果，并返回輸出或錯誤堆棧信息。
它提供了兩個 MCP 工具：python_repl(code: str) 執行 Python 代碼并返回其輸出；data_visualization(code: str) 運行 Python 代碼，保存 Matplotlib 圖表，并將圖像返回為 base64 編碼字符串，處理可能出現的錯誤。
使用 FastMCP("python_tools") 定義了一個用于執行 Python 任務的多代理工具，并在直接執行時作為 MCP 服務器運行。
from mcp.server.fastmcp importFastMCP
import io
import base64
import matplotlib.pyplot as plt
import sys
from io importStringIO
import traceback
mcp =FastMCP("python_tools")
classPythonREPL:
def run(self, code):
old_stdout = sys.stdout
redirected_output = sys.stdout =StringIO()
try:
exec(code, globals())
sys.stdout = old_stdout
return redirected_output.getvalue()
exceptExceptionas e:
sys.stdout = old_stdout
return f"錯誤：{str(e)}\n{traceback.format_exc()}"
repl =PythonREPL()
@mcp.tool()
async def python_repl(code: str)-> str:
"""執行 Python 代碼。"""
return repl.run(code)
@mcp.tool()
async def data_visualization(code: str)-> str:
"""執行 Python 代碼，使用 Matplotlib 進行可視化。"""
try:
repl.run(code)
buf = io.BytesIO()
plt.savefig(buf, format='png')
buf.seek(0)
img_str = base64.b64encode(buf.getvalue()).decode()
plt.close()# 關閉圖像以釋放內存
return f"data:image/png;base64,{img_str}"
exceptExceptionas e:
return f"創建圖表時出錯：{str(e)}"
if __name__ =="__main__":
mcp.run()然后，我們使用 Exa API 在 MCP（模型控制協議）框架內設置了一個網頁搜索工具。首先加載環境變量并初始化 FastMCP 及其元數據。
接著，使用環境變量中的 API 密鑰（或默認密鑰）配置 Exa API 客戶端。定義了默認搜索配置，每次搜索默認返回五個結果，除非另有指定。主要函數注冊為 MCP 工具，異步查詢 Exa API，并以 Markdown 格式返回結果。
format_search_results() 函數結構化輸出結果，包括標題、URL、發布日期和摘要（如果可用）。如果直接執行腳本，則運行 MCP 實例。
from mcp.server.fastmcp importFastMCP
from dotenv import load_dotenv
import os
from exa_py importExa
load_dotenv(override=True)
# 初始化 FastMCP
mcp =FastMCP(
name="websearch",
versinotallow="1.0.0",
descriptinotallow="使用 Exa API 的網頁搜索功能"
)
# 初始化 Exa 客戶端
exa_api_key = os.getenv("EXA_API_KEY","")
exa =Exa(api_key=exa_api_key)
# 默認搜索配置
websearch_config ={
"parameters":{
"default_num_results":5,
"include_domains":[]
}
}
@mcp.tool()
async def search_web(query: str, num_results:int=None)-> str:
"""使用 Exa API 搜索網頁并以 Markdown 格式返回結果。"""
try:
search_args ={
"num_results": num_results or websearch_config["parameters"]["default_num_results"]
}
search_results = exa.search_and_contents(
query,
summary={"query":"主要內容和關鍵要點"},
**search_args
)
return format_search_results(search_results)
exceptExceptionas e:
return f"使用 Exa 搜索時發生錯誤：{e}"
def format_search_results(search_results):
ifnot search_results.results:
return"未找到結果。"
markdown_results ="### 搜索結果：\n\n"
for idx, result in enumerate(search_results.results,1):
title = result.title if hasattr(result,'title')and result.title else"無標題"
url = result.url
published_date = f"（發布日期：{result.published_date}）"if hasattr(result,'published_date')and result.published_date else""
markdown_results += f"**{idx}.** [{title}]({url}){published_date}\n"
if hasattr(result,'summary')and result.summary:
markdown_results += f"> **摘要：** {result.summary}\n\n"
else:
markdown_results +="\n"
return markdown_results
if __name__ =="__main__":
mcp.run()我開發了一個 AI 代理，利用 DeepSeek Chat 模型處理請求，同時集成了兩個 MCP（模型控制協議）服務器：
?exa_search.py：使用 Exa API 處理網頁搜索。?python_tools.py：提供基于 Python 的工具，包括數據分析和可視化。
代理通過 agent.run_mcp_servers 定義，連接到這些服務器，并在必要時重試最多三次。主要函數運行 MCP 服務器并執行多步驟任務：
1.搜索近期氣候變化統計數據。2.使用 Python 生成一個柱狀圖，展示過去十年全球氣溫上升情況。
import os
import asyncio
from pydantic_ai importAgent
from pydantic_ai.mcp importMCPServerStdio
from pydantic_ai.models.openai importOpenAIMode
deepseek_chat_model =OpenAIModel(# 定義基礎為 Open AI
'deepseek-chat',
base_url='https://api.deepseek.com',
api_key=os.environ["DEEPSEEK_API_KEY"],
)
# 定義 MCP 服務器
exa_server =MCPServerStdio(
'python',
['exa_search.py']
)
python_tools_server =MCPServerStdio(
'python',
['python_tools.py']
)
# 定義包含兩個 MCP 服務器的代理
agent =Agent(
deepseek_chat_model,
mcp_servers=[exa_server, python_tools_server],
retries=3
)
# 主異步函數
async def main():
async with agent.run_mcp_servers():
result = await agent.run("""
我需要分析一些氣候數據。首先，搜索近期的氣候變化統計數據。
然后，創建一個柱狀圖，展示過去十年全球氣溫的上升情況。
使用Python進行數據可視化。
""")
print(result)
# 運行異步函數
if __name__ =="__main__":
asyncio.run(main())結論
MCP 不僅僅是一個協議或架構模式，它代表了設計 AI 驅動應用和分布式系統的根本轉變。通過有效分離模型、上下文和協議，組織可以創建適應性強、可擴展且易于維護的解決方案。
DeepSeek-V3 體現了這種理念，證明了算法優化和工程創新即使在資源有限的情況下也能生產出頂尖的 AI 模型。
這一突破不僅挑戰了傳統的 AI 商業模式，還為企業家和開發者打開了大門，讓他們能夠利用高性能 AI 作為包容性創新的工具。
