Python

当类型安全遇上 AI Agent Pydantic AI 是 Pydantic 团队在 2025 年推出的 Agent 框架，核心理念是将 Python 的类型系统引入 Agent 开发。在大多数 Agent 框架中，LLM 输出的结构化数据依赖运行时验证——如果 LLM 输出不符合预期，你只能在运行时发现问题。Pydantic AI 通过编译时类型检查 + 运行时 Schema 验证，将这类问题消灭在开发阶段。 核心设计 类型驱动的 Agent 定义 from pydantic_ai import Agent, RunContext from pydantic import BaseModel, Field from typing import Literal from dataclasses import dataclass # 定义结构化输出 class ResearchReport(BaseModel): title: str = Field(description="报告标题") summary: str = Field(description="执行摘要，不超过200字") key_findings: list[str] = Field(description="关键发现列表") confidence: float = Field(ge=0, le=1, description="置信度 0-1") sources: list[str] = Field(description="信息来源 URL 列表") recommendation: Literal["buy", "hold", "sell"] = Field(description="投资建议") # 定义依赖类型 @dataclass class ResearchDeps: api_keys: dict[str, str] database_url: str max_sources: int = 10 # 创建 Agent research_agent = Agent( model="openai:gpt-4o", deps_type=ResearchDeps, output_type=ResearchReport, # 类型安全的输出 system_prompt="你是一名专业的研究分析师..." ) # 类型安全的工具定义 @research_agent.tool async def search_database(ctx: RunContext[ResearchDeps], query: str, limit: int = 10) -> list[dict]: """ 搜索内部数据库。 Args: query: 搜索查询 limit: 返回结果数量上限 """ # ctx.deps 是 ResearchDeps 类型，IDE 自动补全 async with httpx.AsyncClient() as client: response = await client.post( f"{ctx.deps.database_url}/search", json={"query": query, "limit": min(limit, ctx.deps.max_sources)}, headers={"Authorization": f"Bearer {ctx.deps.api_keys['database']}"} ) return response.json()["results"] @research_agent.tool async def fetch_web_page(ctx: RunContext[ResearchDeps], url: str) -> str: """获取网页内容。""" async with httpx.AsyncClient() as client: response = await client.get(url, timeout=10) return response.text[:5000] # 限制内容长度 # 运行 Agent —— 输出类型自动验证 result = await research_agent.run( "分析 2026 年 AGI 芯片市场", deps=ResearchDeps( api_keys={"database": "xxx"}, database_url="https://api.example.com" ) ) # result.output 是 ResearchReport 类型，IDE 完整支持 print(result.output.title) print(result.output.confidence) print(result.output.recommendation) # 如果 LLM 输出不符合 Schema，Pydantic 会抛出 # ValidationError，而非静默返回错误数据 核心特性 1. 结构化输出保证 from pydantic_ai import Agent from pydantic import BaseModel, validator from typing import Optional class CodeReview(BaseModel): file_name: str issues: list["CodeIssue"] overall_rating: int = Field(ge=1, le=10, description="整体评分 1-10") approved: bool @validator("file_name") def validate_filename(cls, v): if not v.endswith((".py", ".js", ".ts")): raise ValueError("仅支持 .py/.js/.ts 文件") return v class CodeIssue(BaseModel): line_number: int = Field(ge=1) severity: Literal["error", "warning", "info"] message: str suggestion: Optional[str] = None review_agent = Agent( model="anthropic:claude-4-sonnet", output_type=CodeReview, system_prompt="你是代码审查专家，分析代码质量问题。" ) # Agent 输出保证符合 CodeReview Schema result = await review_agent.run("审查以下代码: ...") review: CodeReview = result.output # 类型保证 2. 多模型支持与切换 from pydantic_ai.models import Model from pydantic_ai.models.openai import OpenAIModel from pydantic_ai.models.anthropic import AnthropicModel from pydantic_ai.models.groq import GroqModel # 环境感知模型选择 def get_model() -> Model: env = os.getenv("ENVIRONMENT", "dev") if env == "production": return OpenAIModel("gpt-4o", api_key=os.getenv("OPENAI_API_KEY")) elif env == "staging": return AnthropicModel("claude-4-sonnet", api_key=os.getenv("ANTHROPIC_API_KEY")) elif env == "dev": return GroqModel("llama-4-70b", api_key=os.getenv("GROQ_API_KEY")) else: # 使用 Ollama 本地模型 return OpenAIModel( "qwen3:72b", base_url="http://localhost:11434/v1", api_key="ollama" ) agent = Agent( model=get_model(), output_type=ResearchReport, system_prompt="..." ) # 运行时切换模型 result = await agent.run("...", model=AnthropicModel("claude-4-opus")) 3. 流式输出 from pydantic_ai import Agent from pydantic_ai.messages import Part streaming_agent = Agent( model="openai:gpt-4o", system_prompt="你是技术写作助手..." ) # 流式输出 async with streaming_agent.run_stream("写一篇关于 AGI 的短文") as result: async for message in result.stream_text(delta=True): print(message, end="", flush=True) # 流式结构化输出（2026 新特性） class ArticleOutline(BaseModel): title: str sections: list[str] estimated_words: int async with streaming_agent.run_stream( "生成文章大纲", output_type=ArticleOutline ) as result: # 部分结果流式返回 async for partial in result.stream_structured(): print(f"当前大纲: {partial.title} ({len(partial.sections)} 节)") final = await result.get_output() print(f"最终大纲: {final}") 4. 依赖注入 from pydantic_ai import Agent, RunContext from dataclasses import dataclass import httpx import asyncio @dataclass class AppDeps: http_client: httpx.AsyncClient db_pool: asyncpg.Pool cache: redis.Redis config: dict # 所有工具共享同一个依赖上下文 agent = Agent("openai:gpt-4o", deps_type=AppDeps) @agent.tool async def get_user_profile(ctx: RunContext[AppDeps], user_id: int) -> dict: # 从缓存获取 cached = await ctx.deps.cache.get(f"user:{user_id}") if cached: return json.loads(cached) # 从数据库获取 async with ctx.deps.db_pool.acquire() as conn: user = await conn.fetchrow( "SELECT * FROM users WHERE id = $1", user_id ) # 写入缓存 await ctx.deps.cache.setex(f"user:{user_id}", 3600, json.dumps(dict(user))) return dict(user) @agent.tool async def call_external_api(ctx: RunContext[AppDeps], url: str) -> dict: response = await ctx.deps.http_client.get(url) return response.json() # 运行时注入依赖 async def main(): async with httpx.AsyncClient() as http_client: db_pool = await asyncpg.create_pool(DATABASE_URL) cache = redis.Redis() deps = AppDeps( http_client=http_client, db_pool=db_pool, cache=cache, config={"max_results": 50} ) result = await agent.run("查询用户 123 的信息", deps=deps) print(result.output) 5. Agent 组合 from pydantic_ai import Agent # 子 Agent：各自有独立的输出类型 research_agent = Agent( model="openai:gpt-4o", output_type=ResearchReport, system_prompt="你是研究员..." ) writing_agent = Agent( model="anthropic:claude-4-sonnet", output_type=Article, system_prompt="你是技术写手..." ) review_agent = Agent( model="openai:gpt-4o", output_type=ReviewResult, system_prompt="你是审稿人..." ) # 主 Agent 协调子 Agent class PipelineResult(BaseModel): research: ResearchReport article: Article review: ReviewResult final_status: str pipeline_agent = Agent( model="openai:gpt-4o", output_type=PipelineResult, system_prompt="你是项目经理，协调多个子任务..." ) # 主 Agent 可以委派给子 Agent @pipeline_agent.tool async def run_research(ctx: RunContext, topic: str) -> ResearchReport: result = await research_agent.run(topic, deps=ctx.deps) return result.output @pipeline_agent.tool async def run_writing(ctx: RunContext, research: ResearchReport) -> Article: result = await writing_agent.run( f"基于以下研究撰写文章: {research.model_dump_json()}", deps=ctx.deps ) return result.output 评估与测试 from pydantic_ai.evals import TestCase, Evaluator # 定义测试用例 test_cases = [ TestCase( input="分析苹果公司 2026 年 Q1 财报", expected_output=ResearchReport( title="苹果 2026 Q1 财报分析", summary="...", key_findings=["营收增长 15%", "服务业务创新高"], confidence=0.9, sources=["https://investor.apple.com"], recommendation="buy" ) ), # 更多测试用例... ] # 评估器 class ResearchEvaluator(Evaluator): async def evaluate(self, output: ResearchReport, expected: ResearchReport) -> float: score = 0.0 # 检查结构化字段 if output.recommendation == expected.recommendation: score += 0.3 # 检查发现重叠度 overlap = len(set(output.key_findings) & set(expected.key_findings)) score += 0.4 * (overlap / max(len(expected.key_findings), 1)) # 检查置信度合理性 if abs(output.confidence - expected.confidence) < 0.2: score += 0.3 return score # 运行评估 results = await Evaluator.run( agent=research_agent, test_cases=test_cases, evaluator=ResearchEvaluator() ) print(f"平均得分: {results.avg_score}") print(f"通过率: {results.pass_rate}") 框架对比 特性 Pydantic AI LangChain CrewAI smolagents 类型安全 ⭐⭐⭐⭐⭐ ⭐⭐ ⭐⭐⭐ ⭐⭐ 结构化输出 编译时+运行时 运行时 运行时 无 学习曲线 中 高 低 低 IDE 支持 完整 部分 部分 部分 依赖注入 内置 无 无 无 流式输出 文本+结构化 文本 文本 文本 测试框架 内置 第三方 无 无 多模型 完整 完整 完整 HF 优先 性能基准 指标 Pydantic AI LangChain CrewAI Cold Start 0.2s 0.8s 0.5s 简单调用 1.1s 1.5s 1.8s 结构化输出 1.3s 1.8s 2.1s 内存占用 85MB 256MB 312MB Schema 验证开销 <50ms 100-200ms N/A 适用场景 最适合 企业级应用：类型安全是生产环境的刚需 API 后端：结构化输出直接映射 API 响应 数据管道：类型保证数据处理链路的可靠性 团队协作：类型系统作为 Agent 接口契约 不太适合 快速原型：类型定义增加了前期开发量 创意类任务：非结构化输出场景下类型约束是负担 复杂 Agent 图：不支持状态机编排 总结 Pydantic AI 在 2026 年的 Agent 框架竞争中找到了独特的定位：类型安全。这不是一个噱头——在实际的企业开发中，LLM 输出的不可预测性是最大的痛点之一。Pydantic AI 通过编译时类型检查、运行时 Schema 验证、依赖注入、内置测试框架，为 Agent 开发带来了真正的工程严谨性。 ...

