基于Pydantic的AI任务编排框架：构建类型安全的智能体工作流

1. 项目概述一个基于Pydantic的AI任务编排框架最近在GitHub上看到一个挺有意思的项目叫vstorm-co/pydantic-ai-todo。光看名字你可能会觉得这又是一个“用AI写TODO应用”的玩具项目。但实际深入探究后我发现它的野心远不止于此。这本质上是一个利用Pydantic模型来定义、编排和执行AI驱动任务的框架。简单来说它试图解决一个核心痛点如何将大语言模型LLM的能力像调用函数一样稳定、可靠、结构化地嵌入到你的应用程序流程中。想象一下这个场景你需要开发一个客服机器人它不仅要理解用户意图还要能查询数据库、调用外部API、并根据复杂规则生成回复。传统的做法可能是写一堆胶水代码用if-else串联提示词Prompt然后费力地解析LLM返回的非结构化文本。这个过程充满了不确定性调试起来也异常痛苦。而pydantic-ai-todo这类框架的思路是让你用Pydantic数据模型来“声明”一个任务输入是什么需要调用哪些工具函数输出应该是什么结构。框架则负责与LLM对话、管理状态、处理错误最终给你一个符合你定义模型的、类型安全的Python对象。这个项目名中的“todo”可能有点误导它更像是一个“待办事项”或“任务清单”的隐喻指的是AI需要完成的一系列动作。其核心价值在于它提供了一种声明式的、类型安全的方式来构建AI智能体Agent或工作流。对于任何正在尝试将LLM集成到生产系统中的开发者来说这类工具代表着一种更工程化、更可控的范式转变。接下来我将拆解这个项目的核心设计、实现原理并分享如何基于类似思想构建你自己的AI任务编排系统。2. 核心设计理念与架构拆解2.1 为什么是Pydantic类型安全与契约优先项目的基石是Pydantic这是一个在Python生态中用于数据验证和设置管理的库凭借FastAPI而广为人知。在AI应用开发中Pydantic的价值被进一步放大。LLM的本质是一个概率模型其输出是开放式的文本。而我们的应用程序需要的是结构化的、确定性的数据。Pydantic正好充当了这两者之间的“契约”和“转换器”。使用Pydantic模型来定义AI任务的输入和输出相当于为LLM的“自由发挥”划定了一个明确的边界。你告诉LLM“你必须返回一个符合这个格式的JSON对象。” 这不仅提高了输出的可靠性还让后续的代码处理变得非常简单因为你可以直接获得一个具有属性提示和自动验证的Python对象。这种“契约优先”的设计将不可靠的文本解析问题转变为了可靠的数据验证问题是工程化AI应用的关键一步。2.2 任务编排的核心组件Agent、Task与Tool虽然原项目pydantic-ai-todo的具体实现需要查看其源码但这类框架通常围绕几个核心抽象进行构建。理解这些抽象就能理解整个框架的运作模式。Agent智能体这是任务的执行者或协调者。它持有LLM的配置如API密钥、模型选择、对话历史管理逻辑以及可用的工具集。你可以把它想象成一个拥有“大脑”LLM和“双手”工具的虚拟员工。Task任务这是需要完成的具体工作单元。一个任务通常由以下几个部分定义目标描述用自然语言告诉AI要做什么。输入模型一个Pydantic模型定义任务所需的结构化输入。输出模型一个Pydantic模型定义任务应返回的结构化输出。工具集任务执行过程中可以调用的函数列表。Tool工具这是AI可以调用的具体函数。一个工具通常包含函数本身、其功能描述供LLM理解、以及输入参数的Pydantic模型。例如“查询天气”工具、“发送邮件”工具、“计算数据”工具等。框架负责将工具的描述格式化进提示词并在LLM决定调用时安全地执行对应的Python函数。这种架构的优势在于关注点分离。业务开发者只需要关注1定义清晰的数据模型2实现纯粹的工具函数3用自然语言描述任务。而复杂的提示工程、函数调用调度、状态管理和错误重试则由框架底层处理。2.3 与主流框架的对比与选型思考目前市场上类似的框架不少例如 LangChain、LlamaIndex 的早期版本以及新兴的microsoft/autogen、langchain-ai/langgraph等。pydantic-ai-todo这类项目通常定位更轻量、更聚焦。vs LangChainLangChain功能全面但抽象层次多学习曲线陡峭有时被称为“胶水链的胶水”。pydantic-ai-todo的思路可能更倾向于“少即是多”深度集成Pydantic让开发者用自己熟悉的Python类型和模型来工作减少新概念的学习成本。vs 直接使用OpenAI Function Calling直接使用各大模型提供商提供的函数调用API是最直接的方式。但你需要自己管理对话状态、处理工具调用循环、组装提示词。pydantic-ai-todo这类框架的价值在于它封装了这些样板代码提供了更高层、更声明式的API并能更容易地切换底层LLM提供商。选择的关键在于你的需求复杂度。如果你的任务相对简单、固定直接使用官方的SDK和函数调用可能就够了。但如果你需要构建包含多步骤推理、复杂工具调用的智能体或者希望有一套统一的模式来管理越来越多的AI功能那么采用一个基于Pydantic的轻量级编排框架会显著提升开发效率和代码可维护性。3. 从零开始实现一个简易版核心引擎为了彻底理解其原理我们不妨动手实现一个极度简化的“Pydantic AI任务引擎”。这个实现将忽略流式输出、复杂历史管理等高级特性聚焦于最核心的“模型定义-提示生成-函数调用-结果解析”循环。3.1 定义基础模型消息、工具与结果首先我们需要定义几个核心的Pydantic模型。from pydantic import BaseModel, Field from typing import Any, Callable, Dict, List, Optional, Type, Union import inspect import json # 1. 消息模型用于构建对话历史 class Message(BaseModel): role: str # system, user, assistant, tool content: str # 2. 工具调用请求模型模拟OpenAI的function call class ToolCall(BaseModel): id: str type: str function function: Dict[str, Any] # 包含 name 和 arguments # 3. 工具调用结果模型 class ToolResult(BaseModel): tool_call_id: str output: str # 4. 任务结果模型 class TaskResult(BaseModel): final_output: Any # 符合输出模型的数据 messages: List[Message] # 完整的对话历史 usage: Dict[str, int] # 可选的token使用情况3.2 实现工具装饰器与注册机制接下来我们需要一个机制将普通的Python函数“包装”成AI可用的工具。这里我们创建一个装饰器。class Tool: 工具描述类存储函数及其元信息 def __init__(self, func: Callable, description: str): self.func func self.name func.__name__ self.description description # 使用Pydantic模型来自动生成参数schema self.args_model self._create_args_model(func) def _create_args_model(self, func): 从函数签名动态生成Pydantic参数模型 sig inspect.signature(func) fields {} for param_name, param in sig.parameters.items(): if param_name self: continue # 简化处理假设所有参数都是字符串实际应用需更复杂的类型映射 fields[param_name] (str, Field(..., descriptionf参数 {param_name})) # 动态创建模型类 model_name f{self.name}Args return type(model_name, (BaseModel,), {__annotations__: fields}) def to_openai_schema(self): 生成OpenAI函数调用兼容的schema schema self.args_model.schema() return { type: function, function: { name: self.name, description: self.description, parameters: { type: object, properties: schema.get(properties, {}), required: schema.get(required, []) } } } def tool(description: str): 装饰器将普通函数注册为工具 def decorator(func: Callable): return Tool(func, description) return decorator # 示例工具 tool(description获取指定城市的当前天气) def get_weather(city: str) - str: # 这里应该是真实的API调用返回模拟数据 return f{city}的天气是晴朗25摄氏度。 tool(description计算两个数字的和) def add(a: int, b: int) - str: return str(int(a) int(b))3.3 构建任务运行器Agent的核心逻辑这是引擎最复杂的部分它需要管理对话、调用LLM、处理工具调用并循环直到任务完成。class SimpleAgent: def __init__(self, llm_client, system_prompt: str 你是一个有帮助的助手。): llm_client: 一个模拟的LLM客户端需要有 chat.completions.create 方法 并能处理 tools 和 tool_choice 参数。 self.llm llm_client self.system_prompt system_prompt self.tools: Dict[str, Tool] {} self.messages: List[Message] [Message(rolesystem, contentsystem_prompt)] def register_tool(self, tool: Tool): 注册工具到智能体 self.tools[tool.name] tool def _run_tool(self, tool_call: ToolCall) - ToolResult: 执行工具调用 func_name tool_call.function[name] if func_name not in self.tools: return ToolResult( tool_call_idtool_call.id, outputf错误工具 {func_name} 未找到。 ) tool_obj self.tools[func_name] try: # 解析参数 args json.loads(tool_call.function[arguments]) # 使用Pydantic模型验证参数 validated_args tool_obj.args_model(**args) # 调用函数 result tool_obj.func(**validated_args.dict()) return ToolResult( tool_call_idtool_call.id, outputstr(result) ) except Exception as e: return ToolResult( tool_call_idtool_call.id, outputf工具执行错误{str(e)} ) async def run_task(self, user_input: str, output_model: Type[BaseModel], max_turns: int 5) - TaskResult: 运行一个任务直到获得符合输出模型的结果或达到最大轮数 self.messages.append(Message(roleuser, contentuser_input)) turns 0 while turns max_turns: turns 1 # 1. 准备调用LLM openai_tools [t.to_openai_schema() for t in self.tools.values()] # 注意这里需要将self.messages转换成LLM客户端所需的格式 # 为简化我们假设self.messages可以直接使用 # 2. 调用LLM模拟 # 在真实场景中这里会是response await self.llm.chat.completions.create(...) # 我们模拟一个响应 llm_response self._mock_llm_call(openai_tools) # 3. 处理响应 self.messages.append(Message(roleassistant, contentllm_response.get(content, ))) # 4. 检查是否有工具调用 tool_calls llm_response.get(tool_calls, []) if tool_calls: tool_messages [] for tc in tool_calls: # 执行工具 result self._run_tool(ToolCall(**tc)) # 将结果添加到消息历史 tool_msg Message(roletool, contentresult.output, tool_call_idresult.tool_call_id) self.messages.append(tool_msg) tool_messages.append(result) # 本轮有工具调用继续循环让LLM基于工具结果进行下一步思考 continue else: # 5. 没有工具调用尝试解析最终输出 final_content llm_response.get(content, ) try: # 这里是一个关键点我们需要让LLM的输出符合output_model的JSON格式 # 在实际框架中会在系统提示词中严格要求LLM以指定JSON格式回复 # 我们模拟解析 parsed_data json.loads(final_content) # 假设LLM返回了JSON字符串 validated_output output_model(**parsed_data) return TaskResult( final_outputvalidated_output, messagesself.messages.copy(), usage{prompt_tokens: 100, completion_tokens: 50} # 模拟数据 ) except (json.JSONDecodeError, Exception): # 解析失败可能LLM还在“思考”或者格式不对。 # 在实际框架中这里可能会注入一条系统消息要求LLM格式化输出。 # 为简化我们直接返回错误或继续循环。 self.messages.append(Message( rolesystem, contentf请将你的回答严格格式化为以下JSON结构{output_model.schema_json()} )) continue raise RuntimeError(f任务在 {max_turns} 轮后未完成。) def _mock_llm_call(self, tools: List[Dict]): 模拟LLM调用。在实际项目中这里会替换为真实的OpenAI、Anthropic等API调用。 # 这是一个极其简化的模拟仅用于演示逻辑。 # 它根据最后一条用户消息决定是调用工具还是直接回复。 last_msg self.messages[-1].content.lower() if 天气 in last_msg and tools: # 模拟LLM决定调用天气工具 return { content: , tool_calls: [{ id: call_123, type: function, function: { name: get_weather, arguments: json.dumps({city: 北京}) # 简单提取城市名 } }] } elif 加起来 in last_msg or 求和 in last_msg: return { content: , tool_calls: [{ id: call_456, type: function, function: { name: add, arguments: json.dumps({a: 5, b: 3}) } }] } else: # 模拟直接回复并尝试格式化为JSON class MockOutput(BaseModel): answer: str return { content: json.dumps({answer: 这是一个模拟的最终回答。}) }3.4 定义任务并运行一个完整的示例现在让我们用上面的简易引擎来定义一个具体的任务并运行它。# 1. 定义任务的输出模型 class WeatherAnswer(BaseModel): city: str weather_condition: str temperature: str summary: str # 2. 模拟一个LLM客户端在实际中使用openai、anthropic等库 class MockLLMClient: async def chat(self): return self async def create(self, **kwargs): # 这里应该调用真实的API我们返回一个模拟响应 return {choices: [{message: SimpleAgent._mock_llm_call([])}]} # 3. 创建智能体注册工具 async def main(): agent SimpleAgent(llm_clientMockLLMClient(), system_prompt你是一个天气和计算助手。) agent.register_tool(get_weather) agent.register_tool(add) # 4. 运行任务 try: result await agent.run_task( user_input请问北京天气怎么样, output_modelWeatherAnswer ) print(任务成功) print(最终输出:, result.final_output) print(输出类型:, type(result.final_output)) # 应该是WeatherAnswer实例 print(对话轮数:, len(result.messages)) except RuntimeError as e: print(任务失败:, e) except Exception as e: print(发生错误:, e) # 运行在异步环境中 import asyncio asyncio.run(main())这个简易实现清晰地展示了核心流程注册工具 - 构建提示 - LLM决策 - 执行工具 - 验证输出。真正的pydantic-ai-todo或类似工业级框架会在这些基础环节上增加大量功能如更智能的参数提取、流式响应、对话历史管理、复杂任务分解Chain of Thought、以及对不同LLM提供商API的适配层。4. 实战构建一个智能数据查询助手为了更贴近实际应用我们设计一个更复杂的场景一个智能数据查询助手。它能够理解用户用自然语言提出的数据分析请求自动调用相应的数据查询和计算工具并最终生成一个结构化的报告。4.1 定义领域模型与工具假设我们有一个简单的销售数据库。首先定义数据模型和工具。from datetime import date from enum import Enum from pydantic import BaseModel, Field from typing import List # 数据模型 class ProductCategory(Enum): ELECTRONICS 电子产品 BOOKS 图书 CLOTHING 服装 class SalesRecord(BaseModel): date: date product_name: str category: ProductCategory amount: float quantity: int # 模拟一个内存“数据库” mock_database [ SalesRecord(datedate(2024, 5, 1), product_name智能手机, categoryProductCategory.ELECTRONICS, amount5999.0, quantity2), SalesRecord(datedate(2024, 5, 1), product_namePython编程书, categoryProductCategory.BOOKS, amount89.0, quantity10), SalesRecord(datedate(2024, 5, 2), product_nameT恤衫, categoryProductCategory.CLOTHING, amount199.0, quantity25), SalesRecord(datedate(2024, 5, 2), product_name笔记本电脑, categoryProductCategory.ELECTRONICS, amount8999.0, quantity1), SalesRecord(datedate(2024, 5, 3), product_name小说合集, categoryProductCategory.BOOKS, amount120.0, quantity8), ] # 查询工具 tool(description根据条件查询销售记录。可以按日期、产品类别筛选。) def query_sales(start_date: str None, end_date: str None, category: str None) - str: 查询销售记录。日期格式应为 YYYY-MM-DD。 filtered mock_database if start_date: start date.fromisoformat(start_date) filtered [r for r in filtered if r.date start] if end_date: end date.fromisoformat(end_date) filtered [r for r in filtered if r.date end] if category: try: cat_enum ProductCategory(category) filtered [r for r in filtered if r.category cat_enum] except ValueError: return f错误未知类别 {category}。可选值{, .join([e.value for e in ProductCategory])} # 返回格式化的字符串结果便于LLM阅读 if not filtered: return 未找到符合条件的销售记录。 result_lines [] for r in filtered: result_lines.append(f{r.date.isoformat()} | {r.product_name} | {r.category.value} | 销售额:{r.amount}元 | 数量:{r.quantity}) return \n.join(result_lines) tool(description对一组销售记录进行统计分析计算总销售额、平均销售额等。) def analyze_sales(sales_data_summary: str) - str: 分析销售数据文本摘要。 输入应是由query_sales工具返回的格式化文本。 # 这是一个简化的分析。在实际中你可能需要更复杂的解析。 lines sales_data_summary.split(\n) total_amount 0.0 total_quantity 0 count 0 for line in lines: if | not in line: continue parts line.split(|) if len(parts) 5: # 提取销售额和数量 amount_str parts[3].replace(销售额:, ).replace(元, ).strip() qty_str parts[4].replace(数量:, ).strip() try: total_amount float(amount_str) total_quantity int(qty_str) count 1 except ValueError: continue if count 0: return 无有效数据可供分析。 avg_amount total_amount / count if count else 0 return f分析结果总销售额 {total_amount:.2f} 元总销量 {total_quantity} 件平均每单销售额 {avg_amount:.2f} 元共计 {count} 条记录。 # 输出报告模型 class SalesAnalysisReport(BaseModel): question: str Field(description用户原始问题) data_summary: str Field(description查询到的数据摘要) analysis_result: str Field(description统计分析结果) insight: str Field(description基于分析的简要洞察或建议)4.2 配置智能体与系统提示词系统提示词是引导LLM行为的关键。一个好的提示词能显著提升任务完成的准确率。system_prompt 你是一个专业的数据分析助手。你的职责是理解用户关于销售数据的问题并通过调用合适的工具来获取和分析数据最终生成一份结构化的报告。 你必须遵循以下规则 1. 首先理解用户的问题确定是否需要查询数据。如果需要调用 query_sales 工具。 2. 获得原始数据后思考是否需要进一步分析如计算总和、平均值。如果需要调用 analyze_sales 工具并将上一步查询到的数据摘要传递给它。 3. 在获得所有必要信息后你必须将最终答案严格按照 SalesAnalysisReport 模型的JSON格式输出。不要输出任何其他内容。 4. 如果你无法理解问题或工具返回错误请在报告的 insight 字段中说明情况。 工具使用说明 - query_sales: 用于查询销售记录。你可以按日期范围start_date, end_date或产品类别category筛选。类别必须是电子产品、图书、服装 中的一个。 - analyze_sales: 用于分析销售数据。其输入必须是 query_sales 返回的文本格式。 请一步步思考并只在需要时调用工具。 4.3 执行复杂查询任务现在我们可以用之前构建的SimpleAgent需要稍作增强以支持更复杂的提示词和工具链来执行任务。假设我们有一个增强版的Agent。# 假设我们使用一个更完善的框架如 langchain 或 原项目 pydantic-ai-todo的伪代码 async def run_sales_analysis(question: str): # 初始化Agent并传入强大的系统提示词 agent EnhancedAgent(llm_client, system_promptsystem_prompt) agent.register_tool(query_sales) agent.register_tool(analyze_sales) # 运行任务指定输出模型 result await agent.run_task( user_inputquestion, output_modelSalesAnalysisReport, max_turns10 ) return result.final_output # 模拟用户提问 questions [ 帮我看看五月份前三天的销售情况并做个总结。, 分析一下图书类产品的销售表现。, 2024年5月2日哪类产品卖得最好, ] for q in questions: print(f\n用户问题: {q}) try: # 注意这里需要异步执行为演示方便我们直接调用假设已完成的函数 # report await run_sales_analysis(q) # print(report.json(indent2)) print((模拟输出) 报告已生成包含问题、数据摘要、分析结果和洞察。) except Exception as e: print(f任务执行出错: {e})在这个场景中LLM需要理解“五月份前三天”对应start_date2024-05-01和end_date2024-05-03先调用query_sales然后将返回的文本数据传递给analyze_sales进行计算最后将原始问题、数据、分析结果和生成的洞察一起填充到SalesAnalysisReport模型中。整个过程通过清晰的模型定义和工具描述被自动化、结构化地完成。5. 生产环境部署的考量与最佳实践将这样一个AI任务框架用于生产环境远不止是写好模型和工具那么简单。以下是几个关键的考量点和实践建议。5.1 性能优化与成本控制LLM API调用是主要的延迟和成本来源。缓存策略对确定性查询如相同的用户输入和系统提示的结果进行缓存。可以使用内存缓存如functools.lru_cache或分布式缓存如Redis并为缓存键设置合理的TTL。令牌使用优化精心设计系统提示词和工具描述在清晰的前提下力求简洁。对于长对话考虑使用“摘要”技术将过长的历史消息总结成一段短文而不是全部发送。一些框架支持“消息修剪”功能只保留最近N条或最近N个令牌的消息。异步与并发确保你的任务运行器是异步的async/await这样可以高效地处理多个并发请求尤其是在等待LLM API响应或执行I/O密集型工具时。模型选型不一定总是使用最强大、最昂贵的模型如GPT-4。对于简单的工具调用和格式化工单性能较好的小模型如GPT-3.5-Turbo、Claude Haiku可能更具性价比。可以设计一个路由层根据任务复杂度动态选择模型。5.2 错误处理与鲁棒性增强AI应用的不确定性要求我们有更健壮的错误处理机制。工具调用异常捕获每个工具函数内部都应该有完善的try...except返回清晰的错误信息给LLM而不是让整个进程崩溃。LLM输出格式重试即使有严格的提示LLM偶尔也可能不返回有效的JSON。框架应具备重试逻辑当解析失败时可以自动向对话中注入一条修正指令如“请严格按JSON格式回复”并重新调用LLM最多重试N次。超时与断路为LLM API调用和每个工具执行设置超时。对于频繁失败的工具或API考虑实现断路器模式暂时禁用该组件以防止级联故障。验证与清洗Pydantic模型本身提供了强大的验证。但针对LLM输入可能需要额外的清洗逻辑比如处理用户输入中的特殊字符、过滤敏感信息等。5.3 可观测性与调试调试一个多步骤的AI智能体比调试普通代码困难得多。结构化日志记录每一轮对话的完整消息历史、工具调用详情输入、输出、耗时、LLM响应原始内容、最终解析结果等。使用结构化日志如JSON格式便于后续查询和分析。追踪与关联ID为每个用户会话或任务请求生成一个唯一的追踪IDtrace_id并将其贯穿所有日志、工具调用和API请求。这样可以在出现问题时快速定位完整的执行链路。可视化工具考虑开发或集成一个简单的UI能够回放某个trace_id的完整执行过程查看LLM的“思考”链和工具调用序列。这对于理解AI为何做出某个决策至关重要。成本与用量监控监控每个任务、每个用户的令牌消耗和API调用次数设置告警阈值便于成本分析和优化。5.4 安全与权限控制当AI可以调用工具尤其是写数据库、发邮件、操作外部系统时安全是重中之重。工具执行沙箱对于执行不可信代码如用户自定义工具的场景应考虑在沙箱环境中运行。基于角色的工具访问控制RBAC不是所有用户都能调用所有工具。可以在Agent层面或工具注册时绑定权限标签。在执行任务前验证当前用户是否有权调用所需的工具。输入验证与净化再次强调所有从LLM传递给工具的参数都必须经过严格的Pydantic模型验证和业务逻辑验证防止注入攻击。审计日志所有工具调用特别是修改性操作必须记录详尽的审计日志包括操作者、时间、参数和结果。6. 常见问题与排查技巧实录在实际开发和运维基于此类框架的应用时你一定会遇到各种问题。以下是一些典型问题及其排查思路。6.1 LLM不按预期调用工具症状你定义了工具但LLM总是直接回答而不调用工具。排查检查工具描述工具的描述是否清晰、无歧义LLM依赖描述来决定是否以及何时调用工具。确保描述准确说明了工具的功能和适用场景。检查系统提示词系统提示词是否明确指示LLM在特定情况下要使用工具例如“当你需要获取实时信息时请使用查询工具。”检查tool_choice参数在调用LLM API时是否将tool_choice参数设置为了auto或required如果设置为noneLLM将不会调用工具。查看LLM的“思考”过程如果使用的模型支持如GPT-4在调试时开启logprobs或查看其内部推理过程如果平台提供看它是否考虑了工具但最终否决了。解决优化工具描述和系统提示词。有时在用户问题中更明确地暗示使用工具也有帮助例如用户问“查询一下北京的天气”比“北京天气怎么样”更能触发工具调用。6.2 工具调用参数解析错误症状LLM决定调用工具但传递的参数格式错误、类型不匹配或缺少必填字段导致Pydantic验证失败。排查查看原始参数在工具被调用前打印或记录下LLM生成的argumentsJSON字符串。检查其是否符合你定义的模型。检查参数schema确保你提供给LLM的工具参数schema由to_openai_schema生成是准确的。特别是枚举类型、日期格式等复杂类型在schema中是否有正确的描述和约束LLM的“幻觉”LLM可能会“捏造”一些不存在的参数。确保你的工具schema只包含必要的参数并在描述中说明每个参数的用途。解决在Pydantic模型中使用更严格的字段类型和验证器。在工具描述中用更直白的语言说明每个参数的要求例如“start_date字符串格式必须为YYYY-MM-DD例如2024-05-01”。实现一个“参数修正”步骤当第一次调用因参数错误失败时将错误信息反馈给LLM要求它重新生成正确的参数。许多框架内置了这种重试逻辑。6.3 输出格式不符合Pydantic模型症状任务流程看似正常但最后一步无法将LLM的回复解析成指定的输出模型。排查查看LLM的最终回复检查LLM在最后一轮返回的纯文本内容是什么。它是否是一个完整的JSON字符串还是夹杂了其他解释性文字检查系统提示词是否在最后阶段明确、强硬地要求LLM“只输出JSON不要有任何其他文字”这是一个非常关键的指令。使用JSON模式JSON Mode如果使用的LLM API支持如OpenAI的response_format{ type: json_object }务必开启。这会强制LLM输出合法的JSON。解决在系统提示词的最后部分用非常醒目的方式强调输出格式要求。例如“你的最终输出必须是且只能是一个JSON对象其结构完全符合以下JSON Schema{{schema}}。”在代码中实现一个后处理层如果解析失败尝试用正则表达式从回复中提取可能的JSON块。考虑使用“输出解析器”Output Parser模式这是LangChain等框架中的一个标准组件专门处理LLM输出到结构化数据的转换。6.4 任务陷入无限循环或轮数过多症状AI在几个工具间来回调用或者反复询问同一个问题无法达成最终状态。排查检查停止条件你的Agent是否有明确的停止条件例如在获得符合输出模型的结果后立即停止还是设定了最大轮数max_turns分析对话历史查看完整的消息流。AI是否陷入了“困惑-提问-再困惑”的循环这可能是因为工具返回的信息不足或模糊。工具设计的副作用是否某些工具调用会改变状态导致后续条件永远无法满足解决务必设置合理的max_turns如10-15轮作为安全网。设计更智能的系统提示词让AI学会在拥有足够信息时做出决断并结束对话。引入“超时”或“用户确认”机制。如果循环超过一定轮数可以主动中断或向用户请求明确指示。6.5 处理LLM的“幻觉”与不确定性这是所有AI应用的根本挑战。策略一提供充足的上下文确保提供给LLM的工具描述、系统指令和用户问题背景是清晰、无歧义的。信息越充分幻觉概率越低。策略二让AI“三思而后行”在提示词中鼓励AI进行链式思考Chain-of-Thought例如要求它“首先分析用户的问题列出需要的信息。然后决定调用哪个工具。最后根据工具结果给出答案。” 这可以通过在消息中设置roleassistant的“思考”内容来实现。策略三后验验证对于关键操作如发送邮件、修改数据库不要完全依赖AI的决策。可以设计一个模式让AI生成操作草案然后由另一个简单的规则引擎或人工确认流程进行二次校验。策略四持续评估与迭代建立测试用例集定期运行监控任务成功率、输出质量。根据失败案例不断优化你的提示词、工具设计和输出模型。通过深入理解pydantic-ai-todo这类项目背后的设计哲学并亲手实践其核心原理你就能掌握一种强大的、用于构建可靠AI应用的工程化方法。它不仅仅是封装API调用更是通过类型、契约和声明式编程在不确定的AI世界与确定的软件系统之间架起了一座坚固的桥梁。