Magentic：无缝集成LLM到Python函数中

Magentic 是一个 Python 库，旨在简化将大型语言模型（LLMs）集成到 Python 代码中的过程。它允许开发者使用 @prompt和 @chatprompt 装饰器创建函数，这些函数实际上是 LLM 的提示模板，并返回结构化的输出。

核心特性

  • 结构化输出： 支持使用 Pydantic 模型和 Python 内置类型来定义 LLM 函数的输出结构。
  • 流式输出： 支持流式传输结构化输出和函数调用，允许在生成过程中使用它们。
  • LLM辅助重试： 通过 LLM 辅助的重试机制，提高LLM生成符合复杂输出模式结果的可能性。
  • 可观测性： 利用 OpenTelemetry 提供了可观测性，并原生集成了 Pydantic Logfire。
  • 类型注解： 类型注解能与代码检查工具和 IDE 良好配合。
  • 灵活的配置： 支持多种 LLM 提供商，包括 OpenAI、Anthropic 和 Ollama。
  • 其他特性： 包括 Chat Prompting、并行函数调用、视觉能力、格式化和异步支持等。

安装

使用 pip 安装：

pip install magentic

或者使用 uv：

uv add magentic

安装完成后，需要配置 OpenAI API 密钥，可以通过设置 OPENAI_API_KEY 环境变量来实现。 如果使用其他 LLM 提供商，请参考官方文档的 Configuration 章节进行配置。

使用示例

@prompt 装饰器

@prompt 装饰器允许你将 LLM 提示定义为 Python 函数。当调用这个函数时，参数会被插入到提示模板中，然后发送给 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 装饰器

@chatprompt 装饰器与 @prompt 类似，但允许传递聊天消息作为模板，而不是单个文本提示。这可以用于提供系统消息或进行小样本提示，通过提供示例响应来指导模型的输出。

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>'

@prompt_chain 装饰器

@prompt_chain 装饰器会自动解析 FunctionCall 对象，并将输出传递回 LLM，直到获得最终答案。

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

流式输出

可以使用 StreamedStr 类来流式传输 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 ...'

异步

异步函数可以并发查询 LLM，显著提高生成速度。

import asyncio
from time import time
from typing import AsyncIterable
from magentic import prompt

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

# For each president listed, generate a description concurrently
start_time = time()
tasks = []
async for president in await iter_presidents():
    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)

后端/LLM 配置

Magentic 支持多个 LLM 提供商，可以通过配置进行切换。

  • OpenAI: 默认后端，使用 openai Python 包，支持所有 Magentic 的功能。
  • Ollama via OpenAI: 通过 OpenAI 兼容的 API 使用 Ollama 模型。
  • Anthropic: 使用 anthropic Python 包。
  • LiteLLM: 使用 litellm Python 包，支持多个 LLM 提供商。
  • Mistral: 使用修改后的 openai Python 包以兼容 Mistral API。

可以通过环境变量来配置 Magentic 使用的 LLM 后端和相关参数，例如：

  • MAGENTIC_BACKEND: 选择使用的后端 (anthropic/openai/litellm)。
  • MAGENTIC_OPENAI_MODEL: OpenAI model 名称 (e.g., gpt-4)。
  • MAGENTIC_OPENAI_API_KEY: OpenAI API Key。

具体支持的环境变量可以参考官方文档。

总结

Magentic 提供了一种简洁而强大的方式来集成 LLM 到 Python 代码中，它通过装饰器、结构化输出、流式处理和异步支持等特性，极大地简化了 LLM 应用的开发流程。无论是构建简单的文本生成应用还是复杂的 AI 代理，Magentic 都是一个值得尝试的工具。