Python函数式LLM编程：magentic框架实现类型安全的大模型集成

1. 项目概述：当Python函数遇见LLM的魔法

如果你和我一样，日常工作中既需要编写严谨的Python代码来处理结构化数据，又时不时要调用大语言模型（LLM）来处理一些非结构化的、充满不确定性的文本任务，那你一定体会过那种“精神分裂”般的开发体验。一边是def、return、try-except的确定性世界，另一边是openai.ChatCompletion.create()返回的、需要你小心翼翼去解析的choices[0].message.content。代码里充斥着字符串拼接、JSON解析和错误处理，原本清晰的业务逻辑被胶水代码弄得支离破碎。

就在我为此头疼时，magentic这个项目进入了我的视野。它的Slogan非常吸引人：“将LLM无缝集成到你的Python代码中，就像调用普通函数一样。” 这听起来像是魔法——你定义一个函数签名，写上文档字符串（docstring），magentic就能自动帮你处理与LLM的交互，将自然语言输出转换成你指定的Python类型。它不是一个试图用LLM生成整个程序的框架，而是一个精巧的“粘合剂”，旨在让LLM成为你现有Python工具箱中一个顺滑、类型安全的扩展。简单来说，它让“提示工程”变得像函数调用一样自然和可靠。

这个项目适合任何希望在Python应用中引入LLM能力，但又不想被繁琐的API调用和输出解析困扰的开发者。无论是构建一个智能客服的意图分类器，一个从用户自由文本中提取结构化信息的处理器，还是一个根据描述生成代码片段的工具，magentic都能大幅提升开发体验和代码的可维护性。接下来，我将带你深入拆解这个项目的设计哲学、核心用法，并分享我在实际集成中踩过的坑和总结出的最佳实践。

2. 核心设计哲学与架构拆解

2.1 函数即提示：声明式编程范式的胜利

magentic最核心、也最迷人的思想，是“函数即提示”（Function as Prompt）。在传统的LLM调用中，我们需要显式地构建一个提示词（Prompt）字符串，这个字符串通常混合了指令、示例和用户输入。这种方式是“命令式”的，开发者需要关注通信的细节。

magentic反其道而行之，采用了“声明式”的范式。你不需要告诉LLM“怎么回答”，你只需要声明你“想要什么”。这个声明，就是通过标准的Python函数定义来完成的。函数的名称、参数（包括类型注解和参数名）、返回值类型注解，以及最重要的——函数的文档字符串（docstring），共同构成了一个对LLM的完整、类型化的指令集。

举个例子，你想让LLM判断一段文本的情感是积极还是消极。传统方式你可能要写：“请分析以下文本的情感，如果是积极的，返回‘positive’；如果是消极的，返回‘negative’。文本：{user_input}”。而在magentic中，你只需要这样定义：

from magentic import prompt from enum import Enum class Sentiment(Enum): POSITIVE = “positive” NEGATIVE = “negative” @prompt def analyze_sentiment(text: str) -> Sentiment: “””判断给定文本的情感倾向。“”” ...

当你调用analyze_sentiment(“这部电影真是太棒了！”)时，magentic会在幕后将函数签名和docstring组合成一个结构化的提示，发送给LLM，并强制要求LLM的输出必须能被解析成Sentiment枚举类型。如果LLM返回了“happy”这样的字符串，magentic的解析器会抛出异常，而不是让一个非法值流入你的核心业务逻辑。这种设计将输入输出规范从松散的字符串提升到了严格的类型系统层面，极大地增强了程序的健壮性。

2.2 类型系统的双向桥梁：确保确定性的关键

LLM的本质是概率模型，其输出具有不确定性。而Python程序（尤其是后端服务）追求的是确定性。如何弥合这中间的鸿沟？magentic的答案是：充分利用Python强大的类型注解系统。

magentic内置了对常见Python类型的支持，如str,int,float,bool,list,dict等。更强大的是，它支持Pydantic的BaseModel和Python的Enum。这意味着你可以用这些类型来精确描述你期望的输出结构。

