Pydantic Graph：复杂AI应用的图形化编排

【免费下载链接】pydantic-ai Agent Framework / shim to use Pydantic with LLMs 项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai

Pydantic Graph作为Pydantic AI框架的核心组件，提供了一种基于类型提示的图形化代理编排解决方案，让开发者能够以声明式的方式构建复杂的AI工作流。文章详细介绍了其核心架构组件（节点、图形运行上下文、终止节点）、类型安全的图形定义方法、可视化与调试功能、状态持久化机制、异步执行模型以及错误处理与重试机制，展示了如何通过图形化编排方式清晰地表达复杂的业务逻辑。

图形化代理编排的基本概念

在现代AI应用开发中，随着业务逻辑的复杂性不断增加，传统的线性控制流往往难以应对多步骤、多分支的智能决策流程。Pydantic Graph作为Pydantic AI框架的核心组件，提供了一种基于类型提示的图形化代理编排解决方案，让开发者能够以声明式的方式构建复杂的AI工作流。

核心架构组件

Pydantic Graph的架构围绕几个关键抽象构建，每个组件都承担着特定的职责：

节点（Nodes） - 执行单元

节点是图形中的基本执行单元，每个节点都是一个继承自BaseNode的数据类。节点通过run方法定义其业务逻辑，并通过返回类型声明其出边连接。

from dataclasses import dataclass
from pydantic_graph import BaseNode, GraphRunContext, End

@dataclass
class QuestionNode(BaseNode[QuestionState]):
    """提问节点，负责生成问题"""
    
    async def run(self, ctx: GraphRunContext[QuestionState]) -> AnswerNode:
        # 使用AI代理生成问题
        result = await ask_agent.run('生成一个简单问题')
        ctx.state.question = result.output
        return AnswerNode(result.output)

图形运行上下文（GraphRunContext）

GraphRunContext提供了节点执行时的上下文环境，包含状态管理、依赖注入和追踪信息：

属性	类型	描述
state	StateT	图形运行的状态对象
deps	DepsT	依赖注入的对象
traceparent	str	分布式追踪标识

终止节点（End）

End类型用于标识图形执行的终止条件，可以携带任意类型的返回数据：

@dataclass
class EvaluationNode(BaseNode[QuestionState, None, str]):
    """评估节点，可能终止图形执行"""
    
    answer: str
    
    async def run(self, ctx: GraphRunContext[QuestionState]) -> End[str] | RetryNode:
        if self.answer == "正确":
            return End("回答正确！")
        else:
            return RetryNode("请重试")

类型安全的图形定义

Pydantic Graph充分利用Python的类型系统，通过泛型参数确保图形定义的类型安全：

# 定义图形状态
@dataclass
class ConversationState:
    messages: list[str] = field(default_factory=list)
    current_topic: str = ""

# 创建类型安全的图形
conversation_graph = Graph[
    ConversationState,  # 状态类型
    DatabaseDeps,       # 依赖类型  
    str                 # 返回类型
](nodes=[StartNode, ProcessNode, EndNode])

可视化与调试

Pydantic Graph内置Mermaid图表生成功能，可以自动将图形定义转换为可视化图表：

# 生成Mermaid图表代码
mermaid_code = conversation_graph.mermaid_code(start_node=StartNode)
print(mermaid_code)

状态持久化机制

图形执行支持状态持久化，允许中断后恢复执行：

from pydantic_graph.persistence.file import FileStatePersistence

# 创建文件持久化器
persistence = FileStatePersistence(Path('graph_state.json'))
persistence.set_graph_types(conversation_graph)

# 使用持久化运行图形
async with conversation_graph.iter(
    start_node=StartNode(), 
    state=initial_state,
    persistence=persistence
) as run:
    # 图形执行过程会自动持久化状态
    ...

异步执行模型

Pydantic Graph完全基于异步IO设计，支持高效的并发执行：

方法	描述	使用场景
run()	异步执行整个图形	服务端应用
run_sync()	同步包装的异步执行	脚本和CLI工具
iter()	迭代式执行	交互式应用

错误处理与重试机制

图形化编排天然支持复杂的错误处理和重试逻辑：

通过这种图形化的编排方式，开发者可以清晰地表达复杂的业务逻辑，同时保持代码的可维护性和可扩展性。Pydantic Graph的类型安全特性和可视化支持使得复杂AI应用的开发和调试变得更加高效和直观。

