magentic

将大型语言模型无缝集成到 Python 代码中。使用 @prompt 和 @chatprompt 装饰器创建从 LLM 返回结构化输出的函数。将 LLM 查询和工具使用与传统 Python 代码相结合，构建复杂的智能体系统。

特性

• 使用 Pydantic 模型和内置 Python 类型实现【结构化输出】。
• 支持结构化输出和函数调用的【流式传输】，以便在生成过程中实时使用。
• 【LLM 辅助重试】功能，以提高 LLM 对复杂输出模式的遵循度。
• 基于 OpenTelemetry 的【可观测性】，并原生集成【Pydantic Logfire】。
• 【类型注解】，便于与 linter 和 IDE 良好配合。
• 针对多个 LLM 提供商（包括 OpenAI、Anthropic 和 Ollama）的【配置】选项。
• 还有更多特性：【聊天提示】、【并行函数调用】、【视觉处理】、【格式化】、【异步支持】等。

安装

pip install magentic

或使用 uv：

uv add magentic

通过设置 OPENAI_API_KEY 环境变量来配置您的 OpenAI API 密钥。如需配置其他 LLM 提供商，请参阅【配置】部分以获取更多信息。

使用方法

@prompt

@prompt 装饰器允许您将大型语言模型 (LLM) 提示模板定义为一个 Python 函数。当此函数被调用时，参数会被插入到模板中，然后该提示会被发送到 LLM，由 LLM 生成函数的输出。

from magentic import prompt


@prompt('给以下句子增加“兄弟感”：{phrase}')
def dudeify(phrase: str) -> str: ...  # 无需函数体，因为该函数不会被执行


dudeify("你好，最近怎么样？")
# "嘿，兄弟！咋样啊？老铁，过得咋样？"

@prompt 装饰器会尊重被装饰函数的返回值类型注解。这可以是 Pydantic 支持的任何类型，包括一个 pydantic 模型。

from magentic import prompt
from pydantic import BaseModel


class Superhero(BaseModel):
    name: str
    age: int
    power: str
    enemies: list[str]


@prompt("创建一位名为 {name} 的超级英雄。")
def create_superhero(name: str) -> Superhero: ...


create_superhero("花园侠")
# Superhero(name='花园侠', age=30, power='掌控植物', enemies=['污染侠', '水泥女'])

更多内容请参见【结构化输出】。

@chatprompt

@chatprompt 装饰器的工作方式与 @prompt 类似，但它允许您将聊天消息作为模板传递，而不是单个文本提示。这可用于提供系统消息，或进行少样本提示，其中您提供示例响应来引导模型的输出。用大括号 {example} 标记的格式字段将在所有消息中填充（除了 FunctionResultMessage）。

from magentic import chatprompt, AssistantMessage, SystemMessage, UserMessage
from pydantic import BaseModel


class Quote(BaseModel):
    quote: str
    character: str


@chatprompt(
    SystemMessage("你是个电影迷。"),
    UserMessage("你最喜欢的《哈利·波特》台词是什么？"),
    AssistantMessage(
        Quote(
            quote="沉溺于梦想而忘记生活是没有意义的。",
            character="阿不思·邓布利多",
        )
    ),
    UserMessage("你最喜欢的 {movie} 台词是什么？"),
)
def get_movie_quote(movie: str) -> Quote: ...


get_movie_quote("钢铁侠")
# Quote(quote='我是钢铁侠。', character='托尼·斯塔克')

更多内容请参见【聊天提示】。

FunctionCall

LLM 也可以决定调用函数。在这种情况下，使用 @prompt 装饰器的函数会返回一个 FunctionCall 对象，该对象可以被调用以使用 LLM 提供的参数执行函数。

from typing import Literal

from magentic import prompt, FunctionCall


def search_twitter(query: str, category: Literal["latest", "people"]) -> str:
    """搜索 Twitter 上的查询内容。"""
    print(f"正在 Twitter 上搜索 {query!r}，类别为 {category!r}")
    return "<twitter 搜索结果>"


def search_youtube(query: str, channel: str = "all") -> str:
    """搜索 YouTube 上的查询内容。"""
    print(f"正在 YouTube 上搜索 {query!r}，频道为 {channel!r}")
    return "<youtube 搜索结果>"


@prompt(
    "使用适当的搜索函数回答：{question}",
    functions=[search_twitter, search_youtube],
)
def perform_search(question: str) -> FunctionCall[str]: ...


output = perform_search("关于 LLM 的最新消息是什么？")
print(output)
# > FunctionCall(<function search_twitter at 0x10c367d00>, 'LLMs', 'latest')
output()
# > 正在 Twitter 上搜索 'Large Language Models news'，类别为 'latest'
# '<twitter 搜索结果>'

更多内容请参见【函数调用】。

@prompt_chain

有时，LLM 需要调用一个或多个函数才能生成最终答案。@prompt_chain 装饰器会自动解析 FunctionCall 对象，并将输出传递回 LLM，直到获得最终答案为止。

在下面的例子中，当调用 describe_weather 时，LLM 首先调用 get_current_weather 函数，然后利用其结果形成最终答案并返回。

from magentic import prompt_chain


def get_current_weather(location, unit="fahrenheit"):
    """获取指定地点的当前天气情况"""
    # 模拟查询 API
    return {"temperature": "72", "forecast": ["晴朗", "多风"]}


@prompt_chain(
    "波士顿的天气如何？",
    functions=[get_current_weather],
)
def describe_weather(city: str) -> str: ...


describe_weather("波士顿")
# '波士顿目前的天气是 72°F，晴朗且多风。'

使用 @prompt、@chatprompt 和 @prompt_chain 创建的 LLM 驱动函数，可以像普通 Python 函数一样，作为 functions 参数传递给其他 @prompt/@prompt_chain 装饰器。这使得 LLM 驱动的功能越来越复杂，同时允许单独测试和改进各个组件。

流式处理

StreamedStr（和 AsyncStreamedStr）类可用于流式传输 LLM 的输出。这使您能够在文本生成过程中对其进行处理，而不是一次性接收整个输出。

from magentic import prompt, StreamedStr


@prompt("告诉我关于 {country} 的事情")
def describe_country(country: str) -> StreamedStr: ...


# 在接收到每个数据块时立即打印
for chunk in describe_country("巴西"):
    print(chunk, end="")
# '巴西，正式名称为巴西联邦共和国，是 ...'

可以同时创建多个 StreamedStr 对象，以并发地流式传输 LLM 的输出。在下面的示例中，为多个国家生成描述所需的时间与为单个国家生成描述所需的时间大致相同。

from time import time

countries = ["澳大利亚", "巴西", "智利"]


# 依次生成各国的描述
start_time = time()
for country in countries:
    # 将 `StreamedStr` 转换为 `str` 会阻塞，直到 LLM 输出完全生成
    description = str(describe_country(country))
    print(f"{time() - start_time:.2f}s : {country} - {len(description)} 字符")

# 22.72s : 澳大利亚 - 2130 字符
# 41.63s : 巴西 - 1884 字符
# 74.31s : 智利 - 2968 字符


# 同时创建 `StreamedStr` 对象，以并发方式生成各国的描述
start_time = time()
streamed_strs = [describe_country(country) for country in countries]
for country, streamed_str in zip(countries, streamed_strs):
    description = str(streamed_str)
    print(f"{time() - start_time:.2f}s : {country} - {len(description)} 字符")

# 22.79s : 澳大利亚 - 2147 字符
# 23.64s : 巴西 - 2202 字符
# 24.67s : 智利 - 2186 字符

对象流式处理

也可以通过使用返回值类型注解 Iterable（或 AsyncIterable）从 LLM 流式传输结构化输出。这样可以在生成下一个项目的同时处理当前项目。

from collections.abc import Iterable
from time import time

from magentic import prompt
from pydantic import BaseModel


class 超级英雄(BaseModel):
    名字: str
    年龄: int
    能力: str
    敌人: list[str]


@prompt("创建一个名为 {name} 的超级英雄团队。")
def 创建超级英雄团队(name: str) -> Iterable[超级英雄]: ...


start_time = time()
for hero in 创建超级英雄团队("美食小队"):
    print(f"{time() - start_time:.2f}s : {hero}")

# 2.23s : 名字='披萨侠' 年龄=30 能力='能从手中射出披萨片' 敌人=['饥饿军团', '垃圾食品帮']
# 4.03s : 名字='胡萝卜队长' 年龄=35 能力='吃胡萝卜后拥有超强力量和敏捷性' 敌人=['糖分小队', '油腻帮']
# 6.05s : 名字='冰淇淋女孩' 年龄=25 能力='能凭空制造冰淇淋' 敌人=['辣酱小队', '健康饮食者']

更多内容请参阅 [流式处理]。

Asyncio

可以使用异步函数/协程来并发查询 LLM。这不仅能显著提高整体生成速度，还能在等待 LLM 输出时运行其他异步代码。在下面的示例中，LLM 在等待列表中的下一位总统时，会为每位美国总统生成一段描述。通过测量每秒生成的字符数可以看出，该示例相比串行处理实现了 7 倍的加速。

import asyncio
from time import time
from typing import AsyncIterable

from magentic import prompt


@prompt("列出美国的十位总统")
async def 遍历总统() -> AsyncIterable[str]: ...


@prompt("给我讲讲关于 {topic} 的事")
async def 讲讲关于(topic: str) -> str: ...


# 为每位总统列出的描述并发生成
start_time = time()
tasks = []
async for president in await 遍历总统():
    # 使用 asyncio.create_task 安排协程执行，然后再等待其完成
    # 这样在生成总统列表的同时，描述也会开始生成
    task = asyncio.create_task(讲讲关于(president))
    tasks.append(task)

descriptions = await asyncio.gather(*tasks)

# 测量每秒生成的字符数
总字符数 = sum(len(desc) for desc in descriptions)
时间流逝 = time() - start_time
print(总字符数, 时间流逝, 总字符数 / 时间流逝)
# 24575 28.70 856.07


# 测量描述一位总统时每秒生成的字符数
start_time = time()
out = await 讲讲关于("乔治·华盛顿")
时间流逝 = time() - start_time
print(len(out), 时间流逝, len(out) / 时间流逝)
# 2206 18.72 117.78

更多内容请参阅 [Asyncio]。

其他功能

• @prompt 的 functions 参数可以包含异步函数/协程。当调用相应的 FunctionCall 对象时，必须等待结果。
• 可以使用 Annotated 类型注解为函数参数提供描述和其他元数据。请参阅 Pydantic 文档中关于使用 Field 描述函数参数的内容。
• @prompt 和 @prompt_chain 装饰器还接受 model 参数。您可以传入 OpenaiChatModel 实例以使用 GPT-4，或配置不同的温度。详情见下文。
• 按照 Pandas DataFrame 的示例笔记本，注册其他类型作为 @prompt 函数的返回值类型注解。

后端/LLM 配置

Magentic 支持多种 LLM 提供商或“后端”。这大致指用于与 LLM API 交互的 Python 包。目前支持以下后端：

OpenAI

默认后端，使用 openai Python 包，并支持 magentic 的所有功能。

无需额外安装。只需从 magentic 中导入 OpenaiChatModel 类即可。

from magentic import OpenaiChatModel

model = OpenaiChatModel("gpt-4o")

通过 OpenAI 使用 Ollama

Ollama 支持与 OpenAI 兼容的 API，因此你可以通过 OpenAI 后端使用 Ollama 模型。

首先，从 ollama.com 安装 Ollama。然后拉取你想要使用的模型。

ollama pull llama3.2

接着，在创建 OpenaiChatModel 实例时，指定模型名称和 base_url。

from magentic import OpenaiChatModel

model = OpenaiChatModel("llama3.2", base_url="http://localhost:11434/v1/")

其他与 OpenAI 兼容的 API

当使用 openai 后端时，设置 MAGENTIC_OPENAI_BASE_URL 环境变量，或者在代码中使用 OpenaiChatModel(..., base_url="http://localhost:8080")，即可将 magentic 与任何 OpenAI 兼容的 API 配合使用，例如 Azure OpenAI 服务、LiteLLM OpenAI 代理服务器、LocalAI 等。请注意，如果 API 不支持工具调用，则无法创建返回 Python 对象的提示函数，但 magentic 的其他功能仍可正常使用。

若要将 Azure 与 openai 后端配合使用，需将 MAGENTIC_OPENAI_API_TYPE 环境变量设置为 "azure"，或在代码中使用 OpenaiChatModel(..., api_type="azure")，并同时设置 openai 包访问 Azure 所需的环境变量。详情请参阅：https://github.com/openai/openai-python#microsoft-azure-openai

Anthropic

此后端使用 anthropic Python 包，并支持 magentic 的所有功能。

可以使用包含 anthropic 附加组件的 magentic 包进行安装，或直接安装 anthropic 包。

pip install "magentic[anthropic]"

然后导入 AnthropicChatModel 类。

from magentic.chat_model.anthropic_chat_model import AnthropicChatModel

model = AnthropicChatModel("claude-3-5-sonnet-latest")

LiteLLM

此后端使用 litellm Python 包，允许从 众多不同的提供商 查询大语言模型。注意：部分模型可能不支持 magentic 的所有功能，例如函数调用、结构化输出和流式响应。

可以使用包含 litellm 附加组件的 magentic 包进行安装，或直接安装 litellm 包。

pip install "magentic[litellm]"

然后导入 LitellmChatModel 类。