类型作为输出约束：当你指定返回值类型为list[str]时，magentic不仅会在提示中告诉LLM“请返回一个字符串列表”，还会在收到响应后，尝试用ast.literal_eval或json.loads来解析它，并验证结果是否确实是字符串列表。这相当于为LLM的输出加上了一层强制性的“语法校验”。

类型作为输入描述：函数的参数名和类型注解，同样会被magentic用于构建更清晰的提示。例如，一个参数email_address: str会比单纯的text: str更能让LLM理解这个字符串应该符合邮箱格式的预期，尽管这不是硬性约束，但能显著提升交互质量。

这种深度集成类型系统的做法，使得“与LLM对话”这件事从一门艺术（高度依赖提示词技巧）向着一门工程（可预测、可测试、可维护）迈进了一大步。它迫使开发者在设计功能时首先思考输入输出的边界和形态，这是一种良好的设计驱动。

2.3 异步优先与流式响应：面向现代应用的架构

现代Python应用离不开异步IO来提高并发性能。magentic从设计之初就拥抱了asyncio。所有主要的装饰器（如@prompt）都同时支持同步和异步函数。当你进行批处理或在一个Web后端（如FastAPI）中调用时，使用异步版本可以避免阻塞事件循环，轻松实现并发调用多个LLM任务。

另一个紧跟潮流的特性是对流式响应（Streaming）的支持。当LLM生成一个很长的回答（比如一篇邮件草稿或一段解释）时，等待全部生成完毕再返回会给用户带来明显的延迟感。magentic允许你将函数返回值类型注解为Iterable[str]（或异步的AsyncIterable[str]）。

from magentic import prompt from collections.abc import Iterable @prompt def generate_story(theme: str) -> Iterable[str]: “””根据给定的主题，生成一个短篇故事。“”” ... # 调用时，你可以逐个token地接收输出 for chunk in generate_story(“太空探险”): print(chunk, end=“”, flush=True) # 实现打字机效果

这在构建交互式聊天应用或需要实时反馈的场景中至关重要。magentic在底层处理了与LLM API流式接口的对接，为你提供了干净的Python迭代器抽象，简化了开发。

注意：流式响应和异步通常结合使用。在处理高并发流式请求时，务必注意资源管理，避免生成器未正确关闭导致的内存或连接泄漏。我的经验是，在Web框架中使用时，确保在响应结束后或连接中断时，主动关闭或丢弃相关的异步生成器。

3. 核心功能深度解析与实战

3.1@prompt装饰器：一切魔法的起点

@prompt装饰器是magentic的瑞士军刀，绝大多数功能都围绕它展开。它的核心职责是：拦截对被装饰函数的调用，将函数元信息转化为LLM提示，调用配置好的LLM，解析响应，并返回类型化的结果。

基础用法与参数注入： 最简单的用法就是直接装饰一个函数。但magentic的强大之处在于，你可以在调用函数时，动态地覆盖某些默认行为。这是通过向被装饰函数注入特殊参数实现的：

• magentic_model: 临时指定本次调用使用的LLM模型（如从默认的gpt-4切换到gpt-3.5-turbo以节省成本）。
• magentic_prompt_template: 覆盖默认的提示模板。默认模板会使用函数签名和docstring，但你可以提供一个字符串模板，其中用{参数名}来引用函数参数。
• magentic_output_parser: 指定一个自定义的解析器来处理LLM的原始输出。

from magentic import prompt, OpenAIChatModel from magentic.chat_model.openai_chat_model import OpenAIChatModel @prompt def explain_concept(concept: str) -> str: “””用简单的语言解释{concept}这个概念。“”” # 动态使用更便宜的模型进行单次调用 result = explain_concept( “递归”, magentic_model=OpenAIChatModel(“gpt-3.5-turbo”) )

这种设计提供了极大的灵活性，允许你在全局配置和单次调用之间做精细的权衡。

多函数与函数调用（Function Calling）集成：magentic优雅地支持了OpenAI风格的函数调用（Function Calling）。你可以定义多个@prompt函数，然后将它们作为一个“函数工具包”提供给LLM。LLM会根据对话上下文，自动决定是否调用、以及调用哪个函数。