类型提示定义图形结构的方法

Pydantic Graph采用了基于Python类型提示的创新方法来定义图形结构，这种方法将图形编排的复杂性转化为类型安全的代码结构。通过利用Python的类型系统，开发者可以以声明式的方式定义节点之间的连接关系，同时获得静态类型检查的好处。

核心类型系统组件

Pydantic Graph的类型系统建立在几个核心泛型类型变量之上：

StateT = TypeVar('StateT', default=None)  # 状态类型变量
RunEndT = TypeVar('RunEndT', covariant=True, default=None)  # 运行结束类型变量
NodeRunEndT = TypeVar('NodeRunEndT', covariant=True, default=Never)  # 节点运行结束类型变量
DepsT = TypeVar('DepsT', default=None, contravariant=True)  # 依赖类型变量

这些类型变量为图形结构提供了强大的类型安全保证，确保状态、依赖和返回值在整个图形执行过程中保持一致。

节点定义与类型提示

每个节点都是BaseNode的子类，通过run方法的返回类型注解来定义出边关系：

from dataclasses import dataclass
from pydantic_graph import BaseNode, End, GraphRunContext

@dataclass
class ProcessData(BaseNode[DataState, None, str]):
    input_data: str

    async def run(self, ctx: GraphRunContext[DataState]) -> ValidateData | End[str]:
        if self.input_data == "invalid":
            return End("Invalid data processed")
        return ValidateData(self.input_data.upper())

在这个例子中，ProcessData节点的run方法返回类型ValidateData | End[str]明确声明了两种可能的出边：

• 返回ValidateData实例时，执行流转移到验证节点
• 返回End[str]实例时，图形执行结束并返回字符串结果

类型驱动的边缘定义

Pydantic Graph通过分析run方法的返回类型注解来自动推断图形结构：

这种类型驱动的边缘定义机制确保了：

1. 编译时验证：类型检查器可以验证所有引用的节点类都存在
2. 运行时安全：图形执行时只能跳转到类型声明中指定的节点
3. 自文档化：类型注解本身就描述了图形的拓扑结构

复杂类型组合与联合类型

对于更复杂的图形逻辑，可以使用Python的联合类型和泛型组合：

from typing import Annotated, Union
from pydantic_graph import Edge

@dataclass  
class DecisionNode(BaseNode[AppState, None, Result]):
    async def run(self, ctx: GraphRunContext[AppState]) -> Union[
        Annotated[ProcessA, Edge(label="选择A路径")],
        Annotated[ProcessB, Edge(label="选择B路径")], 
        Annotated[End[Result], Edge(label="直接完成")]
    ]:
        if ctx.state.condition_a:
            return ProcessA()
        elif ctx.state.condition_b:
            return ProcessB() 
        else:
            return End(Result(status="completed"))

这种模式允许单个节点根据运行时状态动态选择多个出边路径，同时保持类型安全。

状态类型的一致性保证

Pydantic Graph通过泛型参数确保状态类型在整个图形中保持一致：

@dataclass
class UserState:
    user_id: int
    profile: dict
    session_data: str

# 所有节点必须使用相同的StateT类型参数
class FetchProfile(BaseNode[UserState]):
    async def run(self, ctx: GraphRunContext[UserState]) -> ProcessProfile:
        # 可以安全地访问ctx.state.user_id等字段
        pass

class ProcessProfile(BaseNode[UserState, None, str]):
    async def run(self, ctx: GraphRunContext[UserState]) -> End[str]:
        # 类型安全的状态访问
        return End(f"Processed {ctx.state.user_id}")

这种设计防止了状态类型不匹配的错误，确保所有节点操作的是相同类型的状态对象。

依赖注入的类型安全

依赖类型也通过泛型系统得到保证：

@dataclass
class AppDependencies:
    db_connection: Database
    cache: RedisCache
    config: Config

class DatabaseOperation(BaseNode[None, AppDependencies, str]):
    async def run(self, ctx: GraphRunContext[None, AppDependencies]) -> End[str]:
        # 可以安全地使用依赖项
        result = await ctx.deps.db_connection.query("SELECT * FROM users")
        return End(str(result))