from magentic.chat_model.litellm_chat_model import LitellmChatModel

model = LitellmChatModel("gpt-4o")

Mistral

此后端使用 openai Python 包，并做了一些小修改以使 API 请求与 Mistral API 兼容。它支持 magentic 的所有功能。然而，工具调用（包括结构化输出）不会进行流式传输，而是会一次性接收全部结果。

注意：magentic 的未来版本可能会改用 mistral Python 包。

无需额外安装。只需导入 MistralChatModel 类即可。

from magentic.chat_model.mistral_chat_model import MistralChatModel

model = MistralChatModel("mistral-large-latest")

配置后端

magentic 默认使用的 ChatModel（在 @prompt、@chatprompt 等装饰器中）可以通过多种方式配置。当调用 prompt 函数或 chatprompt 函数时，所使用的 ChatModel 会按照以下优先级顺序确定：

1. 作为 model 参数传递给 magentic 装饰器的 ChatModel 实例
2. 使用 with MyChatModel: 创建的当前聊天模型上下文
3. 根据环境变量和 src/magentic/settings.py 中的默认设置创建的全局 ChatModel

以下代码片段演示了这一行为：

from magentic import OpenaiChatModel, prompt
from magentic.chat_model.anthropic_chat_model import AnthropicChatModel


@prompt("Say hello")
def say_hello() -> str: ...


@prompt(
    "Say hello",
    model=AnthropicChatModel("claude-3-5-sonnet-latest"),
)
def say_hello_anthropic() -> str: ...


say_hello()  # 使用环境变量或默认设置

with OpenaiChatModel("gpt-4o-mini", temperature=1):
    say_hello()  # 由于上下文管理器的存在，使用 openai 的 gpt-4o-mini 并设置 temperature=1
    say_hello_anthropic()  # 显式配置为 Anthropic claude-3-5-sonnet-latest

可以设置以下环境变量：

环境变量	描述	示例
MAGENTIC_BACKEND	作为 LLM 后端使用的包	anthropic / openai / litellm
MAGENTIC_ANTHROPIC_MODEL	Anthropic 模型	claude-3-haiku-20240307
MAGENTIC_ANTHROPIC_API_KEY	magentic 将使用的 Anthropic API 密钥	sk-...
MAGENTIC_ANTHROPIC_BASE_URL	兼容 Anthropic 的 API 的基础 URL	http://localhost:8080
MAGENTIC_ANTHROPIC_MAX_TOKENS	最大生成标记数	1024
MAGENTIC_ANTHROPIC_TEMPERATURE	温度	0.5
MAGENTIC_LITELLM_MODEL	LiteLLM 模型	claude-2
MAGENTIC_LITELLM_API_BASE	查询的基础 URL	http://localhost:11434
MAGENTIC_LITELLM_MAX_TOKENS	LiteLLM 最大生成标记数	1024
MAGENTIC_LITELLM_TEMPERATURE	LiteLLM 温度	0.5
MAGENTIC_MISTRAL_MODEL	Mistral 模型	mistral-large-latest
MAGENTIC_MISTRAL_API_KEY	magentic 将使用的 Mistral API 密钥	XEG...
MAGENTIC_MISTRAL_BASE_URL	兼容 Mistral 的 API 的基础 URL	http://localhost:8080
MAGENTIC_MISTRAL_MAX_TOKENS	最大生成标记数	1024
MAGENTIC_MISTRAL_SEED	确定性采样的种子	42
MAGENTIC_MISTRAL_TEMPERATURE	温度	0.5
MAGENTIC_OPENAI_MODEL	OpenAI 模型	gpt-4
MAGENTIC_OPENAI_API_KEY	magentic 将使用的 OpenAI API 密钥	sk-...
MAGENTIC_OPENAI_API_TYPE	允许选项：“openai”、“azure”	azure
MAGENTIC_OPENAI_BASE_URL	兼容 OpenAI 的 API 的基础 URL	http://localhost:8080
MAGENTIC_OPENAI_MAX_TOKENS	OpenAI 最大生成标记数	1024
MAGENTIC_OPENAI_SEED	确定性采样的种子	42
MAGENTIC_OPENAI_TEMPERATURE	OpenAI 温度	0.5

类型检查

许多类型检查器会对带有 @prompt 装饰器的函数发出警告或错误，因为这些函数没有函数体或返回值。有几种方法可以处理这种情况：

1. 在类型检查器中全局禁用该检查。例如，在 mypy 中通过禁用错误代码 empty-body 来实现。

   # pyproject.toml
   [tool.mypy]
   disable_error_code = ["empty-body"]
   