from magentic import prompt, FunctionCall from pydantic import BaseModel class WeatherInfo(BaseModel): location: str temperature_c: float condition: str @prompt def get_weather(city: str) -> WeatherInfo: “””获取指定城市的当前天气信息。这是一个模拟函数。“”” # 在实际应用中，这里会调用真实的天气API return WeatherInfo(location=city, temperature_c=22.5, condition=“Sunny”) @prompt def set_reminder(task: str, time: str) -> str: “””设置一个提醒。“”” return f“Reminder ‘{task}’ set for {time}.” # 在一个聊天循环中，你可以将这两个函数作为工具提供 # magentic 会处理与LLM的交互，自动选择并执行函数

这使得构建具备复杂工具使用能力的智能体（Agent）变得异常简单。magentic负责了底层的协议转换和调度，你只需要关心每个工具函数本身的业务逻辑。

3.2 聊天会话管理：让LLM拥有记忆

单次的@prompt函数调用是独立的。但对于聊天机器人或多轮对话场景，我们需要LLM记住之前的对话历史。magentic提供了Chat类来管理会话状态。

一个Chat对象维护着一个消息列表（messages），其中包含用户消息、助理消息以及可能的函数调用结果消息。它的核心方法是step，用于向会话中添加一条用户消息，并获取LLM的响应。

from magentic import Chat, OpenAIChatModel from magentic.chat_model.message import UserMessage # 初始化一个聊天会话，可以指定系统提示词和模型 chat = Chat( model=OpenAIChatModel(“gpt-4”), system_prompt=“你是一个乐于助人的编程助手。” ) # 进行多轮对话 user_message = UserMessage(“Python中的装饰器是什么？”) response1 = chat.step(user_message) print(response1.content) user_message2 = UserMessage(“能给我一个例子吗？”) response2 = chat.step(user_message2) print(response2.content) # 你可以随时访问完整的对话历史 for msg in chat.messages: print(f“{msg.role}: {msg.content}”)

Chat类的设计是显式且直观的。它没有试图隐藏状态，而是将其暴露给你完全控制。你可以轻松地将会话历史保存到数据库（例如，将chat.messages序列化为JSON），并在下次用户连接时恢复，从而实现持久化的对话体验。

实操心得：在Web应用中管理Chat实例时，一个常见的陷阱是将Chat对象直接存储在类似Flask的session或全局变量中。这在多进程/多服务器环境下会失败。正确的做法是使用中心化的存储（如Redis），以用户会话ID为键，存储序列化后的消息列表。每次请求时，从存储中加载消息列表，实例化一个新的Chat对象。虽然Chat对象本身可能无法直接序列化，但其核心的messages列表通常是可序列化的Pydantic模型列表，这是存储和恢复的关键。

3.3 输出解析与错误处理：构建健壮的生产级应用

LLM的输出是不可靠的，因此强大的输出解析和错误处理机制是生产应用的生命线。magentic在这方面提供了多层防御。

内置解析器与回退策略： 当使用@prompt装饰器时，magentic会根据返回值类型自动选择解析器。例如，对于Pydantic模型，它会尝试将LLM的输出解析为JSON，然后用它来实例化模型。如果解析失败（比如JSON无效，或字段类型不匹配），magentic默认会抛出异常（如OutputParserError）。

对于生产环境，直接抛出异常可能过于粗暴。你可以通过装饰器参数或全局配置设置重试策略：

from magentic import prompt, OpenAIChatModel from magentic.typing import is_annotated from typing import Annotated import tenacity # 方法1：使用tenacity库进行装饰器重试 @retry(stop=tenacity.stop_after_attempt(3), reraise=True) @prompt def extract_info(text: str) -> MyModel: ... # 方法2：更优雅的方式，利用magentic对Annotated类型的支持（如果版本支持） # 这提示magentic在解析失败时返回None，而不是抛出异常 from typing import Optional @prompt def extract_info_safe(text: str) -> Optional[MyModel]: “””从文本中提取信息，如果失败则返回None。“”” ...

自定义解析器应对复杂场景： 有时，LLM的输出可能不是标准的JSON，而是一段包含结构化信息的文本。此时，你可以定义自定义解析器。