类型提示的运行时解析

Pydantic Graph在运行时解析类型提示来构建图形结构：

# 在Graph初始化时解析所有节点的类型提示
graph = Graph(nodes=[ProcessData, ValidateData, FinalizeData])

# 内部会调用每个节点的get_node_def方法
node_def = ProcessData.get_node_def(local_ns=None)
# 返回NodeDef对象包含出边信息

这个过程提取run方法的返回类型注解，分析联合类型，识别所有可能的出边节点，并构建完整的图形拓扑结构。

错误检测与验证

类型提示系统还提供了强大的错误检测能力：

# 如果节点返回类型引用了未包含在图形中的节点类
# Graph初始化时会抛出GraphSetupError
try:
    graph = Graph(nodes=[ProcessData])  # ValidateData未包含
except GraphSetupError as e:
    print(f"图形设置错误: {e.message}")

这种编译时错误检测大大减少了运行时错误的可能性。

与传统图形定义方法的对比

传统的图形定义通常使用配置文件或DSL：

# 传统YAML配置方式
nodes:
  - id: process_data
    type: function
    next:
      - validate_data
      - end_node
  - id: validate_data
    type: function
    next: finalize_data

而Pydantic Graph的类型提示方法提供了：

• 更好的IDE支持：代码补全、类型检查和重构功能
• 更强的类型安全：编译时错误检测
• 更自然的开发体验：使用熟悉的Python语法
• 更好的可维护性：图形结构与业务逻辑代码在一起

高级类型模式

对于复杂的应用场景，Pydantic Graph支持多种高级类型模式：

条件类型出边：

from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from .other_module import SpecialNode

class ConditionalNode(BaseNode[StateT]):
    async def run(self, ctx: GraphRunContext[StateT]) -> "SpecialNode | End[None]":
        # 运行时延迟导入
        from .other_module import SpecialNode
        return SpecialNode() if condition else End(None)

动态类型解析：

class DynamicNode(BaseNode[StateT]):
    async def run(self, ctx: GraphRunContext[StateT]) -> BaseNode:
        # 返回BaseNode允许跳转到任何节点
        return get_next_node_based_on_state(ctx.state)

类型提示的最佳实践

在使用类型提示定义图形结构时，建议遵循以下最佳实践：

1. 明确的类型注解：为所有节点提供完整的类型参数
2. 一致的命名约定：使用有意义的节点类名
3. 模块化设计：将相关节点组织在同一模块中
4. 类型别名：为复杂类型定义类型别名以提高可读性
5. 文档字符串：为节点和边缘添加详细的文档

# 使用类型别名提高可读性
NodeResult = Union[ProcessNext, ValidateInput, End[FinalResult]]

class BusinessNode(BaseNode[BusinessState, None, FinalResult]):
    """处理核心业务逻辑的节点"""
    
    async def run(self, ctx: GraphRunContext[BusinessState]) -> NodeResult:
        """根据业务状态决定下一步操作"""
        if ctx.state.should_validate:
            return ValidateInput()
        elif ctx.state.is_complete:
            return End(FinalResult(success=True))
        else:
            return ProcessNext()

通过这种类型提示驱动的方法，Pydantic Graph将图形编排的复杂性转化为类型安全的代码结构，为构建复杂的AI应用提供了强大而可靠的基础设施。

复杂业务流程的图形化实现

在现代AI应用开发中，复杂业务流程往往涉及多个步骤、条件分支和状态转换。Pydantic Graph通过其强大的图形化编排能力，为开发者提供了一种优雅的方式来建模和执行这些复杂流程。

状态驱动的业务流程建模

Pydantic Graph的核心优势在于其状态驱动的设计理念。通过定义清晰的状态模型，可以精确控制业务流程的每一步执行。以下是一个典型的电商订单处理流程示例：

from dataclasses import dataclass
from enum import Enum
from typing import Optional
from pydantic_graph import BaseNode, End, Graph, GraphRunContext

