Magentic：一行代码调用 LLM 就像调函数

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

使用 pydantic 模型和 Python 内置类型实现结构化输出。

支持流式传输结构化输出和函数调用，以便在生成过程中直接使用它们。

通过 LLM 辅助重试，提高 LLM 对复杂输出模式的遵循程度。

使用 OpenTelemetry 实现可观测性，并原生集成 Pydantic Logfire。

使用类型注解，与代码检查工具和 IDE 良好配合。

提供适用于多个 LLM 提供商的配置选项，包括 OpenAI、Anthropic 和 Ollama。

还有更多功能：聊天提示、并行函数调用、视觉、格式化、Asyncio……

pip install magentic

uv add magentic

通过设置 OPENAI_API_KEY 环境变量来配置 OpenAI API 密钥。要配置其他 LLM 提供商，请参阅“配置”了解更多信息。

@prompt 装饰器允许你以 Python 函数的形式定义大语言模型（LLM）提示词模板。调用此函数时，参数会被插入模板中，然后该提示词会被发送给 LLM，由 LLM 生成函数输出。

from magentic import prompt

@prompt('Add more "dude"ness to: {phrase}')
def dudeify(phrase: str) -> str: ...  # 没有函数体，因为它永远不会被执行

dudeify("Hello, how are you?")
# "Hey, dude! What's up? How's it going, my man?"

@prompt 装饰器会遵循被装饰函数的返回类型注解。它可以是 pydantic 支持的任意类型，包括 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: ...

create_superhero("Garden Man")
# Superhero(name='Garden Man', age=30, power='Control over plants', enemies=['Pollution Man', 'Concrete Woman'])

更多信息请参阅“结构化输出”。

@chatprompt 装饰器的工作方式与 @prompt 相同，但它允许你传入聊天消息作为模板，而不是单条文本提示词。你可以使用它提供系统消息，也可以进行少样本提示，即提供示例响应来引导模型输出。由花括号表示的格式字段 {example} 会在所有消息中填充（FunctionResultMessage 除外）。

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: ...

get_movie_quote("Iron Man")
# Quote(quote='I am Iron Man.', character='Tony Stark')

更多信息请参阅“聊天提示”。

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

from typing import Literal

from magentic import prompt, FunctionCall

def search_twitter(query: str, category: Literal["latest", "people"]) -> str:
    """Searches Twitter for a query."""
    print(f"Searching Twitter for {query!r} in category {category!r}")
    return "<twitter results>"

def search_youtube(query: str, channel: str = "all") -> str:
    """Searches YouTube for a query."""
    print(f"Searching YouTube for {query!r} in channel {channel!r}")
    return "<youtube results>"

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

output = perform_search("What is the latest news on LLMs?")
print(output)
# > FunctionCall(<function search_twitter at 0x10c367d00>, 'LLMs', 'latest')
output()
# > Searching Twitter for 'Large Language Models news' in category 'latest'
# '<twitter results>'

更多信息请参阅“函数调用”。

有时，LLM 需要进行一次或多次函数调用才能生成最终答案。@prompt_chain 装饰器会自动解析 FunctionCall 对象，并将输出传回 LLM，使其继续执行，直至得到最终答案。

在下面的示例中，调用 describe_weather 时，LLM 会先调用 get_current_weather 函数，然后使用其结果组织最终答案并返回。

from magentic import prompt_chain

def get_current_weather(location, unit="fahrenheit"):
    """Get the current weather in a given location"""
    # Pretend to query an API
    return {"temperature": "72", "forecast": ["sunny", "windy"]}

@prompt_chain(
    "What's the weather like in {city}?",
    functions=[get_current_weather],
)
def describe_weather(city: str) -> str: ...

describe_weather("Boston")
# 'The current weather in Boston is 72°F and it is sunny and windy.'

使用 @prompt、@chatprompt 和 @prompt_chain 创建的 LLM 驱动函数，可以像普通 Python 函数一样，作为函数提供给其他 @prompt/@prompt_chain 装饰器。这使你能够构建日益复杂的 LLM 驱动功能，同时还能独立测试和改进各个组件。

StreamedStr（以及 AsyncStreamedStr）类可用于流式传输 LLM 的输出。这样，你可以在文本生成过程中对其进行处理，而不必等待一次性接收完整输出。

from magentic import prompt, StreamedStr

@prompt("Tell me about {country}")
def describe_country(country: str) -> StreamedStr: ...

# Print the chunks while they are being received
for chunk in describe_country("Brazil"):
    print(chunk, end="")
# 'Brazil, officially known as the Federative Republic of Brazil, is ...'

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

from time import time

countries = ["Australia", "Brazil", "Chile"]

# Generate the descriptions one at a time
start_time = time()
for country in countries:
    # Converting `StreamedStr` to `str` blocks until the LLM output is fully generated
    description = str(describe_country(country))
    print(f"{time() - start_time:.2f}s : {country} - {len(description)} chars")

# 22.72s : Australia - 2130 chars
# 41.63s : Brazil - 1884 chars
# 74.31s : Chile - 2968 chars