极简主义的胜利 在 Agent 框架越来越复杂的 2026 年，HuggingFace 的 smolagents 反其道而行之——用不到 1000 行核心代码实现一个功能完备的 Agent 框架。这不是噱头，而是对"Agent 本质"的深刻理解：Agent 的核心就是 LLM + 工具调用，其余都是装饰。 设计哲学 smolagents 的三个设计原则： Code as Action：Agent 直接生成 Python 代码作为 Action，而非 JSON/函数调用 Minimal Abstraction：最少抽象层，开发者直接控制每个细节 HF Ecosystem First：与 HuggingFace 生态无缝集成 快速上手 安装 pip install smolagents 第一个 Agent from smolagents import CodeAgent, HfApiModel, tool # 定义工具 @tool def get_weather(city: str) -> str: """ 获取指定城市的天气信息。 Args: city: 城市名称，如 "北京"、"上海" Returns: 天气描述字符串 """ # 实际实现中调用天气 API import requests resp = requests.get(f"https://api.weather.com/v1/{city}") data = resp.json() return f"{city}：{data['condition']}，温度 {data['temp']}°C" @tool def calculate(expression: str) -> float: """ 安全地计算数学表达式。 Args: expression: 数学表达式，如 "2 + 3 * 4" Returns: 计算结果 """ import ast import operator ops = { ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, ast.Pow: operator.pow } node = ast.parse(expression, mode='eval').body if isinstance(node, ast.BinOp): return ops[type(node.op)]( calculate(ast.unparse(node.left)), calculate(ast.unparse(node.right)) ) return ast.literal_eval(node) # 创建 Agent agent = CodeAgent( model=HfApiModel("Qwen/Qwen2.5-72B-Instruct"), # 免费使用 HF Inference API tools=[get_weather, calculate], max_steps=10, verbosity_level=2 ) # 运行 result = agent.run( "北京和上海今天哪个温度更高？高多少度？" ) # Agent 会： # 1. 调用 get_weather("北京") # 2. 调用 get_weather("上海") # 3. 调用 calculate("上海温度 - 北京温度") # 4. 返回自然语言回答 Code Agent：代码即行动 smolagents 最独特的特性是 Code Agent 模式。与传统 Tool Calling 不同，Code Agent 直接生成可执行 Python 代码： ...

