构建强大AI Agent的关键 = Pydantic AI + MCP + Advanced Web Scraping

引言在文中，我将展示一个超快速教程，教你如何使用 Pydantic AI、MCP 和高级网页抓取技术，构建一个强大的智能聊天机器人，适用于商业或个人用途。 MCP 逐渐被接受，因为它是一个开放标准。我制作了一些非常酷的视频，你一定会喜欢。

引言

在文中，我将展示一个超快速教程，教你如何使用 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 importMCPServerStdio

exa_search.py

复制

from mcp.server.fastmcp importFastMCP
from dotenv import load_dotenv
import os
from exa_py importExa

python_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 作为包容性创新的工具。

构建强大AI Agent的关键 = Pydantic AI + MCP + Advanced Web Scraping

引言

演示

DeepSeek V3 是如何训练的？

精确激活部分“脑细胞”

数据处理与节能创新

开始编码

app.py

exa_search.py

python_tools.py

结论

相关资讯

一文带你入门 MCP（模型上下文协议）

Manus平替方案：用DeepSeek+MCP Server构建AI自主工作流

mcp-agent发布:轻量级框架助力智能体应用高效构建