class OrderStatus(Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    SHIPPED = "shipped"
    DELIVERED = "delivered"
    CANCELLED = "cancelled"

@dataclass
class OrderState:
    order_id: str
    status: OrderStatus = OrderStatus.PENDING
    payment_confirmed: bool = False
    inventory_reserved: bool = False
    shipping_label_created: bool = False
    tracking_number: Optional[str] = None
    customer_notified: bool = False

@dataclass
class ValidateOrder(BaseNode[OrderState, None, str]):
    order_id: str

    async def run(self, ctx: GraphRunContext[OrderState]) -> ProcessPayment | End[str]:
        # 验证订单有效性
        if not self._is_order_valid(self.order_id):
            return End("订单验证失败")
        
        ctx.state.order_id = self.order_id
        ctx.state.status = OrderStatus.PROCESSING
        return ProcessPayment()

    def _is_order_valid(self, order_id: str) -> bool:
        # 实际的订单验证逻辑
        return len(order_id) > 0

@dataclass
class ProcessPayment(BaseNode[OrderState]):
    async def run(self, ctx: GraphRunContext[OrderState]) -> ReserveInventory | End[str]:
        try:
            # 模拟支付处理
            payment_success = self._process_payment(ctx.state.order_id)
            if payment_success:
                ctx.state.payment_confirmed = True
                return ReserveInventory()
            else:
                ctx.state.status = OrderStatus.CANCELLED
                return End("支付处理失败")
        except Exception as e:
            return End(f"支付处理异常: {str(e)}")

    def _process_payment(self, order_id: str) -> bool:
        # 实际的支付处理逻辑
        return True

@dataclass
class ReserveInventory(BaseNode[OrderState]):
    async def run(self, ctx: GraphRunContext[OrderState]) -> CreateShippingLabel | End[str]:
        # 库存预留逻辑
        inventory_available = self._check_inventory(ctx.state.order_id)
        if inventory_available:
            ctx.state.inventory_reserved = True
            return CreateShippingLabel()
        else:
            ctx.state.status = OrderStatus.CANCELLED
            return End("库存不足")

    def _check_inventory(self, order_id: str) -> bool:
        # 实际的库存检查逻辑
        return True

@dataclass
class CreateShippingLabel(BaseNode[OrderState]):
    async def run(self, ctx: GraphRunContext[OrderState]) -> NotifyCustomer | End[str]:
        # 创建物流标签
        tracking_info = self._generate_shipping_label(ctx.state.order_id)
        if tracking_info:
            ctx.state.shipping_label_created = True
            ctx.state.tracking_number = tracking_info
            ctx.state.status = OrderStatus.SHIPPED
            return NotifyCustomer()
        else:
            return End("物流标签创建失败")

    def _generate_shipping_label(self, order_id: str) -> str:
        # 实际的物流标签生成逻辑
        return f"TRK{order_id}123456"

@dataclass
class NotifyCustomer(BaseNode[OrderState, None, str]):
    async def run(self, ctx: GraphRunContext[OrderState]) -> End[str]:
        # 通知客户
        notification_sent = self._send_notification(
            ctx.state.order_id, 
            ctx.state.tracking_number
        )
        if notification_sent:
            ctx.state.customer_notified = True
            ctx.state.status = OrderStatus.DELIVERED
            return End("订单处理完成")
        else:
            return End("客户通知发送失败")

    def _send_notification(self, order_id: str, tracking_number: str) -> bool:
        # 实际的通知发送逻辑
        print(f"订单 {order_id} 已发货，追踪号: {tracking_number}")
        return True

# 创建订单处理图
order_processing_graph = Graph(nodes=[
    ValidateOrder, ProcessPayment, ReserveInventory, 
    CreateShippingLabel, NotifyCustomer
])

可视化业务流程执行

通过Mermaid图表，可以清晰地展示整个业务流程的执行路径：

条件分支与错误处理

复杂业务流程往往需要处理多种条件和异常情况。Pydantic Graph通过类型安全的返回注解来实现条件分支：

@dataclass
class ComplexDecisionNode(BaseNode[OrderState]):
    scenario: str

    async def run(self, ctx: GraphRunContext[OrderState]) -> NodeA | NodeB | NodeC | End[str]:
        if self.scenario == "case_a":
            return NodeA()
        elif self.scenario == "case_b":
            return NodeB()
        elif self.scenario == "case_c":
            return NodeC()
        else:
            return End("无效的场景选择")

状态持久化与恢复

对于长时间运行的业务流程，状态持久化至关重要。Pydantic Graph提供了完善的状态管理机制：

from pydantic_graph.persistence.file import FileStatePersistence
from pathlib import Path

async def run_order_processing(order_id: str):
    state = OrderState(order_id=order_id)
    persistence = FileStatePersistence(Path(f"order_{order_id}.json"))
    
    # 设置图形类型以便正确序列化
    persistence.set_graph_types(order_processing_graph)
    
    result = await order_processing_graph.run(
        ValidateOrder(order_id), 
        state=state, 
        persistence=persistence
    )
    
    return result.output

业务流程监控与调试

通过集成监控工具，可以实时跟踪业务流程的执行状态：

async def monitor_business_process():
    state = OrderState(order_id="ORD123")
    
    async with order_processing_graph.iter(
        ValidateOrder("ORD123"), 
        state=state
    ) as run:
        async for node in run:
            print(f"当前节点: {node.__class__.__name__}")
            print(f"订单状态: {state.status.value}")
            print(f"支付确认: {state.payment_confirmed}")
            print(f"库存预留: {state.inventory_reserved}")
            print("---")

多条件业务流程示例

以下展示一个更复杂的多条件审批流程：

@dataclass
class ApprovalWorkflowState:
    request_id: str
    current_approver: str
    approval_level: int = 1
    comments: list[str] = field(default_factory=list)
    approved: bool = False
    rejected: bool = False

@dataclass
class RouteToApprover(BaseNode[ApprovalWorkflowState]):
    async def run(self, ctx: GraphRunContext[ApprovalWorkflowState]) -> GetApproval | End[str]:
        approver = self._get_next_approver(ctx.state.approval_level)
        if approver:
            ctx.state.current_approver = approver
            return GetApproval()
        else:
            return End("无可用审批人")

@dataclass
class GetApproval(BaseNode[ApprovalWorkflowState]):
    async def run(self, ctx: GraphRunContext[ApprovalWorkflowState]) -> RouteToApprover | FinalizeApproval | End[str]:
        decision = self._get_approval_decision(ctx.state.current_approver)
        
        if decision == "approve":
            if ctx.state.approval_level >= 3:  # 最终审批
                ctx.state.approved = True
                return FinalizeApproval()
            else:
                ctx.state.approval_level += 1
                return RouteToApprover()
        elif decision == "reject":
            ctx.state.rejected = True
            return End("审批被拒绝")
        else:
            return End("审批决策无效")

这个复杂的审批流程展示了如何通过图形化编排处理多级审批、条件路由和状态转换，确保了业务流程的清晰性和可维护性。

图形执行引擎与性能优化

Pydantic Graph 的执行引擎采用异步架构设计，通过精心优化的执行流程和状态管理机制，为复杂AI应用提供高性能的图形化编排能力。执行引擎的核心设计理念是在保证类型安全的同时，最大化执行效率。

异步执行架构

Pydantic Graph 的执行引擎完全基于异步编程模型构建，充分利用 Python 的 asyncio 框架。每个节点的 run 方法都是异步的，允许在节点执行过程中进行I/O操作而不会阻塞整个图形执行流程。

@dataclass
class DataProcessingNode(BaseNode[ProcessingState]):
    async def run(self, ctx: GraphRunContext) -> NextNode | End[ProcessedData]:
        # 异步数据获取
        raw_data = await fetch_data_async(ctx.state.data_source)
        
        # 异步数据处理
        processed_data = await process_data_async(raw_data)
        
        if processed_data.is_valid():
            return End(processed_data)
        else:
            return NextNode(ctx.state.with_retry_count())

这种异步设计使得图形执行引擎能够高效处理以下场景：

• 网络请求和API调用
• 数据库查询操作
• 文件I/O操作
• 外部服务集成

状态管理优化

执行引擎采用智能的状态快照机制，通过 BaseStatePersistence 接口实现状态的持久化和恢复。系统提供多种状态存储实现：

存储类型	适用场景	性能特点
SimpleStatePersistence	内存存储，开发测试	最高性能，无序列化开销
FileStatePersistence	文件持久化，生产环境	中等性能，支持状态恢复
自定义实现	特定需求	可根据业务定制优化

状态快照的工作流程如下：

执行流程优化

执行引擎通过以下技术手段优化性能：

1. 延迟类型推断 图形在首次执行时自动推断状态类型和结果类型，避免不必要的类型检查开销：

def _infer_name(self, function_frame: types.FrameType | None) -> None:
    """智能推断图形名称，减少配置开销"""
    if self.name is None and function_frame is not None:
        self.name = _utils.get_caller_name(function_frame)

2. 智能迭代控制 提供灵活的迭代接口，支持手动控制和自动执行两种模式：

# 自动迭代模式
async with graph.iter(start_node, state=initial_state) as run:
    async for node in run:
        # 实时监控节点执行
        monitor_node_performance(node)

# 手动控制模式
async with graph.iter(start_node, state=initial_state) as run:
    while True:
        next_node = await run.next()
        if isinstance(next_node, End):
            break
        # 自定义控制逻辑
        if should_skip_node(next_node):
            continue

3. 执行上下文优化 GraphRunContext 提供高效的上下文管理，减少状态传递开销：

@dataclass
class OptimizedGraphRunContext(Generic[StateT, DepsT]):
    """优化后的执行上下文"""
    state: StateT
    deps: DepsT
    traceparent: str | None
    
    @cached_property
    def expensive_resource(self):
        """延迟加载昂贵资源"""
        return initialize_expensive_resource()

性能监控与诊断

执行引擎集成 Pydantic Logfire 进行实时性能监控：

# 自动性能检测
graph = Graph(nodes=my_nodes, auto_instrument=True)

# 手动性能span
async with logfire_api.span('critical_path_execution') as span:
    result = await graph.run(critical_node, state=state)
    span.set_attributes({
        'execution_time': result.duration,
        'nodes_executed': len(result.history)
    })

性能监控指标包括：

• 节点执行时间分布
• 状态序列化/反序列化开销
• 异步等待时间分析
• 内存使用情况

并发执行优化

对于需要并行执行的场景，执行引擎支持并发节点执行：

缓存策略

执行引擎实现多级缓存策略提升性能：

1. 节点定义缓存

# 节点定义缓存优化
node_defs = {}
for node in nodes:
    node_id = node.get_node_id()
    if node_id not in self.node_defs:
        # 缓存节点定义，避免重复解析
        self.node_defs[node_id] = node.get_node_def(parent_namespace)

2. 类型推断缓存

@cached_property
def inferred_types(self) -> tuple[type[StateT], type[RunEndT]]:
    """缓存类型推断结果"""
    if not hasattr(self, '_inferred_types_cache'):
        self._inferred_types_cache = self._infer_types()
    return self._inferred_types_cache

3. 序列化优化 使用 Pydantic 的高效序列化机制，减少状态序列化开销：

def _prep_state(self, state: StateT) -> StateT:
    """状态序列化前预处理"""
    if hasattr(state, 'optimize_for_serialization'):
        return state.optimize_for_serialization()
    return state

执行引擎配置调优

通过合理的配置参数优化执行性能：

# 性能优化配置示例
optimized_graph = Graph(
    nodes=production_nodes,
    name='high_performance_graph',
    auto_instrument=True,  # 开启性能监控
    # 显式指定类型，避免推断开销
    state_type=OptimizedState,
    run_end_type=ProcessedResult
)

配置参数对性能的影响：

参数	默认值	性能影响	建议
auto_instrument	True	轻微开销	生产环境建议开启
显式类型指定	自动推断	减少推断时间	复杂图形建议显式指定
持久化策略	内存	影响恢复性能	根据可靠性需求选择

通过上述优化措施，Pydantic Graph 的执行引擎能够在保证类型安全和功能完整性的同时，为复杂AI应用提供高性能的图形化编排能力。执行引擎的设计充分考虑了生产环境的需求，提供了丰富的性能调优选项和监控能力。

总结

Pydantic Graph通过其强大的图形化编排能力，为复杂AI应用提供了高效、可靠的执行引擎。文章全面探讨了其异步执行架构、状态管理优化、执行流程优化、性能监控与诊断、并发执行优化以及多级缓存策略等关键技术。执行引擎的设计充分考虑了生产环境的需求，在保证类型安全和功能完整性的同时，通过精心优化的执行流程和状态管理机制，为开发者提供了丰富的性能调优选项和监控能力，使得构建和维护复杂AI应用变得更加高效和直观。

【免费下载链接】pydantic-ai Agent Framework / shim to use Pydantic with LLMs 项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai