PythonAI
基于 Pydantic AI 与 Llama3.3 构建智能研究代理实战
本文介绍如何使用 Pydantic AI 框架结合 Llama3.3 模型和 Tavily 搜索工具构建智能研究代理。通过类型安全的响应验证、依赖注入系统以及 Streamlit 界面，实现从查询输入到结构化结果输出的完整流程。内容涵盖环境配置、核心代码解析、Agent 定义及 UI 开发，旨在提供生产级 AI 应用开发的最佳实践参考。
本文介绍如何使用 Pydantic AI 框架结合 Llama3.3 模型和 Tavily 搜索工具构建智能研究代理。通过类型安全的响应验证、依赖注入系统以及 Streamlit 界面，实现从查询输入到结构化结果输出的完整流程。内容涵盖环境配置、核心代码解析、Agent 定义及 UI 开发，旨在提供生产级 AI 应用开发的最佳实践参考。
在检索增强生成（RAG）和基于大型语言模型（LLM）的工作流中，结构化输出显著提高了准确性和清晰度，使数据更易于理解和后续处理。许多开发人员在验证或转换数据为正确格式时面临挑战，特别是在处理接口数据时，复杂的数据格式容易导致难以发现的错误。
Pydantic 作为知名的数据验证工具，在此过程中发挥着关键作用。OpenAI、Anthropic、LangChain 和 LlamaIndex 等主流框架均将 Pydantic 作为核心组件，负责数据验证等重要功能。近期，Pydantic 团队推出了 PydanticAI，这是一个基于 Pydantic 的 AI 代理框架，旨在简化 AI 应用开发的复杂性，并解决 AI 代理开发中的各种痛点。
本文将详细演示如何使用 Pydantic AI、Web Scraper（Tavily）和 Llama3.3 创建一个多代理聊天机器人，为您的业务或个人使用构建一个强大的代理系统。
本方案采用以下核心技术组合：
PydanticAI 提倡类型安全操作、结构化响应验证以及一种新颖的依赖注入系统，所有这些都在熟悉的 Python 最佳实践领域内进行。这使得开发人员能够在不牺牲代码质量或安全性的情况下，利用生成性 AI 的力量。
在深入应用程序之前，我们需要创建一个理想的环境以使代码正常工作。首先，确保已安装 Python 3.9+ 环境。
我们将使用 requirements.txt 来管理依赖，或者手动安装以下库：
pip install pydantic-ai pydantic tavily-python streamlit python-dotenv nest_asyncio openai dataclasses-json
如果存在 requirements.txt 文件，可以直接运行：
pip install -r requirements.txt
安装完成后，我们导入 Pydantic AI、dataclasses、tavily、streamlit 和 devtools 等相关库。
import os
import asyncio
import datetime
from typing import Any
from dataclasses import dataclass
import nest_asyncio
nest_asyncio.apply()
from openai import AsyncOpenAI
from pydantic_ai.models.openai import OpenAIModel
import streamlit as st
from pydantic_ai import Agent, RunContext
from pydantic import BaseModel, Field
from tavily import AsyncTavilyClient
from dotenv import load_dotenv
load_dotenv() # 加载环境变量
Pydantic 直接与 OpenAI、Groq 和 VertexAI 配合使用。但在本教程中，我们将使用 Ollama，它现在与 OpenAI Chat Completions API 内置兼容，使得可以在本地使用更多工具和应用程序，无需支付 API 费用。
client = AsyncOpenAI(
base_url='http://localhost:11434/v1',
api_key='ollama', # Ollama 通常不需要真实 Key，填任意字符串即可
)
model = OpenAIModel('llama3.3:latest', openai_client=client)
注意：请确保已在本地安装并运行 Ollama，且拉取了 llama3.3 模型。
我们使用 Tavily 来抓取浏览器、过滤和聚合数据。需要获取 API 密钥并设置为环境变量 TAVILY_API_KEY。
TAVILY_API_KEY = os.getenv("TAVILY_API_KEY")
if not TAVILY_API_KEY:
raise ValueError("Please set TAVILY_API_KEY environment variable.")
tavily_client = AsyncTavilyClient(api_key=TAVILY_API_KEY)
为了规范输入输出，我们定义了三个类。这些类确保了数据的结构化和可验证性。
SearchDataclass 是一个数据类，用于存储与搜索相关的信息，特别是最大结果数 (max_results) 和今天的日期 (todays_date)。
@dataclass
class SearchDataclass:
max_results: int
todays_date: str
第二个类 ResearchDependencies 是另一个数据类，仅存储今天的日期，用于传递给 Agent 的系统提示中，确保回答的时效性。
@dataclass
class ResearchDependencies:
todays_date: str
第三个类 ResearchResult 继承自 BaseModel，表示一篇研究文章，包含文章标题 (research_title)、主要内容 (research_main) 和一组总结要点的项目符号 (research_bullets)。
Field 函数用于为每个属性添加描述，有助于验证和文档，同时指导 LLM 生成符合预期的内容。
class ResearchResult(BaseModel):
research_title: str = Field(description='Markdown heading describing the article topic, prefixed with #')
research_main: str = Field(description='A main section that provides a detailed news article')
research_bullets: str = Field(description='A set of bullet points summarizing key points')
我们创建了一个 Agent (基于 Llama3.3) 用于研究任务。它使用 ResearchDependencies 数据类作为输入，使用 ResearchResult 类作为输出。
接着，我们编写一个 系统提示，指示它从查询中识别关键词，执行多次搜索，然后将这些结果合并成详细响应。
search_agent = Agent(
model,
deps_type=ResearchDependencies,
result_type=ResearchResult,
system_prompt='You are a helpful research assistant, you are an expert in research. '
'When given a query, you will identify strong keywords to do 3-5 searches using the provided search tool. '
'Then combine results into a detailed response.'
)
我们创建一个 add_current_date 函数，以指示代理从给定问题中识别强关键词，使用这些关键词进行 3-5 次搜索，并将结果合并成详细响应，同时确保信息准确且最新。
@search_agent.system_prompt
async def add_current_date(ctx: RunContext[ResearchDependencies]) -> str:
todays_date = ctx.deps.todays_date
system_prompt = (
f"You're a helpful research assistant and an expert in research. "
f"When given a question, write strong keywords to do 3-5 searches in total "
f"(each with a query_number) and then combine the results. "
f"If you need today's date it is {todays_date}. "
f"Focus on providing accurate and current information."
)
return system_prompt
我们定义了两个异步函数：get_search 和 do_search。
get_search 是 search_agent 用于执行搜索的工具。它接受搜索查询和搜索上下文（包括最大结果数），并使用 tavily_client 来检索搜索结果，将其作为字典返回。do_search 通过创建 SearchDataclass 的实例（包括当前日期和最大结果数）来准备必要的依赖项。然后它使用这些依赖项和查询运行 search_agent，等待结果。@search_agent.tool
async def get_search(search_data: RunContext[SearchDataclass], query: str, query_number: int) -> dict[str, Any]:
"""Perform a search using the Tavily client."""
max_results = search_data.deps.max_results
results = await tavily_client.get_search_context(query=query, max_results=max_results)
return results
async def do_search(query: str, max_results: int):
# Prepare dependencies
current_date = datetime.date.today()
date_string = current_date.strftime("%Y-%m-%d")
deps = SearchDataclass(max_results=max_results, todays_date=date_string)
# Run the agent
result = await search_agent.run(query, deps=deps)
return result.data
注意：在实际生产中，建议增加异常处理机制，例如当 Tavily 请求失败时的重试逻辑。
让我们设置一个 Streamlit 应用程序，用户可以输入查询并指定要检索的搜索结果数量。在我们点击按钮以启动搜索后，应用程序将获取相关的研究数据（包括标题、主要文章和关键要点），并以组织良好的格式显示。
st.set_page_config(page_title="AI News Researcher", layout="centered")
st.title("Large Language Model News Researcher")
st.write("Stay updated on the latest trends and developments in Large Language Model.")
## User input section
st.sidebar.title("Search Parameters")
query = st.sidebar.text_input("Enter your query:", value="latest Large Language Model news")
max_results = st.sidebar.slider("Number of search results:", min_value=3, max_value=10, value=5)
st.write("Use the sidebar to adjust search parameters.")
if st.button("Get Latest Large Language Model News"):
with st.spinner("Researching, please wait..."):
try:
result_data = asyncio.run(do_search(query, max_results))
st.markdown(result_data.research_title)
# A bit of styling for the main article
st.markdown(f"<div style='line-height:1.6;'>{result_data.research_main}</div>", unsafe_allow_html=True)
st.markdown("### Key Takeaways")
st.markdown(result_data.research_bullets)
except Exception as e:
st.error(f"An error occurred during research: {e}")
在生产环境中，网络请求可能会超时或失败。建议在 do_search 函数中加入 try-except 块，捕获 TimeoutError 或 ConnectionError，并提供友好的错误提示给用户。
对于相同的查询，可以引入 Redis 或其他缓存机制，避免重复调用 Tavily API 和 LLM，从而降低成本并提高响应速度。
当前架构为单 Agent 模式。未来可以扩展为多 Agent 架构，例如一个 Agent 负责搜索，另一个 Agent 负责摘要，第三个 Agent 负责事实核查，以提高输出的准确性。
Pydantic AI 是一个出色的库，提供了许多实现相同功能的替代方案之外的独特价值。通过严格的类型系统和标准化的开发模型，它帮助开发者构建更可靠的应用程序。
无论是构建一个简单的聊天机器人还是一个复杂的系统，Pydantic AI 提供的功能使开发过程更加顺畅，最终产品更加可靠。结合 Llama3.3 的强大推理能力和 Tavily 的精准搜索能力，我们可以构建出真正具备实用价值的 AI 研究代理。
希望本文提供的示例能够帮助您更快、更轻松地掌握 Pydantic AI 及其生态系统的用法。随着 AI 技术的不断发展，期待看到更多创新的应用涌现。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online