from magentic import prompt, FunctionCall from magentic.output_parsers import OutputParser import re class MyCustomParser(OutputParser): def parse(self, output: str, **kwargs) -> dict: # 使用正则表达式从文本中提取键值对 pattern = r“(\w+):\s*(.+)” matches = re.findall(pattern, output) return dict(matches) @prompt(output_parser=MyCustomParser()) def parse_unstructured_text(report: str) -> dict: “””从自由格式的报告中提取关键信息。“”” ...

结构化异常处理： 在你的应用代码中，你应该预料到@prompt函数可能失败。建议使用清晰的异常处理逻辑。

from magentic import OutputParserError from openai import APIError, RateLimitError try: result = my_prompt_function(some_input) except OutputParserError as e: # LLM的输出无法解析为预期类型 logger.warning(f“LLM returned malformed output: {e}”) # 可以回退到默认值，或向用户请求澄清 result = get_fallback_value() except (APIError, RateLimitError) as e: # 底层API调用失败（网络、鉴权、限流） logger.error(f“LLM API call failed: {e}”) # 实施降级策略或友好错误提示 raise ServiceUnavailableError(“AI服务暂时不可用”) from e

将LLM调用视为一个可能失败的外部服务，而非可靠的本地函数，是构建稳定应用的心态基础。magentic提供的清晰错误类型，有助于你建立分层的、精细的异常处理策略。

4. 高级用法与性能优化实战

4.1 提示词工程进阶：少样本学习与思维链

虽然magentic通过函数签名和docstring自动生成提示，但你仍然可以（也经常需要）进行深入的提示词工程。最有效的方法就是在docstring中嵌入少样本学习（Few-shot Learning）示例和思维链（Chain-of-Thought, CoT）引导。

在Docstring中嵌入少样本示例： docstring不仅仅是指令，它也是主要的上下文载体。通过提供输入输出的例子，你可以极大地提升LLM的表现。

from pydantic import BaseModel from magentic import prompt class Classification(BaseModel): sentiment: str # “positive”, “negative”, “neutral” confidence: float keywords: list[str] @prompt def analyze_customer_feedback(feedback: str) -> Classification: “”” 分析客户反馈文本，识别情感、置信度和关键词。 示例1： 输入：“产品很好用，但送货太慢了。” 输出：{{“sentiment”: “neutral”, “confidence”: 0.8, “keywords”: [“好用”, “送货慢”]}} 示例2： 输入：“这是我用过最糟糕的服务，客服根本不解决问题！” 输出：{{“sentiment”: “negative”, “confidence”: 0.95, “keywords”: [“最糟糕”, “客服”, “不解决”]}} 现在请分析以下反馈：{feedback} “”” ...

注意，在docstring的示例中，我直接使用了JSON字面量来描述输出。这是因为LLM（特别是GPT-4）非常擅长理解这种内联的示例格式。magentic在构建最终提示时，会将整个docstring（包括这些示例）原样发送给LLM，同时附加一条“请以指定JSON格式返回”的指令。LLM会参考示例的格式进行输出。

引导思维链（CoT）： 对于需要推理的复杂任务，在docstring中要求LLM“逐步思考”可以显著提高答案的准确性和可靠性。

@prompt def solve_math_word_problem(problem: str) -> float: “”” 解决以下数学应用题。请按步骤思考，并在最后一行以“答案：<数字>”的格式给出最终答案。 问题：{problem} 让我们一步步来： 1. 首先，理解问题在问什么。 2. 其次，找出已知条件和未知数。 3. 然后，建立方程或逻辑关系。 4. 接着，执行计算。 5. 最后，给出答案。 答案： “”” ...

在这种模式下，LLM的输出会包含完整的推理过程，而magentic的解析器需要能够从文本中提取出最后的答案数字。你可以搭配一个自定义解析器，使用正则表达式来匹配“答案：”后面的数字。

4.2 模型管理与成本控制：多模型策略与缓存

在生产环境中，我们通常需要与多个LLM提供商（如OpenAI、Anthropic、本地部署的模型）打交道，并且需要严格控制成本。magentic通过统一的ChatModel接口和依赖注入设计，让这些变得容易。

