Magentic：用Python装饰器实现LLM结构化输出与函数调用

1. 项目概述Magentic让LLM成为你的Python函数如果你正在用Python构建AI应用大概率绕不开一个核心问题如何优雅、可靠地将大语言模型LLM的“非结构化”文本输出转换成你代码里可以直接使用的“结构化”数据是手动写一堆繁琐的正则表达式去解析还是忍受着脆弱的提示词工程祈祷模型每次都能返回正确的JSON格式今天要聊的magentic就是来解决这个痛点的。它不是一个庞大的AI应用框架而是一个极其精悍的Python库核心思想就一句话用装饰器把LLM调用变成普通的Python函数。想象一下你定义一个函数指定返回类型是一个Pydantic模型然后这个函数体里什么都不用写它的逻辑完全由一个LLM根据你给的提示词模板来生成并且返回值自动就是你定义的那个模型实例。这听起来是不是有点像魔法magentic通过prompt和chatprompt这两个核心装饰器把这种“声明式”的LLM编程变成了现实。它底层深度集成了Pydantic来做类型验证和序列化让你能用Python的类型系统str,int,list, 甚至自定义的类来“约束”和“塑造”LLM的输出从而构建出类型安全、易于测试和集成的AI功能模块。我最初是在一个需要从用户自由文本中提取标准化事件信息的项目里接触到它的。传统方法要么准确率堪忧要么代码又臭又长。用了magentic之后我只需要定义一个数据模型和一个带占位符的提示词剩下的解析、校验、格式化工作全交给库和LLM了代码量减少了70%可维护性却大大提升。它特别适合那些需要将自然语言转换为结构化数据的场景比如智能客服意图分类、文档信息抽取、代码生成后的解析或者作为更复杂智能体Agent系统中的可靠工具调用单元。2. 核心设计哲学当类型提示遇见提示词工程magentic的设计非常巧妙它站在了两个巨人的肩膀上Python强大的类型注解生态系统和现代LLM的函数调用能力。它的核心工作流程可以概括为你声明函数签名包括参数类型和返回类型magentic负责将类型信息与你的提示词模板结合生成LLM能理解的、带有严格输出格式要求的指令最后将LLM的回复自动解析并实例化为指定的Python对象。2.1 类型即契约Pydantic的核心作用为什么是Pydantic因为Pydantic在Python社区已经是数据验证和序列化的事实标准。magentic利用Pydantic的模型定义来为LLM生成一个清晰、机器可读的输出模式Schema。当你用prompt装饰一个返回类型为Superhero一个Pydantic模型的函数时magentic在幕后会做两件事模式生成它将Superhero模型的JSON Schema提取出来。这个Schema精确描述了输出应该包含哪些字段name,age,power,enemies每个字段的类型是什么str,int,list[str]以及是否有默认值、是否可选等约束。指令合成它将这个Schema与你的文本提示词融合形成给LLM的最终指令。对于支持“函数调用”或“结构化输出”的模型如GPT-4o、Claude 3.5这个Schema会以工具Tool或JSON模式的形式直接发送对于较老的模型magentic会在提示词中附加严格的文本格式要求例如“你必须以如下JSON格式回复...”。这样一来LLM不再是“自由发挥”而是在一个明确的框架内生成内容。返回的结果会通过Pydantic自动进行验证和解析如果LLM的输出不符合模式比如age字段返回了非数字Pydantic会抛出验证错误而不是让你的程序带着脏数据继续运行。注意这种强类型约束是一把双刃剑。它极大地提高了输出的可靠性和程序的安全性但也可能因为模式过于严格而增加LLM的生成难度。对于复杂或开放性的生成任务可能需要设计更宽松的Schema或结合LLM重试机制。2.2 装饰器无缝的开发者体验prompt和chatprompt装饰器是magentic的API设计精髓。它们让LLM调用看起来和感觉上都像一个普通的同步/异步函数极大地降低了认知负担。prompt用于单轮对话的简单提示。你提供一个字符串模板用花括号{}包裹参数名。调用函数时参数值被注入模板发送给LLM返回值就是解析后的结构化数据。from magentic import prompt from pydantic import BaseModel class Recipe(BaseModel): name: str ingredients: list[str] steps: list[str] prompt(根据以下食材生成一个菜谱{ingredients_list}) def generate_recipe(ingredients_list: str) - Recipe: ... # 调用就像调用普通函数一样 recipe generate_recipe(西红柿鸡蛋盐糖) print(recipe.name) # 输出西红柿炒蛋 print(recipe.steps[0]) # 输出1. 西红柿洗净切块...chatprompt用于多轮对话或小样本Few-shot提示。你传入一个消息列表SystemMessage,UserMessage,AssistantMessage同样支持模板化。这在需要提供上下文或示例时非常有用能更好地引导LLM的行为。from magentic import chatprompt, SystemMessage, UserMessage, AssistantMessage chatprompt( SystemMessage(你是一个严谨的科技文章翻译助手擅长将中文技术术语准确翻译成英文。), UserMessage(请翻译神经网络), AssistantMessage(neural network), UserMessage(请翻译{term}), ) def translate_tech_term(term: str) - str: ... print(translate_tech_term(反向传播)) # 输出backpropagation这种设计使得包含LLM逻辑的代码模块化程度极高每个被装饰的函数都是一个独立的、可测试的单元。你可以像组合乐高积木一样将这些函数组合成更复杂的业务流程。3. 核心功能深度解析与实战要点了解了设计理念后我们深入看看magentic的几个杀手级功能以及在实际使用中需要注意的细节。3.1 结构化输出从文本到对象的自动化流水线结构化输出是magentic的立身之本。除了返回基本的Python类型str,int,bool,list,dict和Pydantic模型它还支持更复杂的嵌套结构和联合类型。实战示例处理不确定的输出有时LLM的答案可能属于几种预定义类型之一。比如用户可能询问一个概念的定义也可能要求一个代码示例。from typing import Union from pydantic import BaseModel class Definition(BaseModel): concept: str explanation: str class CodeExample(BaseModel): language: str code: str description: str prompt(根据用户请求提供信息{query}) def answer_query(query: str) - Union[Definition, CodeExample]: ... result answer_query(Python的列表推导式是什么) if isinstance(result, Definition): print(f概念解释{result.explanation}) else: print(f代码示例{result.language}\n{result.code})这里Union[Definition, CodeExample]告诉magentic和LLM返回结果可以是两种模式中的任意一种。LLM会根据对query的理解选择最合适的模式进行生成。避坑指南模式设计的艺术保持简洁LLM对复杂、深度嵌套的JSON Schema理解能力会下降。尽量保持数据模型扁平字段名语义清晰。提供示例在chatprompt中使用AssistantMessage提供输出示例是引导LLM遵循复杂格式的最有效方法。善用描述利用Pydantic的Field为字段添加描述这会被包含在生成的Schema中帮助LLM理解字段含义。from pydantic import BaseModel, Field class Product(BaseModel): name: str Field(description产品的完整名称) price: float Field(gt0, description产品价格必须大于0) in_stock: bool Field(description当前是否有库存)3.2 函数调用与智能体Agentic工作流magentic真正强大的地方在于它能让LLM不仅生成文本还能“决定”调用哪个真实的Python函数并将结果纳入后续的思考。这是构建智能体Agent系统的基石。FunctionCall对象延迟执行的指令当prompt装饰的函数中传入了functions参数列表LLM可能会返回一个FunctionCall对象。这个对象封装了“要调用的函数”和“调用参数”但它本身并不立即执行。from magentic import prompt, FunctionCall def get_weather(city: str) - dict: # 模拟调用天气API return {city: city, temp: 22, condition: 晴朗} def get_stock_price(symbol: str) - float: # 模拟调用股票API return 150.5 prompt( 回答用户问题必要时可以查询信息。问题{question}, functions[get_weather, get_stock_price] ) def answer_with_tools(question: str) - Union[str, FunctionCall]: ... response answer_with_tools(北京今天天气怎么样) print(type(response)) # 输出class magentic.function_call.FunctionCall print(response) # 输出FunctionCall(function get_weather at ..., 北京) # 此时get_weather函数尚未被调用 weather_info response() # 手动执行函数调用 print(weather_info) # 输出{city: 北京, temp: 22, condition: 晴朗}这种设计将“决策”由LLM做出和“执行”由你的代码控制分离开让你有机会在函数真正执行前插入日志、权限检查、参数修正等逻辑。prompt_chain自动化任务链对于需要连续多次工具调用的复杂任务手动处理每个FunctionCall很繁琐。prompt_chain装饰器可以自动执行这个过程它运行LLM如果返回FunctionCall则执行它将结果作为新的上下文再次发送给LLM如此循环直到LLM返回一个非FunctionCall的最终答案。from magentic import prompt_chain def search_web(query: str) - str: # 模拟网络搜索 return f关于{query}的搜索结果摘要... def calculate(expression: str) - str: # 模拟计算 return str(eval(expression)) # 注意生产环境请勿使用eval prompt_chain( 你是一个助手。请分步骤解决用户问题{problem}, functions[search_web, calculate] ) def solve_problem(problem: str) - str: ... # LLM可能会先调用search_web用结果再调用calculate最后总结答案 final_answer solve_problem(珠穆朗玛峰的高度乘以3是多少米) print(final_answer)这实现了一个简单的ReActReasoning Acting模式智能体。magentic使得构建这种多步推理和行动的工作流变得异常简单。重要安全提示将函数暴露给LLM调用存在风险。务必严格审查functions列表中包含的函数避免LLM调用到具有破坏性如os.remove或高权限的函数。建议为Agent专门设计一套安全的工具函数。3.3 流式处理与异步并发提升响应体验与性能处理大量文本或批量任务时流式输出和异步并发是提升效率的关键。流式输出Streaming对于生成长篇内容等待LLM完全生成再返回会卡住整个程序。magentic支持返回StreamedStr或AsyncStreamedStr允许你逐块chunk处理输出。from magentic import prompt, StreamedStr prompt(写一篇关于人工智能历史的短文) def generate_essay() - StreamedStr: ... stream generate_essay() full_text for chunk in stream: # 可以实时显示到前端或进行初步处理 print(chunk, end, flushTrue) full_text chunk # 循环结束后stream内容也生成完毕对于结构化输出的流式处理可以使用Iterable[T]作为返回类型注解这样每个对象一生成就能立刻被处理非常适合批量生成场景。异步并发Asyncio当需要同时查询多个独立内容时使用异步可以避免不必要的等待。import asyncio from magentic import prompt prompt(总结一下{company}的主要业务) async def summarize_company(company: str) - str: ... async def main(): companies [Apple, Microsoft, Google] # 创建多个异步任务它们会并发执行 tasks [summarize_company(company) for company in companies] summaries await asyncio.gather(*tasks) # 同时等待所有任务完成 for company, summary in zip(companies, summaries): print(f{company}: {summary[:100]}...) asyncio.run(main())在我的一个竞品分析项目中使用异步并发将获取数十家公司信息的时间从线性增长的数分钟缩短到了几乎恒定的十几秒效率提升了一个数量级。4. 多后端配置与模型管理实战magentic支持多种LLM后端这让你可以根据成本、性能、功能需求灵活切换模型避免了被单一供应商锁定的风险。4.1 主流后端配置详解OpenAI默认这是最常用且功能最全的后端。除了官方API还可以通过配置base_url连接任何兼容OpenAI API的服务器比如本地部署的Ollama。from magentic import OpenaiChatModel, prompt # 使用环境变量 OPENAI_API_KEY model_gpt4 OpenaiChatModel(gpt-4o) # 连接本地Ollama需先运行 ollama pull llama3.2 model_llama OpenaiChatModel(llama3.2, base_urlhttp://localhost:11434/v1/) # 指定使用模型和参数 prompt(讲个笑话, modelmodel_gpt4) def tell_joke() - str: ...Anthropic Claude安装magentic[anthropic]后即可使用。Claude模型在长文本和逻辑推理上表现优异。from magentic.chat_model.anthropic_chat_model import AnthropicChatModel model_claude AnthropicChatModel(claude-3-5-sonnet-latest) # 后续在装饰器中指定 modelmodel_claude 即可通过LiteLLM访问多模型LiteLLM是一个统一的LLM调用层支持上百种模型。通过magentic[litellm]后端你可以用一套代码调用OpenAI、Anthropic、Cohere、Replicate甚至众多开源模型。from magentic.chat_model.litellm_chat_model import LitellmChatModel # 使用OpenAI模型 model1 LitellmChatModel(gpt-4o) # 使用Anthropic模型需配置相应环境变量 model2 LitellmChatModel(claude-3-5-sonnet-latest) # 使用本地模型如通过Ollama model3 LitellmChatModel(ollama/llama3.2)注意并非所有模型都支持函数调用结构化输出和流式传输。在使用非主流模型前务必查阅LiteLLM文档确认其功能支持情况。4.2 灵活的配置策略magentic提供了三层配置机制让你能精细控制每个函数使用哪个模型。全局默认配置通过环境变量设置如MAGENTIC_BACKEND、MAGENTIC_OPENAI_MODEL。这是最简单的方式适合项目整体使用同一模型。上下文管理器Context Manager使用with语句临时切换某个代码块内所有prompt函数的默认模型。from magentic import OpenaiChatModel, prompt prompt(任务A) def task_a() - str: ... prompt(任务B) def task_b() - str: ... # 默认使用环境变量配置的模型 task_a() with OpenaiChatModel(gpt-4o-mini, temperature0.2): # 这个块内未显式指定model的函数都会用gpt-4o-mini task_a() task_b()函数级显式配置在prompt装饰器中直接传入model参数。优先级最高用于对特定任务使用特殊模型。from magentic import OpenaiChatModel, AnthropicChatModel, prompt cheap_model OpenaiChatModel(gpt-3.5-turbo) strong_model AnthropicChatModel(claude-3-5-sonnet-latest) prompt(处理简单分类, modelcheap_model) def simple_task(text: str) - str: ... prompt(进行复杂推理, modelstrong_model) def complex_task(text: str) - str: ...配置经验谈开发与生产分离在开发环境可以使用便宜的快速模型如gpt-3.5-turbo通过上下文管理器或环境变量来切换。在生产环境则明确为每个关键函数指定稳定、性能达标的模型。成本监控不同模型价格差异巨大。对于简单的文本补全或格式转换完全没必要使用最顶级的模型。利用magentic的灵活配置可以轻松实现成本优化。降级策略在你的代码中可以尝试先用强模型如果遇到速率限制或错误可以捕获异常并自动切换到备用模型提高系统的鲁棒性。5. 高级特性与生产环境最佳实践掌握了基础用法和配置后我们来看看那些能让你的应用更健壮、更可观测的高级特性。5.1 LLM辅助重试LLM-Assisted Retries即使有严格的输出模式LLM偶尔也会“抽风”返回格式错误或内容不符合要求的答案。手动编写重试逻辑很麻烦。magentic内置了重试机制其独特之处在于“LLM辅助”当解析失败时它会将错误信息例如Pydantic验证错误和之前的对话历史一起反馈给LLM要求它根据错误修正输出。from magentic import prompt from pydantic import BaseModel, Field from magentic.retry import retry class ValidatedOutput(BaseModel): score: int Field(ge1, le10, description1-10的整数评分) reason: str # 使用retry装饰器包裹prompt装饰器 retry(max_attempts3) # 最多重试3次 prompt(为{product}打分并说明理由) def rate_product(product: str) - ValidatedOutput: ... try: result rate_product(某款手机) except Exception as e: print(f重试多次后仍失败{e})在这个例子中如果LLM第一次返回了score: 15超出了1-10的范围Pydantic会验证失败。retry机制会捕获这个错误并将“score字段值必须小于等于10”这样的信息反馈给LLM让它重新生成。这比简单的“重新问一遍”要有效得多。5.2 可观测性与日志集成在生产环境中知道每个LLM调用发生了什么至关重要提示词是什么收到了什么回复消耗了多少Token花了多长时间magentic通过OpenTelemetry提供了开箱即用的可观测性支持并能与Pydantic的亲儿子——Logfire深度集成。基础日志记录设置环境变量MAGENTIC_LOG_LEVELDEBUG你就能在控制台看到详细的请求和响应日志包括格式化后的提示词和解析后的结果。这对于调试提示词和输出模式非常有用。与Logfire集成推荐Logfire是Pydantic官方推出的可观测性平台。集成后你可以在一个漂亮的仪表板里查看所有LLM调用的追踪信息。import logfire from magentic import prompt # 初始化Logfire需要注册获取token logfire.configure() prompt(分析句子情感{sentence}) def analyze_sentiment(sentence: str) - str: ... # 自动被追踪在Logfire控制台可以看到这次调用的详细信息。 with logfire.span(处理用户反馈): sentiment analyze_sentiment(这个产品太棒了)在Logfire的界面上你可以清晰地看到函数调用链、每个LLM请求的输入输出、耗时和Token用量。这对于监控成本、分析性能瓶颈、审计AI行为是不可或缺的工具。5.3 类型检查与IDE友好性由于被prompt装饰的函数没有实际的函数体像mypy、pyright这样的类型检查器会报错例如Function is missing a return statement。magentic提供了几种解决方案全局屏蔽在mypy配置中禁用empty-body错误码。这是最彻底的方法但可能会掩盖其他真正的空函数体错误。# pyproject.toml [tool.mypy] disable_error_code [empty-body]函数体占位使用...省略号或raise NotImplementedError作为函数体。...可能无法满足所有检查器raise则更明确。prompt(生成名字) def generate_name() - str: raise NotImplementedError # 明确表示此函数由LLM实现行内忽略在每个函数后添加类型忽略注释。这样可以为函数保留文档字符串Docstring对代码可读性有帮助。prompt(生成名字) def generate_name() - str: # type: ignore[empty-body] 使用LLM生成一个随机名字。我个人推荐第二种raise NotImplementedError或第三种类型忽略文档字符串方法。它们在保持代码清晰的同时也向阅读代码的同事明确传达了“这是一个魔法函数”的意图。好的IDE如VS Code with Pylance在安装了magentic类型存根后能提供很好的参数和返回类型提示。6. 常见问题、排查技巧与性能优化在实际项目中使用magentic你肯定会遇到一些坑。下面是我总结的一些常见问题和解决方案。6.1 错误处理与调试问题一LLM始终返回格式错误的JSON。可能原因1模式太复杂。简化你的Pydantic模型减少嵌套使用更明确的字段描述。可能原因2提示词指令不清晰。在chatprompt中使用AssistantMessage提供一个完美的输出示例。对于prompt可以在提示词开头加入强指令如“你必须严格按照给定的JSON格式输出不要包含任何其他解释文字。”解决方案启用retry装饰器。LLM辅助重试能解决大部分格式问题。问题二函数调用FunctionCall没有被触发。可能原因1提供的函数描述不清。LLM通过函数名和文档字符串Docstring来理解函数功能。确保你的工具函数有清晰、简洁的文档字符串。可能原因2提示词没有引导LLM使用工具。在提示词中明确告诉LLM“你可以使用以下工具”或“如果需要查询信息请调用相应的函数”。检查步骤将prompt的return类型暂时改为str看看LLM生成的原始文本是什么。它可能已经决定要调用函数但因为格式问题没有被正确解析。问题三异步函数调用报错。牢记如果prompt装饰的函数中传入了异步函数作为工具functions参数那么当LLM返回FunctionCall并执行它时你必须使用await。import asyncio from magentic import prompt, FunctionCall async def async_search(query: str) - str: await asyncio.sleep(1) return f搜索结果{query} prompt(查询{question}, functions[async_search]) async def ask_with_async_tool(question: str) - Union[str, FunctionCall]: ... async def main(): response await ask_with_async_tool(什么是Python) if isinstance(response, FunctionCall): # 注意调用异步FunctionCall必须await result await response() print(result) asyncio.run(main())6.2 性能优化指南批量处理如果需要处理大量相似项目不要用循环一个个调用prompt函数。考虑设计一个提示词让LLM直接返回一个List[YourModel]。或者使用Iterable返回类型进行流式处理边生成边处理。缓存结果对于确定性较高的查询例如“将以下关键词翻译成英文”可以考虑对prompt函数的结果进行缓存。可以使用functools.lru_cache但要注意函数的参数必须是可哈希的。更复杂的场景可能需要基于向量数据库的语义缓存。模型选型延迟敏感选择推理速度快的模型如gpt-3.5-turbo或claude-3-haiku。启用流式响应也能让用户感知延迟更低。成本敏感对于内部任务或对质量要求不高的场景大胆使用小模型。通过temperature参数较低值如0.2来提高输出的确定性弥补模型能力的不足。质量敏感复杂推理、代码生成、创意写作等任务还是需要gpt-4o、claude-3-5-sonnet这类顶级模型。超时与重试网络请求和LLM生成都可能超时。务必为你的HTTP客户端如httpx设置合理的超时时间并配合retry装饰器实现指数退避重试以应对暂时的网络波动或API限流。6.3 架构设计心得经过多个项目的实践我总结出几点使用magentic构建应用的架构建议分层设计将AI功能层与业务逻辑层分离。用magentic构建纯粹的、可复用的“AI工具函数”例如extract_invoice_data(document_text: str) - InvoiceModel。业务层代码像调用普通库一样调用这些函数而不需要关心LLM的细节。这使得替换模型或调整提示词变得非常容易。提示词管理不要把提示词模板硬编码在装饰器里。对于复杂的提示词可以考虑将其存放在外部文件如YAML、JSON或数据库中在运行时动态加载。这便于进行A/B测试和迭代优化。验证与清洗虽然Pydantic提供了类型验证但对于从LLM来的数据额外的业务逻辑验证往往是必要的。例如即使age字段是整数你可能还需要检查它是否在合理范围内0-150。在prompt函数外再包裹一层业务验证逻辑。限流与熔断如果你构建的是面向大量用户的服务必须考虑LLM API的调用限制和成本。在调用magentic函数的前置层实现限流Rate Limiting和熔断Circuit Breaker机制防止一个异常提示词或突发流量导致账单爆炸或服务被禁。magentic这个库的精妙之处在于它用极简的API抽象了LLM集成中最复杂、最易出错的部分——结构化输出和函数调用。它让你能更专注于业务逻辑和提示词工程本身而不是繁琐的API调用和JSON解析。无论是快速构建一个原型还是开发一个需要稳定运行的AI功能模块它都是一个值得深入工具箱的利器。