# Generate the descriptions concurrently by creating the StreamedStrs at the same time
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)} chars")

# 22.79s : Australia - 2147 chars
# 23.64s : Brazil - 2202 chars
# 24.67s : Chile - 2186 chars

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

from collections.abc import Iterable
from time import time

from magentic import prompt
from pydantic import BaseModel

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

@prompt("Create a Superhero team named {name}.")
def create_superhero_team(name: str) -> Iterable[Superhero]: ...

start_time = time()
for hero in create_superhero_team("The Food Dudes"):
    print(f"{time() - start_time:.2f}s : {hero}")

# 2.23s : name='Pizza Man' age=30 power='Can shoot pizza slices from his hands' enemies=['The Hungry Horde', 'The Junk Food Gang']
# 4.03s : name='Captain Carrot' age=35 power='Super strength and agility from eating carrots' enemies=['The Sugar Squad', 'The Greasy Gang']
# 6.05s : name='Ice Cream Girl' age=25 power='Can create ice cream out of thin air' enemies=['The Hot Sauce Squad', 'The Healthy Eaters']

更多信息请参阅“流式传输”。

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

import asyncio
from time import time
from typing import AsyncIterable

from magentic import prompt

```python
@prompt("List ten presidents of the United States")
async def iter_presidents() -> AsyncIterable[str]: ...

@prompt("Tell me more about {topic}")
async def tell_me_more_about(topic: str) -> str: ...

# 对每个列出的总统并发生成描述
start_time = time()
tasks = []
async for president in await iter_presidents():
    # 使用 asyncio.create_task 在等待前安排协程执行
    # 这样描述生成会在总统列表仍在生成时开始
    task = asyncio.create_task(tell_me_more_about(president))
    tasks.append(task)

descriptions = await asyncio.gather(*tasks)

# 测量每秒字符数
total_chars = sum(len(desc) for desc in descriptions)
time_elapsed = time() - start_time
print(total_chars, time_elapsed, total_chars / time_elapsed)
# 24575 28.70 856.07

# 测量描述单个总统的每秒字符数
start_time = time()
out = await tell_me_more_about("George Washington")
time_elapsed = time() - start_time
print(len(out), time_elapsed, len(out) / time_elapsed)
# 2206 18.72 117.78

详见 Asyncio 文档。

@prompt 的 functions 参数可以包含异步/协程函数。当调用相应的 FunctionCall 对象时，结果必须等待。

Annotated 类型注解可用于为函数参数提供描述和其他元数据。详见 pydantic 文档中关于使用 Field 描述函数参数的部分。

@prompt 和 @prompt_chain 装饰器还接受 model 参数。你可以传递 OpenaiChatModel 实例来使用 GPT-4，或配置不同的 temperature。详见下文。

通过遵循 Pandas DataFrame 的示例笔记本，可以注册其他类型用作 @prompt 函数的返回类型注解。

后端/LLM 配置

Magentic 支持多个 LLM 提供商或"后端"。这大致是指使用哪个 Python 包来与 LLM API 交互。支持以下后端。

OpenAI

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

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

from magentic import OpenaiChatModel

model = OpenaiChatModel("gpt-4o")

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 后端时，设置 MAGENTIC_OPENAI_BASE_URL 环境变量或在代码中使用 OpenaiChatModel(..., base_url="http://localhost:8080") 允许你将 magentic 与任何 OpenAI 兼容的 API 一起使用，例如 Azure OpenAI Service、LiteLLM OpenAI Proxy Server、LocalAI。注意，如果 API 不支持函数调用，你将无法创建返回 Python 对象的提示函数，但 magentic 的其他功能仍然可用。

Azure

要在 openai 后端上使用 Azure，你需要将 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 extra 安装 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 包来启用来自许多不同提供商的 LLM 查询。注意：某些模型可能不支持 magentic 的所有功能，例如函数调用/结构化输出和流式处理。

使用 litellm extra 安装 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")

配置默认 ChatModel

magentic 使用的默认 ChatModel（在 @prompt、@chatprompt 等中）可以通过几种方式配置。当调用提示函数或 chatprompt 函数时，要使用的 ChatModel 遵循以下优先级顺序

提供给 magentic 装饰器的 model 参数的 ChatModel 实例

使用 with MyChatModel 创建的当前聊天模型上下文

全局 ChatModel，从环境变量和 src/magentic/settings.py 中的默认设置创建

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

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()  # 使用 env 变量或默认设置

with OpenaiChatModel("gpt-4o-mini", temperature=1):
    say_hello()  # 由于上下文管理器，使用带有 gpt-4o-mini 和 temperature=1 的 openai
    say_hello_anthropic()  # 使用 Anthropic claude-3-5-sonnet-latest，因为已明确配置

可以设置以下环境变量。

Type Checker 警告处理

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

全局禁用类型检查器的检查。 例如在 mypy 中通过禁用错误代码 empty-body。

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

让函数体为 ...（这不满足 mypy）或 raise。

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

在每个函数上使用注释 # type: ignore[empty-body]。 在这种情况下，你可以添加文档字符串而不是 ...。

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