配置多模型与回退策略： 你可以定义多个模型，并根据场景或故障情况切换。

from magentic import OpenAIChatModel, AnthropicChatModel from magentic.chat_model.litellm_chat_model import LiteLLMChatModel import os # 定义不同成本和能力的模型 primary_model = OpenAIChatModel(“gpt-4”, api_key=os.getenv(“OPENAI_API_KEY”)) fallback_model = OpenAIChatModel(“gpt-3.5-turbo”, api_key=os.getenv(“OPENAI_API_KEY”)) claude_model = AnthropicChatModel(“claude-3-sonnet”, api_key=os.getenv(“ANTHROPIC_API_KEY”)) # 通过LiteLLM使用其他兼容API的模型 local_model = LiteLLMChatModel(“togethercomputer/llama-2-70b-chat”, api_key=os.getenv(“TOGETHER_API_KEY”)) # 在函数中动态选择或实现一个简单的回退逻辑 def call_llm_with_fallback(prompt_func, *args, **kwargs): models_to_try = [primary_model, fallback_model, claude_model] for model in models_to_try: try: return prompt_func(*args, **kwargs, magentic_model=model) except (APIError, RateLimitError) as e: logger.warning(f“Model {model.model} failed, trying next.”) continue raise Exception(“All models failed”)

实现请求缓存以节省成本： 许多LLM调用是幂等的，特别是那些基于确定性的输入（如用户查询、产品描述）生成内容的函数。为这些函数添加缓存可以大幅降低成本。

from functools import lru_cache from magentic import prompt import hashlib import json def make_cache_key(func_name, *args, **kwargs): “”“创建一个基于函数名和参数的缓存键。”“” # 将参数序列化为字符串，注意要排序以确保字典参数顺序一致 args_repr = json.dumps(args, sort_keys=True, default=str) kwargs_repr = json.dumps(kwargs, sort_keys=True, default=str) key_str = f“{func_name}:{args_repr}:{kwargs_repr}” return hashlib.md5(key_str.encode()).hexdigest() # 一个简单的内存缓存装饰器（生产环境应使用Redis等） def llm_cache(func): cache = {} def wrapper(*args, **kwargs): # 注意：这里需要排除magentic注入的参数，如`magentic_model` key_kwargs = {k: v for k, v in kwargs.items() if not k.startswith(‘magentic_’)} key = make_cache_key(func.__name__, *args, **key_kwargs) if key not in cache: cache[key] = func(*args, **kwargs) return cache[key] return wrapper @llm_cache @prompt def generate_product_description(product_name: str, features: list[str]) -> str: “””根据产品名称和特性列表，生成一段营销描述。“”” ...

重要提醒：缓存LLM响应时，必须仔细考虑缓存键的构成。除了业务参数，模型名称、温度（temperature）等影响输出的参数也应包含在内。同时，要设置合理的缓存过期策略，特别是对于时效性强的信息。对于分布式应用，务必使用分布式缓存（如Redis）而非内存缓存。

4.3 流式处理与批量操作：提升吞吐量

对于需要处理大量文档或用户请求的应用，性能至关重要。magentic的异步支持和流式特性在这里可以大显身手。

异步批量处理： 假设你需要为一批产品生成描述。

import asyncio from magentic import prompt from magentic.chat_model.openai_chat_model import OpenAIChatModel @prompt async def agenerate_description(product: dict) -> str: “””为产品生成描述。产品信息：{product}“”” ... async def batch_generate_descriptions(products: list[dict]): model = OpenAIChatModel(“gpt-3.5-turbo”) # 使用更经济的模型进行批量任务 tasks = [agenerate_description(p, magentic_model=model) for p in products] # 使用asyncio.gather并发执行，但注意API的速率限制！ # 建议使用semaphore进行并发控制 semaphore = asyncio.Semaphore(10) # 控制最大并发数为10 async def bounded_task(task): async with semaphore: return await task bounded_tasks = [bounded_task(t) for t in tasks] results = await asyncio.gather(*bounded_tasks, return_exceptions=True) # 处理结果和可能的异常 for product, result in zip(products, results): if isinstance(result, Exception): logger.error(f“Failed for product {product[‘id’]}: {result}”) else: # 存储result ...

