基于 Pydantic AI 与 Llama3.3 构建智能研究代理实战

安装完成后，我们导入 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()  # 加载环境变量

四、模型与 API 配置

1. 设置 LLM 提供者

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 模型。

2. 配置 Tavily 搜索客户端

我们使用 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)

五、核心数据结构定义

为了规范输入输出，我们定义了三个类。这些类确保了数据的结构化和可验证性。

1. 搜索上下文数据类

SearchDataclass 是一个数据类，用于存储与搜索相关的信息，特别是最大结果数 (max_results) 和今天的日期 (todays_date)。

@dataclass
class SearchDataclass:
    max_results: int
    todays_date: str

2. 代理依赖数据类

第二个类 ResearchDependencies 是另一个数据类，仅存储今天的日期，用于传递给 Agent 的系统提示中，确保回答的时效性。

@dataclass
class ResearchDependencies:
    todays_date: str

3. 研究结果模型

第三个类 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 定义与工具集成

1. 创建研究 Agent

我们创建了一个 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.'
)

2. 动态系统提示

我们创建一个 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

3. 定义搜索工具

我们定义了两个异步函数：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)

让我们设置一个 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}")

八、最佳实践与扩展建议

1. 错误处理

在生产环境中，网络请求可能会超时或失败。建议在 do_search 函数中加入 try-except 块，捕获 TimeoutError 或 ConnectionError，并提供友好的错误提示给用户。

2. 缓存机制

对于相同的查询，可以引入 Redis 或其他缓存机制，避免重复调用 Tavily API 和 LLM，从而降低成本并提高响应速度。

3. 多 Agent 协作

当前架构为单 Agent 模式。未来可以扩展为多 Agent 架构，例如一个 Agent 负责搜索，另一个 Agent 负责摘要，第三个 Agent 负责事实核查，以提高输出的准确性。

4. 部署建议

• 本地部署：适用于个人使用，数据隐私性好。
• 云端部署：使用 Docker 容器化应用，部署到 Kubernetes 或云服务器，便于扩展和维护。
• API 封装：可以将后端逻辑封装为 FastAPI 服务，供其他前端调用。

九、结论

Pydantic AI 是一个出色的库，提供了许多实现相同功能的替代方案之外的独特价值。通过严格的类型系统和标准化的开发模型，它帮助开发者构建更可靠的应用程序。

无论是构建一个简单的聊天机器人还是一个复杂的系统，Pydantic AI 提供的功能使开发过程更加顺畅，最终产品更加可靠。结合 Llama3.3 的强大推理能力和 Tavily 的精准搜索能力，我们可以构建出真正具备实用价值的 AI 研究代理。

希望本文提供的示例能够帮助您更快、更轻松地掌握 Pydantic AI 及其生态系统的用法。随着 AI 技术的不断发展，期待看到更多创新的应用涌现。