2. 将函数体写成 ...（这不符合 mypy 的要求）或 raise。

   @prompt("Choose a color")
   def random_color() -> str: ...
   

3. 在每个函数上使用注释 # type: ignore[empty-body]。在这种情况下，可以用文档字符串代替 ...。

   @prompt("Choose a color")
   def random_color() -> str:  # type: ignore[empty-body]
       """Returns a random color."""
   

Magentic 快速上手指南

Magentic 是一个用于将大语言模型（LLM）无缝集成到 Python 代码中的开源库。它通过 @prompt 和 @chatprompt 装饰器，让你能够定义返回结构化数据（如 Pydantic 模型）的函数，并结合传统 Python 代码构建复杂的智能体系统。

环境准备

在开始之前，请确保满足以下条件：

• Python 版本：建议安装 Python 3.10 或更高版本。
• API Key：你需要拥有支持的 LLM 提供商的 API Key（默认支持 OpenAI）。
  • 请在环境变量中设置 OPENAI_API_KEY。
  • 若使用 Anthropic 或 Ollama 等其他提供商，需在代码中进行相应配置。
• 依赖库：Magentic 底层依赖 pydantic 进行数据验证和结构化输出。

安装步骤

你可以使用 pip 或现代化的包管理工具 uv 进行安装。

方式一：使用 pip

pip install magentic

国内加速提示：如果遇到下载速度慢的问题，推荐使用国内镜像源：

pip install magentic -i https://pypi.tuna.tsinghua.edu.cn/simple

方式二：使用 uv (推荐)

uv add magentic

基本使用

Magentic 的核心是通过装饰器将 Python 函数转换为 LLM 调用。以下是最基础的使用示例。

1. 简单的文本生成 (@prompt)

使用 @prompt 装饰器定义一个提示词模板。函数参数会自动填入模板，返回值由 LLM 生成。注意：被装饰的函数不需要编写函数体。

from magentic import prompt


@prompt('Add more "dude"ness to: {phrase}')
def dudeify(phrase: str) -> str: ...  # 无需函数体


result = dudeify("Hello, how are you?")
print(result)
# 输出示例："Hey, dude! What's up? How's it going, my man?"

2. 结构化输出 (Pydantic Models)

Magentic 的强大之处在于它能直接返回符合类型注解的结构化数据（如 Pydantic 模型），而不仅仅是字符串。

from magentic import prompt
from pydantic import BaseModel


class Superhero(BaseModel):
    name: str
    age: int
    power: str
    enemies: list[str]


@prompt("Create a Superhero named {name}.")
def create_superhero(name: str) -> Superhero: ...


hero = create_superhero("Garden Man")
print(hero)
# 输出示例：Superhero(name='Garden Man', age=30, power='Control over plants', enemies=['Pollution Man', 'Concrete Woman'])
print(hero.power)
# 可以直接访问属性：Control over plants

3. 多轮对话与上下文 (@chatprompt)

如果需要设置系统角色（System Message）或提供少样本示例（Few-shot prompting），请使用 @chatprompt。

from magentic import chatprompt, AssistantMessage, SystemMessage, UserMessage
from pydantic import BaseModel


class Quote(BaseModel):
    quote: str
    character: str


@chatprompt(
    SystemMessage("You are a movie buff."),
    UserMessage("What is your favorite quote from Harry Potter?"),
    AssistantMessage(
        Quote(
            quote="It does not do to dwell on dreams and forget to live.",
            character="Albus Dumbledore",
        )
    ),
    UserMessage("What is your favorite quote from {movie}?"),
)
def get_movie_quote(movie: str) -> Quote: ...


quote = get_movie_quote("Iron Man")
print(quote)
# 输出示例：Quote(quote='I am Iron Man.', character='Tony Stark')

4. 函数调用 (Function Calling)

让 LLM 决定何时调用你的 Python 函数。Magentic 会返回一个 FunctionCall 对象，执行该对象即可运行实际逻辑。

from typing import Literal
from magentic import prompt, FunctionCall


def search_twitter(query: str, category: Literal["latest", "people"]) -> str:
    """Searches Twitter for a query."""
    return f"Twitter results for {query}"


@prompt(
    "Use the appropriate search function to answer: {question}",
    functions=[search_twitter],
)
def perform_search(question: str) -> FunctionCall[str]: ...


# 获取函数调用对象
call_obj = perform_search("What is the latest news on LLMs?")

# 执行实际的 Python 函数
result = call_obj() 
print(result)
# 输出：Twitter results for LLMs