结合流式响应处理长内容： 当生成内容很长时，流式响应不仅能改善用户体验，有时还能提升整体处理速度（因为可以边生成边处理/发送）。

from magentic import prompt from collections.abc import AsyncIterable @prompt async def stream_long_analysis(document: str) -> AsyncIterable[str]: “””分析这篇长文档，并逐部分输出分析结果。文档：{document}“”” ... async def process_stream(): async for chunk in stream_long_analysis(a_very_long_doc): # 可以实时将chunk发送到WebSocket，或写入文件流 websocket.send(chunk) # 也可以进行实时处理，例如情感分析 process_chunk(chunk)

性能监控与限流： 在投入生产前，务必对你的LLM调用进行性能剖析和压力测试。监控平均响应时间、令牌使用量、错误率等指标。使用令牌桶或漏桶算法在应用层实现限流，以避免触发上游API的速率限制而导致服务中断。magentic本身不提供限流机制，这需要你在业务逻辑层或使用像asyncio.Semaphore这样的工具自行实现。

5. 集成实战：构建一个智能客服意图分类模块

让我们通过一个完整的实战案例，将magentic集成到一个模拟的智能客服系统中。我们的目标是构建一个模块，能够准确地将用户输入的任意文本分类到预定义的意图中，并提取关键参数。

5.1 需求分析与设计

假设我们的客服系统需要处理以下意图：

1. 查询订单状态(query_order): 需要提取订单号。
2. 退货申请(request_return): 需要提取产品名称和问题描述。
3. 投诉建议(complaint_suggestion): 需要提取投诉对象和详细内容。
4. 一般咨询(general_inquiry): 无需提取特定参数。

我们需要一个函数，输入是用户消息字符串，输出是一个结构化的对象，包含intent（意图）和parameters（参数字典）。

5.2 实现步骤

第一步：定义数据模型我们使用Pydantic来定义清晰的数据结构。

from pydantic import BaseModel, Field from typing import Literal, Optional class CustomerIntent(BaseModel): “”“客服用户意图识别结果”“” intent: Literal[“query_order”, “request_return”, “complaint_suggestion”, “general_inquiry”] = Field( …, description=“识别出的用户意图” ) parameters: dict[str, str] = Field( default_factory=dict, description=“从用户消息中提取出的关键参数，键值对形式” ) confidence: float = Field( …, ge=0.0, le=1.0, description=“模型对此分类结果的置信度，0到1之间” )

第二步：实现核心分类函数使用@prompt装饰器，并在docstring中提供清晰的指令和示例。

from magentic import prompt from magentic.chat_model.openai_chat_model import OpenAIChatModel # 全局配置默认模型（可在环境变量中配置） DEFAULT_MODEL = OpenAIChatModel(“gpt-4”, api_key=os.getenv(“OPENAI_API_KEY”)) @prompt def classify_customer_intent(user_message: str) -> CustomerIntent: “”” 你是一个智能客服意图分类器。请将用户的输入消息分类到以下意图之一，并提取相关参数。 如果消息意图不明确或不属于任何类别，请归类为“general_inquiry”。 **可用的意图：** 1. query_order - 用户查询订单状态。需要提取参数：`order_id`（订单号）。 2. request_return - 用户申请退货。需要提取参数：`product_name`（产品名称），`issue`（问题描述）。 3. complaint_suggestion - 用户进行投诉或提出建议。需要提取参数：`subject`（投诉对象，如‘物流’、‘客服’、‘产品质量’），`details`（详细内容）。 4. general_inquiry - 一般性咨询，或无法归入以上类别。 **输出要求：** - 请以JSON格式输出，严格匹配`CustomerIntent`模型的格式。 - `intent`字段必须是上述四个值之一。 - `parameters`字段是一个字典，只包含从消息中明确提取出的参数。如果某个意图要求的参数未在消息中找到，则该参数不应出现在字典中。 - `confidence`字段是你对此次分类的置信度评分（0-1之间）。 **示例：** 输入：“我的订单#12345到哪里了？” 输出：{{“intent”: “query_order”, “parameters”: {{“order_id”: “12345”}}, “confidence”: 0.95}} 输入：“刚买的手机屏幕有裂痕，我想退货。” 输出：{{“intent”: “request_return”, “parameters”: {{“product_name”: “手机”, “issue”: “屏幕有裂痕”}}, “confidence”: 0.9}} 输入：“你们服务挺好的。” 输出：{{“intent”: “general_inquiry”, “parameters”: {{}}, “confidence”: 0.7}} 现在，请对以下用户消息进行分类： 消息：{user_message} “”” # 函数体不会被实际执行，docstring就是提示词 …