LangChain 生态版图 LangChain（工具库） ├── LangGraph（图式 Agent 编排） ├── LangSmith（可观测性） └── LangServe（API 部署） LangChain：工具库时代 定位 LangChain 是最早也是最流行的 LLM 应用框架，核心是"把 LLM 当作可组合的积木"。 核心概念 from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI # Chain = Prompt + LLM + Parser prompt = ChatPromptTemplate.from_template("解释 {topic} 给 5 岁小孩听") llm = ChatOpenAI(model="gpt-4o") parser = StrOutputParser() chain = prompt | llm | parser # 管道语法 result = chain.invoke({"topic": "量子计算"}) LangChain 的问题 # 问题 1：过度抽象 # 简单的 API 调用被包装了 5 层 chain = ( RunnablePassthrough.assign( context=lambda x: retriever.invoke(x["question"]) ) | prompt | llm.bind_tools(tools) | RunnableLambda(parse_tool_calls) | RunnableLambda(execute_tools) | llm | parser ) # 调试困难：报错信息深埋在调用栈中 # 问题 2：版本碎片化 # langchain / langchain-core / langchain-community 版本不同步 # 升级一个包可能破坏其他包 # 问题 3：性能开销 # 每层抽象都有 Python 对象创建开销 # 简单任务直接用 OpenAI SDK 更快 LangGraph：图式编排时代 定位 LangGraph 是 LangChain 团队的"自我革命"——从链式调用转向图式编排，专门为 Agent 设计。 ...