Transcription
今天我们将进行一个关于 Pyantic AI 的速成课程，这是另一个类似 Langchain 和 Llama Index 的 Python 代理框架。我频道上已有关于这两个框架的教程，并且一如既往，这些代理框架略有不同。大部分情况下它们非常相似，但它们确实有不同的侧重点。
Langchain 是一个更通用的端到端框架，你可以说它有点像 Django。它包含了许多内置功能，并且有自己的观点。而 Llama Index 更侧重于检索，今天我们要讲的 Pyantic AI 则更侧重于验证、类型安全、结构化输出，总的来说，它更简洁、更灵活，有点像 FastAPI。
今天我们会讲很多内容。所以，如果你喜欢这个视频，请通过点赞和订阅让我知道。但现在，让我们开始吧。[音乐]>> 好的。所以，Pyantic AI 就是这个速成课程的主题。正如我在介绍中已经提到的，它是那些生成式 AI 代理框架之一，与 Langchain 和 Llama Index 类似。你也可以看到他们在这里说，这是 Pyantic 的方式。
所以，对于那些不知道 Pyantic 是什么的人来说，它基本上是一个 Python 包，用于在许多不同的上下文中进行验证。其中一个就是 FastAPI，但许多其他代理框架也使用了 Pyantic。Langchain 使用它，Llama Index 也使用它。这也是他们在这里提出的论点之一。当你向下看“为什么使用 Pyantic AI”时，它基本上说“为什么使用衍生物，而你可以直接去源头”。
但更严肃地说，使用 Pyantic AI 的实际原因在一定程度上是选择问题。所以有些人就是喜欢 FastAPI 风格来构建这些代理应用程序。社区中的许多人认为，在代码编写方式和使用它的方式上，Pyantic AI 比 Langchain 更简洁。嗯，它也有不同的侧重点。它非常专注于生产就绪、类型安全、验证事物和进行结构化输出。所以这就是这里的重点，类似于 Llama Index 侧重于检索。
我们可以基本上用所有这些框架来做任何事情，但它们都有各自的优势，而我认为这就是 Pyantic AI 的重点或关键卖点。
理论就到这里，让我们开始设置吧。打开一个终端，导航到你想工作的目录。在我的例子中是教程目录，在这里你可以设置一个虚拟环境。如果你愿意，也可以直接使用你的系统安装。
最简单的方法是输入 `pip` 或 `pip3 install` 来安装我们今天要使用的包。我个人总是更喜欢使用 UV。所以我将使用 `uv init` 初始化一个 UV 项目。然后我将使用 `uv add` 而不是 `pip install`。所以，我在 `uv add` 之后说的任何内容，你都可以直接在 `pip install` 之后说，它具有相同的效果，只是在你自己的系统安装上。
我们这里需要的是 `pyantic-ai`。但我们也需要 `python-dotenv`。还有一件事我们需要的是 `jupyter lab`。这就是我们今天将要工作的环境。这是可选的。你也可以只在普通的 Python 文件中工作。我只是在学习东西时更喜欢 Jupyter Notebook，因为我们可以交互式地单独运行单元格。我们不必运行完整的 Python 脚本。
所以，安装这些依赖项，然后我们可以运行 `jupyter lab` 来启动 Jupyter Lab。如果你像我一样使用 UV，你必须运行 `uv run jupyter lab`。这只会打开开发环境。嗯，就像我做 Jupyter Lab 教程一样，我会简要解释一下这个环境。所以在这里我们不处理 Python 文件。我们处理交互式 Python Notebook。听起来比实际情况更花哨。我们只有可以运行的代码单元格。
所以，我可以输入 `a = 10`，然后我可以输入 `print(a)` 在下面。然后我也可以多次 `print(a)`，而不必从头到尾运行所有代码。我可以运行单个单元格，也可以按不同的顺序运行它们。这使得在探索包或数据集或任何你正在做的事情时，交互式地处理事物变得更加容易。
现在，在我们开始编码之前，还需要设置一件事。你需要一个 `.env` 文件，其中包含你想要使用的提供商的 API 密钥。你显然需要使用某种大型语言模型来为你的代理提供智能。所以，你想进入你正在工作的目录。在我的例子中，我现在将从我准备好的代码中复制它。我有一个 YouTube 准备库。我有一个 `pyantic` 目录，里面有一个 `.env` 文件。
我现在不会展示我的 `.env` 文件，因为这个 `.env` 文件包含 API 密钥。但你要做的是创建一个名为 `.env` 的文件。就是一个点符号 `.env`。在这个文件中，你需要有一个 API 密钥。我将在这里输入 `test` 来向你展示它的样子。这只是 OpenAI 的 `API_KEY`，例如，如果是 OpenAI 密钥，否则是 Anthropic 或你正在使用的任何其他密钥，然后是密钥本身。所以你只需要在这里放入密钥，然后像这样保存文件。
基本上就是这样。这是一个 `.env` 文件，它将非常有用，因为当你尝试使用 OpenAI 模型时，它会被 Pyantic 自动识别，并且它将自动与 OpenAI API 进行身份验证。
所以，让我们删除所有这些，然后开始从 `dotenv` 导入 `load_dotenv` 函数。这就是我们安装 `python-dotenv` 的原因，因为这个函数调用允许我们从 `.env` 文件加载 API 密钥。当然，你也可以选择在终端中使用 `export` 语句来指定它们。但这是一种简单的方式。你只需创建一个文件，将密钥放在那里。你也可以在那里放多个密钥，然后 `load_dotenv` 会将其加载到你的脚本或应用程序中。
所以，这就是设置。现在，我总是在探索这些代理框架时喜欢做的是构建一个非常基本的示例。就是发送一条消息给代理，然后就结束了。在 Pyantic 中，我们可以轻松地做到这一点，方法是输入 `from pyantic_ai import Agent`，然后我们将输入 `agent = Agent()`。现在在这里我们将指定模型。
再次，你可以使用多个提供商。这是模型无关的。所以我们可以说，例如，`openai:gpt-4o`。然后我们只需要提供一个系统提示。所以，让我们说 `system_prompt = "You are a helpful assistant."`。为了快速得到答案，我将说“你总是用一个简洁的句子回答”，这样我们就不用等太久才能得到响应。
现在，我们调用它的方式可能与使用其他框架的方式略有不同。它也取决于你运行它的上下文。如果你在 Python 文件中运行它，你也可以使用不同的函数。由于我们在这里使用 Jupyter Notebook，我们需要调用 `run` 函数而不是 `run_sync` 函数。还有一个 `agent.run_sync` 函数，正如你在这里看到的，但我们只使用 `run` 并等待它。
所以，对于那些熟悉异步编程的人来说，有 `async/await`。`await` 基本上迫使我们等待以获得这个异步响应。所以，我将输入 `response = await agent.run("What is Python?")`。然后我可以提供一个消息，比如“Python 是什么？”然后之后，我将打印 `response.output`。
这将发送一个请求并给我一个异常，因为我打错了 `prompt`。所以它应该是 `system_prompt` 像这样。然后“Python 是什么？”“Python 是一种高级的、解释性的编程语言，以其可读性和多功能性而闻名。”很好。
所以，让我向你展示一下如果你不等待会发生什么。不等待会发生什么，你可以说 `response = agent.run("What is Python?")`。然后你可以问一个类似的问题，比如“Python 是什么？”即使我现在没有使用大写字母，但然后我可以立即打印响应而不等待，我得到的结果是一个协程对象。这就是为什么我们需要等待，因为这只是一个承诺，你可以说我们将来会得到答案。我们还没有得到。
在这里，我们等待答案，然后我们将输出打印到命令行。所以，这与我们使用其他框架的方式没有太大区别。
所以，现在让我们在此基础上添加一个基本的结构化输出示例。所以，让我们说我想做什么，为此我将使用 Pyantic 本身，而不是 Pyantic AI，而是核心 Pyantic 包。我将输入 `from pydantic import BaseModel`。所以，这就是我们创建 Pyantic 模型时继承的基类。
让我们说我有一个 `PersonOutput`，或者让我们称之为 `PersonBaseModel`。它将有一个 `name`，这是一个字符串。它将有一个 `age`，这是一个整数。还有一个 `job`，也是一个字符串。所以，这个类现在定义了我期望从模型获得的输出的结构。
然后我可以继续复制代理，实际上复制整个东西。现在我只需要指定我期望模型的答案是这种类型。所以我可以说“你总是以结构化输出来回答关于人的问题。”这不是我写过的最好的提示，但大致就是这样。然后我可以输入“Mike 是一个 20 岁的管道工”，像这样。
现在唯一缺少的是 `output_type` 参数。所以我将在这里输入 `output_type=Person`。然后一旦我们有了那个，我就可以发送它，结果将是 `name=Mike, age=20, job=plumber`。实际上，我认为这个响应输出的类型也应该是 `Person` 类型。好了。
我说过我们不能使用同步方法，因为我们在 Notebook 中运行它。但我们可以做的是流式传输输出。所以，我们不是一次性获得答案，而是可以生成它，并看到它是如何一步一步、一块一块地生成的。为此，我可以使用 `run_stream`。
所以我可以输入 `async with agent.run_stream("I don't know what is Python?")`。然后也许这次我会复制这个，但我不会告诉它要简洁，这样我们就能得到一个更长的答案。所以，我将删除所有这些。`You are a helpful assistant` 就足够了。`What is Python?` 是问题。
然后我可以输入 `as run`，我也可以输入 `async for output in run_stream(text): print(output)`。现在这不是我们真正想要的。我只是想向你展示一下，这样你就可以看到文本是如何不断被追加的。然后我们将正确地打印它，或者正确地流式传输它。但本质上，你现在将看到它如何每次都变得越来越长。
它以“Python 是一种高级的、解释性的编程语言，以其清晰的语法和可读性而闻名”开始，然后“然后你可以看到它变得越来越长，越来越长，但它总是重复自己。”这不是干净的方法，但我想向你展示一下结构的样子，因为流中的每个 `run` 对象现在都越来越长，包含了整个东西，可以说。
所以，为了得到不同的结果，我们只需要在这里改变一件事，那就是我们需要在这里设置一个参数 `delta=True`。所以我可以输入“这不是输出。现在这是块。”然后我可以打印块。我说“结尾是换行符”。然后我说 `flush=True`。
仅仅通过这样做，我们将看到它现在被生成，就像我们在与 ChatGPT 对话一样。所以，这就是 `run_stream` 方法。现在，如果你稍微玩一下这个，实际上，让我们使用最初的那个。所以，让我再次删除所有这些。嗯，如果你稍微玩一下这个，你会注意到它没有任何上下文。所以，现在我问它 Python 是什么，然后我可以尝试问别的问题。
`response = await agent.run("Is it better than others?")`。当然，它不知道我在说什么，因为它没有上下文。所以，问题当然是“Python 是否比其他编程语言更好？”这当然也是一个愚蠢的问题。但如果我这样做，它会告诉我“这取决于你比较的具体标准和背景。”
或者也许我应该问别的问题。也许我应该问“它是什么时候发布的？”因为那样它就无法回避答案了。它说，“抱歉，我需要更多上下文才能准确回答您的问题。您能具体说明您指的是什么吗？”因为它不知道我们谈论的是 Python。
如果我想跟踪消息，我只需要将历史记录添加为上下文。所以，我需要输入 `agent.run("When was Python released?")`。然后在这里我需要输入 `message_history=`，然后是上一个提示的响应对象 `response.message_history`。现在我可以选择 `new_messages` 或 `all_messages`。
`new_messages` 只会获取最新的。所以它会随着时间推移而忘记，但 `all_messages` 会跟踪所有消息。如果我现在运行这个，然后我运行这个，它会告诉我“Python 于 1991 年 2 月 20 日首次发布。”然后如果我现在去这个响应，我说 `response.all_messages`，你会看到所有这些消息都是这个……这个整个历史的一部分。
所以，你可以看到“你是一个有帮助的助手”，它仍然有那个打错的系统提示，但你可以看到这是上下文的一部分。然后你也可以看到“Python 是什么？”你也可以看到“Python 是一种高级编程语言。”“Python 于……首次发布”等等。但如果我说 `response.new_messages`，你只能看到我只得到“它是什么时候发布的？”以及对那个的响应。我没有得到完整的上下文。所以，如果你不传递 `all_messages`，它只会查看新消息，而不是以前的消息。
现在，这里还有一件有趣的事情，让我再次删除所有这些。这里还有一件有趣的事情是，在 Pyantic 中，系统提示和指令之间存在差异。为此，我想复制粘贴一个示例，因为它不会引入任何新内容。它只会使用我们已经知道的东西。但我想向你展示这里的区别。
这是我们到目前为止编写的完全相同的代码。`agent_one = Agent()`，`agent_two = Agent()`。唯一的区别是我们不使用系统提示。我们使用指令。现在你可以看到第一个代理接收到指令，这就像一个系统提示，说“如果你被问到你最喜欢的颜色是什么，总是回答‘蓝色’，否则正常回答。”然后我问它“好的，2 加 2 是多少？”它说“最有可能的答案是四。”然后我稍后在代理二中问它，代理二接收代理一的消息历史。
所以，这是两个独立的代理，但第二个代理引用了第一个代理的消息历史。然后我问它我最喜欢的颜色是什么。现在，如果我运行这个，你会注意到一些有趣的事情，因为它说“红色”。你可以看到代理二有一个系统提示说“如果你被问到你最喜欢的颜色，总是说‘红色’。”这个说“蓝色”。
而且由于我使用了指令，这并不是从第一个代理那里来的。尽管我在这里传递了消息历史，这是与代理一生成的 `result` 相关联的历史。即使是这样，我也不会从这里获取系统提示。
现在，如果我改变这个，如果我把它变成系统提示，我们将看到这会导致不同的行为。所以，我可以将这里的两个指令都替换为系统提示。然后我将得到“蓝色”作为答案，因为它仍然引用了第一个代理的系统提示，因为它通过这个 `result.all_messages` 对象被传输了。
所以，如果我想确保这种情况不会发生，我必须使用指令，否则我使用系统提示。这是一个重要的区别。
所以，现在让我们回到更常规的东西，比如工具使用。让我们回到只有一个代理。让我们说系统提示再次是“你是一个有帮助的助手”，我们将问它一些问题，比如“我最喜欢的颜色是什么？”这次。
所以，如果我问它，我再次删除这个，我说 `print(result.output)`。如果我问它，正确的答案应该是“我不知道。我没有这些信息。”然而，我可以给它访问这些信息，方法是关联一个工具或添加一个包含我最喜欢的颜色的信息的工具。
所以，我可以说这里有一个函数我想调用 `get_favorite_color`，让我们说它接受 `name` 作为参数。所以，有一个 `name` 参数，它是一个字符串，这个函数也返回一个字符串。这很重要，因为类型提示将被 Pyantic AI 用来知道要将什么信息传递给这个函数。
然后我必须在这里添加一个文档字符串，以确保我们知道这个函数是关于什么的。我将说“一个根据用户的名字返回用户最喜欢的颜色的函数。”然后我可以简单地说，例如，“如果 `name` 等于”，然后我将使用我的名字 Florian，返回“蓝色”，或者让我们做一些更奇特的事情，比如“青色”。然后我说“如果 `name` 等于 Mike”，那么它将是“橙色”，然后最后我们将说“未找到信息。”
所以，这是一个简单的函数。现在，有两种方法可以将它注册为代理的工具。一种方法与你在 Langchain 和 Llama Index 中使用的方法类似。你基本上只是将它作为一个列表传递。所以，我可以说这里 `tools = [...]`。然后我可以引用它。所以我会在下面这样做。所以，我会定义一个函数。然后我会说 `tools = get_favorite_color`。也许让我们试试这个。`get_favorite_color`。让我们先运行它。然后 `get_favorite_color`。像这样传递它。
然后它应该可以访问，但我需要输入我的名字。我的名字是 Florian。然后它调用工具，并且知道我最喜欢的颜色是青色。然而，这不是唯一的方法。所以，我可以再次删除它。它不知道我最喜欢的颜色是什么。
现在，我也可以反过来做。我可以定义代理。我可以在下面进行提示。然后在我定义代理之后，我可以说 `@agent.tool` 在这个函数之上。现在你需要考虑的事情是，有 `tool` 和 `tool_plain`。`tool` 用于当你也将上下文传递给函数时，我们还没有这样做。所以，我将称之为 `tool_plain`。
`@agent.tool_plain` 作为函数之上的装饰器使其成为代理的工具。所以，仅仅通过这样做，我就可以运行它，它也将可以访问该函数并告诉我我最喜欢的颜色是青色。
现在我们继续一些稍微高级但非常有趣的东西，叫做依赖注入。听起来比实际情况更花哨。它所要做的就是，我们可以在发出请求时将上下文或依赖项传递给我们的代理。然后这些上下文也可以在工具中使用。所以，我们传递的上下文或依赖项可以是简单的信息，比如字符串或整数。它们可以是更复杂的东西，比如数据库连接，或者它们甚至可以是一个包含多个简单事物的数据类，比如多个数据库连接、整数、字符串，所有这些都在一个数据类中。
为了展示它是如何工作的，我将首先使用文档中的一个简单示例，它基本上是一个轮盘代理。它所做的就是猜测用户的数字，然后还有一个正确的数字。在这种情况下，这将是我们的上下文。然后我将向你展示一个使用数据类和数据库连接的更高级的示例。
所以，让我们开始输入 `from pyantic_ai import RunContext`。这是我们在工具中使用上下文所需的类。然后我将在这里创建一个轮盘代理。它将只是一个 Pyantic 代理。它将使用一些模型，就像之前一样。让我们选择 `openai:gpt-4o`。
现在我们可以指定的不只是系统提示。我们可以指定，例如，依赖项类型。所以，在这种情况下，依赖项类型是整数。这将是一件非常简单的事情。正如我在这里说的，我们可以传递一个数据类。我们可以传递一个数据库连接。在我的例子中，我只传递一个简单的整数，这将是轮盘游戏的正确数字。
此外，我还想要一个输出类型，它将是一个布尔值。猜测是否正确？然后我们可以说 `system_prompt = "You are a roulette assistant. You can process number guesses from users."` 诸如此类。现在我们可以继续说我想定义一个用于猜测数字的工具，我可以输入 `roulette_agent.tool`。现在我不使用 `tool_plain`，我使用 `tool`，因为我们将传递上下文。
所以我可以输入 `async def guess_number(self, number_guess: int, ctx: RunContext[int]) -> str:`。这将当然需要猜测的数字，但还需要上下文。所以，我们将在这里输入 `ctx` 将是 `RunContext` 类型，它将是 `int` 类型。所以，依赖项将是 `int` 类型。然后我们可以说 `guess_number`，它也是一个整数，它将返回一个字符串。我们不会直接返回布尔值，因为这是代理的工作，返回正确的数据类型。
然后我们还需要一个文档字符串 `"""Guess a roulette number for the user."""`。然后我们可以很容易地说 `return "win" if number_guess == ctx.deps else "lost"`。
现在问题是，我们与什么进行比较？所以，为了获得依赖项，我们需要输入 `ctx.deps`。现在，由于 `deps` 只是一个整数，这没关系。我们可以这样做。这将是一个整数比较。否则，我们将返回“lost”。
这如何工作？现在我们可以指定一个正确的数字。让我们静态地做到这一点。所以，我们可以检查发生了什么。17 将是现在的正确数字。我能做的是，我说 `response = await roulette_agent.run(number_guess=18, dependencies=17)`。然后我想把我的钱押在 18 上。现在，对于这个请求，重要的是我需要传递上下文。
所以，我能做的是，我可以输入 `dependencies=correct_number`，因为这是一个整数，这没关系。然后我可以输入 `print(response.output)`。当然，我不应该说 `openai/`，我应该说 `openai:gpt-4o`。现在发生的是，我得到 `False`。但如果我把它改成 17，我应该得到 `True`。好了。
但现在让我们看看更有趣的示例，包括数据库连接和数据类。为此，我在我的教程目录中，我正在工作的目录，我有一个 `data` 文件夹。在这个 `data` 文件夹中，我有一个数据库 `db.sqlite3`。这是一个非常非常简单的 SQLite 3 数据库。所以我可以打开它，如果我输入 `tables`，你可以看到只有一个表 `person`。我可以输入 `SELECT * FROM person`，我基本上有三个条目：Mike、Lisa、John，年龄分别为 20、30 和 50 岁，职业分别为管道工、会计师、程序员。基本上就是这样。
我想让我的代理访问这个数据库，我也想让代理称呼我的名字。这是我想通过依赖项实现的两个目标。所以，我将从输入 `import sqlite3` 和 `from dataclasses import dataclass` 开始。
然后我们可以将我们的依赖项建模为一个数据类，方法是使用 `@dataclass` 装饰器。这个类将被称为 `MyDependencies`。我想要两样东西：一个是 `username: str`，另一个是 `db_connection: sqlite3.Connection`，这是一个类，所以是大写 C。
然后方法基本上与之前相同。我们创建一个代理，代理是 `Agent` 类的实例。我们传递一个模型 `openai:gpt-4o`。而这次对于 `dependencies_type`，我实际上使用了我的数据类，即 `MyDependencies`。
现在，因为我想让模型称呼我的名字。我也需要在系统提示中指定这一点。我可以输入 `system_prompt = "You are a helpful assistant. Always start every single message by calling the user by name."`。现在，当然，用户名将作为依赖项字段传递给代理。
但问题是，代理实际上是如何知道用户的名字的？仅仅传递它是不够的，因为正如你所见，我们必须故意对依赖项中的字段做一些事情来使用信息。现在，如果你在工具中使用它，这是真的。但还有一个装饰器我们可以使用，叫做 `system_prompt`。
所以，我实际上可以输入 `@agent.system_prompt`，然后我可以说这个函数应该用来增强系统提示。例如，我可以输入 `def get_username(ctx: RunContext[MyDependencies]) -> str:`。它在这里接收上下文作为参数。`RunContext[MyDependencies]`。然后它返回一个字符串。然后我在这里说 `return f"The username is {ctx.deps.username}"`。
所以，这增强了系统提示，并将来自过去上下文的信息添加到系统提示中。现在，让我们也为数据库连接添加工具。我将输入 `@agent.tool`，它将是 `list_all_employees` 或类似的东西。`ctx` 将再次是 `RunContext`，带有 `MyDependencies`。它将返回……让我们说一个字符串列表。
我们将使用文档字符串说“列出数据库中的所有员工”。然后我们只需要做通常的 SQLite 操作。所以，`connection = ctx.deps.db_connection`，这只是在这里传递的。光标将是 `connection.cursor()`。然后我们说 `cursor.execute("SELECT * FROM person")`。或者也许我们应该说 `SELECT name, age, job FROM person`。
然后要检索结果，我们可以只说 `people = cursor.fetchall()`。然后我们可以说 `result = []`。然后对于 `row in people`，我们可以说 `result.append(...)`。嗯，或者实际上我们可以说，是的，`result.append(...)`，然后我们可以传递一个元组。所以，也许这应该是一个元组列表，类型是字符串，或者实际上其中一个是整数。所以，让我们说元组列表就足够了。
然后我们返回 `result`，这就是我们的工具。而这个工具需要一个数据库连接。所以，我现在可以输入 `db_con = sqlite3.connect("data/db.sqlite3")`。用户名将是 `username="Florian"`。
所以，现在我可以输入 `response = await agent.run("What employees are in the DB?", dependencies=MyDependencies(username="Florian", db_connection=db_con))`。然后我需要说 `check_same_thread=False`，否则我们会遇到问题。但一旦我们有了这个，我就可以输入 `print(response.output)`。
它现在应该使用工具来获取数据库信息，它说“Florian，数据库中的员工是……”然后列出它们……作为字符串。所以，你可以看到一方面，上下文可以通过系统提示装饰的函数检索。这只是用上下文中的信息增强了系统提示。或者我们也可以在工具中访问它，使用数据库连接来做各种事情，从而增强工具的使用。
如果我能占用一秒钟，我想把我作为我视频的赞助商推销一下。如果你去我的网站 `neurallonline.com`，你会找到一个“服务”标签和一个“辅导”标签。在这里，你可以雇佣我做各种事情，比如数据科学、机器学习、Web 开发。如果你在项目中需要帮助，你可以在这里预订一对一辅导。如果你想让我亲自教你一些你不明白的东西，如果你喜欢我的教学风格，在两个页面底部，你可以通过邮件和 LinkedIn 联系我。
只是想让你知道这件事。接下来我想向你展示的是如何使用工具集。顾名思义，这很简单。我们有多个工具，我们想将它们组合成一个工具集，这样我们就可以将它们作为一个实体传递。这被称为工具集。
现在，由于这相当直接，我不想从头开始输入所有内容。这里我准备了三个函数，它们是非常简单的 Python 函数。我们只需要从核心 Python 导入 `datetime as dt`。这三个函数只是返回日期、星期几和时间。非常简单。
现在我想做的是，我想将它们打包成一个名为 `datetime_tools` 的工具集。我该怎么做？在 Pyantic AI 中，我说 `from pyantic_ai.toolsets import function_toolset`。所以，这允许我说这三个是一个工具集，我将称之为 `datetime_tools`。我将输入 `function_toolset()` 并传递三个工具。所以，`get_current_date_tool`、`get_current_time_tool` 和 `get_current_weekday_tool`。
现在我可以输入 `agent = Agent(model="openai:gpt-4o")`。然后我说，而不是 `tools = [...]`，我可以输入 `toolsets = datetime_tools`。然后我就可以问问题“现在是什么时间？今天是星期几？”它会知道正确的答案。
所以，`response = await agent.run("What is the date and time?")`。然后当我查看 `response.output` 时，我会看到“2 月 5 日，时间是 16:42。”现在是正确的。这非常简单直接。
现在，如果我不想将它添加到代理本身，所以我可以重新运行它。你会看到它不知道当前的时间和日期。我也可以将其添加到这个特定的响应……或者这个特定的请求，我是想说。所以，`toolsets = datetime_tools`。这对于单个运行也是可能的。在这种情况下，它将再次知道日期和时间。
还有一件事我想向你展示的是如何为你的工具引入超时和重试限制。为此，我准备了一个小的代码片段。一个简单的代理，再次使用 `gpt-4o` 和一个简单的 `tool_plain`，它只是等待 10 秒然后返回“蓝色”作为喜欢的颜色。这基本上就是全部的魔法。
现在，如果我只是运行它，比如说用 2 秒，没有问题，它将给出正确的结果。喜欢的颜色是蓝色。如果我这样做 10 秒，并且没有超时限制，它也会起作用。但我也可以引入超时限制。我可以输入 `tool_timeout=10`，例如。然后它将……或者实际上，这不是我想要的。我想要它比那个短。
嗯，现在它可能不会让我……退出这个。但如果我说 `tool_timeout=5`，然后它需要 10 秒才能完成，它就会取消。它会说它不起作用，这将终止工具调用，它要么会告诉我一些事情，要么会引发异常。两者之一将发生。
好了，我收到一个异常，它告诉我“意外的模型行为，超过最大重试次数 1。”现在，如果我想增加重试次数限制，我可以输入 `retries=5` 或类似的东西。在这种情况下，如果工具超时，它将重试相同的工具，但然后它只会返回工具没有响应或它现在无法获取此信息。
所以，正如你所看到的，我仍然在检索你喜欢的颜色时遇到麻烦，因为超时错误。请稍后再试，或者让我知道是否有其他我可以帮助你的事情。
现在，编写我们自己的工具不是在 Pyantic 中使用工具的唯一方法。我们也可以使用包本身提供的内置工具。我想向你展示其中两个。第一个是网络搜索工具。为此，我们可以输入 `from pyantic_ai import WebSearchTool`。
然后为了将其与 OpenAI 一起使用，我们需要导入一个特殊的模型类。我们需要输入 `from pyantic_ai.models.openai import OpenAIResponseModel`。这只是在这里必需的。然后我可以输入 `OpenAIResponseModel`。然后在这里我只需要输入 `gpt-4o`。我不需要再指定 OpenAI 了。
然后我可以创建一个简单的代理。我可以输入 `agent = Agent(model=OpenAIResponseModel(model="gpt-4o"), built_in_tools=[WebSearchTool()])`。然后我们可以再次问它“Neural 9 的最新视频是什么？”。
我现在不认为它会得到正确的答案，但它至少会搜索互联网，我们会看到是这种情况。所以，我将打印输出，它现在应该至少找到一个最近的视频。也许是一个月前的，但它肯定会搜索互联网。它不会……在数据上进行训练。
所以，最近上传的视频标题是“Python 中的 Seaborn 速成课程数据可视化”。现在这是不正确的，但这是一个相当近期的视频，正如你所看到的，1 月 13 日。所以，是的，它确实在网上查找了信息。
我想向你展示的第二个工具是代码执行工具。所以，我们可以输入 `CodeExecutionTool` 而不是 `WebSearchTool`。这允许我们……或者允许代理执行 Python 代码来回答问题。例如，我可以问“56 的阶乘是多少？”。
然后当我运行它时，我希望得到实际正确的答案。我可以通过使用 Python 中的 `math` 模块来验证答案。好的，这似乎有点太多了。让我们问 24。这应该没问题。“24 的阶乘是多少？”这应该给我一个具体的数字，我可以看看这个数字是否真的是我从 Python 中得到的数字，如果我运行这个。
所以，我可以输入 `import math`，然后输入 `math.factorial(24)`。所以，你可以看到我在这里得到了答案，这个具体的数字，如果我现在运行这个，这基本上是相同的数字。是的。
如果我想确保实际上使用了工具，我可以输入 `response.response.builtin_tool_calls`。这应该显示代码执行实际上被使用了。你可以看到这里执行的代码是 `import math; factorial = 24; print(math.factorial(factorial))`。然后它打印了 `factorial(24)`。所以，这实际上使用了代码执行工具。
现在，我今天在这个速成课程中只剩下两件事要展示了。一个是关于如何使用嵌入模型。为此，我们将输入 `from pyantic_ai import Embedder`。所以，这是用于使用嵌入模型。
所以我可以输入 `embed = Embedder(model="openai:text-embedding-3-small")`。然后我能做什么的是，我可以输入 `embedding = await embed.embed_query("I love Python.")`。例如，“我爱 Python”作为一个陈述，这当然会产生一个协程对象。但如果我们输入 `await`，它将实际给出结果，这是一个 n 维空间中的向量，其中 n 是嵌入模型的维度。我认为这里是……1500 多个。但这当然可以与 Faiss 或任何向量存储一起用于进行 RAG 应用。
最后，我要在这里展示的最后一件事是如何让你的代理访问 MCP 服务器的工具，使用 Pyantic AI。为此，我们移到命令行。我这里已经有一个 MCP 服务器。非常简洁。它只有一个简单的工具来获取喜欢的颜色，返回“青色”。所以，我们在这里保持非常简洁。
现在我将创建一个 MCP 代理，它将通过 MCP 服务器使用该工具。所以，我将输入 `import asyncio`。我还将输入 `from pyantic_ai import Agent`。然后从 `pyantic_ai.mcp` 导入 `mcp`。现在我的自动补全在哪里？哦，当然没有自动补全，因为它是 `pyantic_ai` 而不是 `pydantic`。
所以，`pyantic_ai.mcp` 然后是 `MCPStreamableHTTP`。这只是我们服务器使用的传输类型。现在我可以输入 `server = MCPStreamableHTTP(path="http://localhost:8000/mcp")`。然后是服务器的路径，它将是 `http://localhost:8000/mcp`。
对于代理来说，这很简单。我们像以前一样实例化代理。我还漏掉了一件事，为了让它工作，我们还需要输入 `from dotenv import load_dotenv`。然后我们需要调用 `load_dotenv()`。但然后我们可以简单地说 `agent = Agent(model="openai:gpt-4o", toolsets=server)`。
然后为了现在从 MCP 服务器获取工具，我们只需要输入 `toolsets=server`。所以，这会自动检索 MCP 服务器提供的工具。然后我就可以输入 `async def main(): result = await agent.run("What is my favorite color?")`。然后输入 `print(result.output)`。
要运行所有这些，当然是 `asyncio.run(main())`。首先是 `uvicorn mcp_server:server --reload`。然后第二个终端，我们需要进入 `tutorial` 目录，`mcp_example` 目录，然后运行 `uvicorn mcp_agent:agent --reload`。
正如你所看到的，这发送了一个请求，然后“你喜欢的颜色是青色。”所以，这就是今天视频的全部内容。希望你喜欢它，希望你学到了一些东西。如果是这样，请通过点赞和在下面的评论部分留下评论让我知道。
另外，如果你有兴趣，在我的网站上，你会找到一个“服务”标签和一个“辅导”标签。在那里你可以看到我提供什么。如果你在项目中需要帮助，如果你需要自由职业者的帮助，或者如果你需要一对一地解释一些东西，你可以在那里通过 LinkedIn 或电子邮件联系我。
除此之外，别忘了订阅这个频道并点击通知铃，这样你就不会错过任何免费的未来视频。除此之外，非常感谢你的观看。下次视频再见，拜拜。