第三步：添加业务逻辑层与错误处理在实际应用中，分类函数可能失败。我们需要一个服务层来封装它，并处理各种边界情况。

import logging from magentic import OutputParserError from openai import APIError logger = logging.getLogger(__name__) class IntentClassificationService: def __init__(self, fallback_model=None): self.fallback_model = fallback_model async def classify(self, user_message: str) -> CustomerIntent: “”“分类用户意图，具备重试和降级逻辑。”“” models_to_try = [DEFAULT_MODEL] if self.fallback_model: models_to_try.append(self.fallback_model) last_exception = None for model in models_to_try: try: # 注意：classify_customer_intent是同步函数，在异步上下文中需使用asyncio.to_thread # 或者，可以定义异步版本的@prompt函数 result = await asyncio.to_thread( classify_customer_intent, user_message, magentic_model=model ) # 对低置信度结果进行日志记录或特殊处理 if result.confidence < 0.6: logger.warning(f“Low confidence classification: {result} for message ‘{user_message}‘”) return result except OutputParserError as e: logger.error(f“Failed to parse LLM output for message ‘{user_message}‘ with model {model.model}: {e}”) last_exception = e # 解析错误通常重试无效，直接跳出 break except (APIError, TimeoutError) as e: logger.warning(f“LLM API call failed with model {model.model}: {e}. Retrying with next model if available.”) last_exception = e continue # 尝试下一个模型 # 所有尝试都失败 logger.error(f“All models failed for message ‘{user_message}‘. Last error: {last_exception}”) # 返回一个安全的默认值 return CustomerIntent( intent=“general_inquiry”, parameters={}, confidence=0.0 )

第四步：集成到Web框架（以FastAPI为例）

from fastapi import FastAPI, HTTPException from pydantic import BaseModel as FastAPIBaseModel app = FastAPI() service = IntentClassificationService(fallback_model=OpenAIChatModel(“gpt-3.5-turbo”)) class ClassificationRequest(FastAPIBaseModel): message: str class ClassificationResponse(FastAPIBaseModel): intent: str parameters: dict confidence: float @app.post(“/classify”, response_model=ClassificationResponse) async def classify_intent(request: ClassificationRequest): try: result = await service.classify(request.message) return ClassificationResponse( intent=result.intent, parameters=result.parameters, confidence=result.confidence ) except Exception as e: # 记录内部错误，但对外返回通用错误 logger.exception(“Intent classification failed.”) raise HTTPException(status_code=503, detail=“Service temporarily unavailable.”)

5.3 部署与监控考量

将这个模块部署到生产环境，还需要考虑以下几点：

1. 配置管理：将LLM API密钥、模型名称、温度等配置项放在环境变量或配置中心，不要硬编码在代码中。
2. 日志记录：详细记录分类请求、响应、置信度、令牌使用量和延迟。这对于监控成本、分析模型性能和调试问题至关重要。
3. 性能与限流：在API网关或应用层对/classify端点实施限流，防止滥用。同时，监控service.classify方法的平均响应时间，如果超过阈值（如2秒），考虑优化提示词或降级模型。
4. 评估与迭代：定期收集分类错误的样本，用于评估和优化你的提示词（docstring）。可以建立一个简单的标注界面，让客服人员对模型的分类结果进行修正，这些数据是改进系统最宝贵的资产。

通过这个案例，你可以看到magentic如何将复杂的LLM交互抽象成一个简单的、类型安全的函数调用，并如何将其嵌入到一个完整的、健壮的生产级应用架构中。它处理了与LLM通信的复杂性，让你可以专注于业务逻辑和用户体验。