Pydantic AI + Tavily Web Scraper + Llama 3.3 = AI 研究代理

Rifx.Online
Generative AI , Data Science , Programming
28 Feb, 2025

照片由 🇸🇮 Janko Ferlič 提供，来源于 Unsplash

免责声明：我使用 GPT 搜索收集事实。整篇文章由我撰写。概念来自高达烈

介绍

我一直相信，最好的工具来自于去除噪音，专注于真正重要的事情。当我们着手构建一个 AI 研究代理时，我们需要知道三件事：

结构化数据验证 (感谢 Pydantic AI)
实时网络数据 (通过可靠的网络爬虫如 Tavily)
上下文感知推理 (由 Llama 3.3 提供支持)

在本指南中，我将分享如何在 500 行代码以内构建您的生产就绪代理。

为什么选择这个技术栈？

在研究人工智能趋势时，保持准确、结构化的数据至关重要。这个技术栈让我能够：

严格验证数据： Pydantic AI 强制执行严格的输出格式。
获取实时信息： Tavily 过滤掉 SEO 噪音，提取干净的数据。
深入推理： Llama 3.3，拥有 70B 参数，提供的推理甚至超越了 GPT-4 的基准。

只是真实的、可靠的性能。

第一步：设置环境

您可以通过在 requirements.txt 文件中列出依赖项来开始：

pydantic-ai>=0.0.24 tavily-python>=0.3.1 streamlit>=1.33.0 nest-asyncio>=1.5.8 python-dotenv>=1.0.0

然后，操作就简单了：

pip install -r requirements.txt

第2步：理解核心组件

A. Pydantic AI 的魔力 这个工具确保每一条数据都符合预期的结构。它强制类型安全，使用依赖注入来处理 API 密钥和变量，甚至实时验证流式 JSON。

例如，以下是如何定义研究结果：

from pydantic import BaseModel, Field

class ResearchResult(BaseModel): title: str = Field(…, max_length=120) summary: str sources: list[str]

B. Tavily 网页抓取器 Tavily 处理网页抓取的繁重工作。它自动过滤掉 SEO 垃圾，并返回原始 HTML 和清理后的文本以及元数据。

from tavily import AsyncTavilyClient

tavily = AsyncTavilyClient(api_key=“YOUR_KEY”) results = await tavily.search(query=“Latest LLM releases 2025”, max_results=5)

C. Llama 3.3的前沿

Llama 3.3不仅仅是另一个模型——它专为真实推理而构建，拥有70B参数和原生JSON输出以提供结构化响应。此外，它是MIT许可证的，非常适合商业项目。

第 3 步：构建代理管道

您设计的管道首先抓取网络，然后使用 Llama 3.3 处理数据，最后使用 Pydantic 验证输出：

import asyncio
from pydantic_ai import Agent

class ResearchAgent(Agent):
    def __init__(self):
        self.model = OpenAIModel('llama3.3:latest', client=AsyncOpenAI(
            base_url='YOUR_LLAMA_ENDPOINT',
            api_key='your-key'
        ))
        self.scraper = AsyncTavilyClient(api_key=os.getenv("TAVILY_API_KEY"))

    async def run(self, query: str) -> ResearchResult:
        raw = await self.scraper.search(query=query, max_results=5)
        prompt = f"Analyze: {raw}\\nToday's date: {datetime.today().isoformat()}"
        response = await self.model.generate(prompt)
        return ResearchResult.model_validate_json(response)

第4步：像专业人士一样处理边缘情况

即使是最好的系统也会遇到问题。例如：

虚构的URL： LLM可能会编造URL。你的解决方案？将每个来源与抓取的元数据进行交叉检查。

def validate_sources(llm_output: dict, scraped_data: dict) -> bool:
    return all(url in scraped_data['sources'] for url in llm_output['sources'])

速率限制： 与其在首次遇到速率限制时就放弃，不如实现指数退避。

import time

async def safe_scrape(query: str, retries=3):
    for i in range(retries):
        try:
            return await tavily.search(query)
        except RateLimitError:
            wait = 2 ** i
            time.sleep(wait)
    raise Exception("API unavailable")

第5步：使用Streamlit快速部署

将所有内容整合到一个用户界面中只需15行代码：

import streamlit as st

st.title("AI Research Assistant")
query = st.text_input("Ask about AI trends:")

if st.button("Search"):
    with st.spinner("Analyzing..."):
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        result = loop.run_until_complete(ResearchAgent().run(query))

    st.subheader(result.title)
    st.markdown(f"**Summary**: {result.summary}")
    st.write("**Sources**:")
    for url in result.sources:
        st.markdown(f"- [{url}]({url})")

何时使用此架构

您可以为以下情况构建此架构：

实时市场研究
学术论文分析
竞争情报

这是一个快速获取可操作见解的实用工具。

最终生产提示

监控一切： 您可以使用 Pydantic 的 Logfire 集成来跟踪 API 调用成功、LLM 验证问题和爬虫相关性。

from pydantic_ai import logfire

logfire.configure(project_name=“research_agent”) logfire.instrument()

优化： 添加速率限制、缓存和定期监控，以保持系统的可靠性。
成本效益： 使用 Llama 3.3 的 MIT 许可，您可以避免像 GPT-4（$20/百万个令牌）这样的替代品的高成本。

核心代码概述

1. 配置 LLM & 爬虫

client = AsyncOpenAI(base_url=‘YOUR_LLAMA_ENDPOINT’, api_key=‘your-key’) model = OpenAIModel(‘llama3.3:latest’, client=client)

tavily = AsyncTavilyClient(api_key=os.getenv(“TAVILY_API_KEY”))

2. 定义数据结构

from pydantic import BaseModel, Field

class ResearchResult(BaseModel): title: str = Field(description=”# 最新的 LLM 发展”) analysis: str = Field(max_length=1000) sources: list[str] key_points: list[str]

3. 构建代理

class ResearchAgent(Agent): async def run(self, query: str) -> ResearchResult:

raw = await tavily.search(query=query, max\_results=5)  


prompt = f"分析： {raw}\\n今天的日期： {datetime.today().isoformat()}"  
response = await model.generate(prompt)  


return ResearchResult.model\_validate\_json(response)

4. 关键优化数据验证层

@search_agent.tool async def validate_sources(sources: list[str]) -> list[str]: return [url for url in sources if “arxiv.org” in url or “ai-benchmark.org” in url]

错误处理

from tenacity import retry, wait_exponential

@retry(wait=wait_exponential(multiplier=1, min=4, max=10)) async def safe_search(query: str): return await tavily.search(query)

5. 8 行代码部署 (Streamlit)

import streamlit as st

st.title(“AI 研究助手”) query = st.text_input(“询问关于 LLM 的问题：”)

if st.button(“分析”): result = asyncio.run(ResearchAgent().run(query)) st.markdown(f”## {result.title}”) st.write(result.analysis) st.json({“sources”: result.sources, “key_points”: result.key_points})