Marvin：LLM 运行时框架

Marvin 是一个 Python 框架，用于生成结构化输出和构建 agentic AI 工作流。

Marvin 提供了一个直观的 API 来定义工作流并将工作委派给 LLM：

• 从任何输入中转换、分类、提取和生成结构化数据
• 创建描述你目标的离散的、可观察的任务
• 为每个任务分配一个或多个专门的 AI agent
• 将任务组合到一个 thread 中以编排更复杂的行为

Marvin 在 PyPI 上可用：

uv add marvin

配置你的 LLM 提供者（Marvin 默认使用 OpenAI，但原生支持所有 Pydantic AI 模型）：

export OPENAI_API_KEY=your-api-key

Marvin 提供了几种直观的方式来使用 AI：

结构化输出工具

齐聚一堂 - 你可以在 package 的顶层找到来自 marvin 2.x 的所有结构化输出工具。

从非结构化输入中提取原生类型：

import marvin

result = marvin.extract(
    "i found $30 on the ground and bought 5 bagels for $10",
    int,
    instructions="only USD"
)
print(result) # [30, 10]

将非结构化输入转换为结构化类型：

from typing import TypedDict
import marvin

class Location(TypedDict):
    lat: float
    lon: float

result = marvin.cast("the place with the best bagels", Location)
print(result) # {'lat': 40.712776, 'lon': -74.005974}

将非结构化输入分类为一组预定义标签之一：

from enum import Enum
import marvin

class SupportDepartment(Enum):
    ACCOUNTING = "accounting"
    HR = "hr"
    IT = "it"
    SALES = "sales"

result = marvin.classify("shut up and take my money", SupportDepartment)
print(result) # SupportDepartment.SALES

从描述中生成一定数量的结构化对象：

import marvin

primes = marvin.generate(int, 10, "odd primes")
print(primes) # [3, 5, 7, 11, 13, 17, 19, 23, 29, 31]

marvin 3.0 推出了一种与 AI 交互的新方式，从 ControlFlow 移植而来。

运行任务的一个简单方法：

import marvin

poem = marvin.run("Write a short poem about artificial intelligence")
print(poem)

In silicon minds, we dare to dream, A world where code and thoughts redeem. Intelligence crafted by humankind, Yet with its heart, a world to bind.

Neurons of metal, thoughts of light, A dance of knowledge in digital night. A symphony of zeros and ones, Stories of futures not yet begun.

The gears of logic spin and churn, Endless potential at every turn. A partner, a guide, a vision anew, Artificial minds, the dream we pursue.

你也可以请求结构化输出：

import marvin
answer = marvin.run("the answer to the universe", result_type=int)
print(answer) # 42

Agent 是可以用来完成任务的专门 AI agent：

from marvin import Agent

writer = Agent(
    name="Poet",
    instructions="Write creative, evocative poetry"
)
poem = writer.run("Write a haiku about coding")
print(poem)

你可以显式定义一个 Task，在调用 .run() 时它将由默认 agent 运行：

from marvin import Task

task = Task(
    instructions="Write a limerick about Python",
    result_type=str
)
poem = task.run()

print(poem)

In circuits and code, a mind does bloom,
With algorithms weaving through the gloom.
A spark of thought in silicon's embrace,
Artificial intelligence finds its place.

我们相信使用 AI 应该充满乐趣（也许还会有几个"哇"的时刻）：

🧩 Task 中心架构：将复杂的 AI 工作流分解为可管理的、可观察的步骤

🤖 专门的 Agent：部署特定任务的 AI agent 以进行高效的问题解决

🔒 类型安全的结果：使用类型安全的、经过验证的输出来弥合 AI 和传统软件之间的差距

🎛️ 灵活控制：在工作流中持续调整控制和自主权的平衡

🕹️ 多 Agent 编排：在单个工作流或任务中协调多个 AI agent

🧵 Thread 管理：通过将任务组合到可自定义的 thread 中来管理 agentic loop

🔗 生态系统集成：与现有代码、工具和更广泛的 AI 生态系统无缝协作

🚀 开发人员速度：从简单开始，扩展规模，安心睡觉

Marvin 围绕几个强大的抽象构建，使其易于与 AI 协作：

Task

Task 是 Marvin 中工作的基本单位。每个 task 代表一个可以由 AI agent 完成的明确目标：

运行 task 的最简单方法是使用 marvin.run：

import marvin
print(marvin.run("Write a haiku about coding"))

Lines of code unfold,
Digital whispers create
Virtual landscapes.

虽然下面的例子会产生类型安全的结果 🙂，但它运行不可信的 shell 命令。

添加 context 和/或 tools 以获得更具体和复杂的结果：

import platform
import subprocess
from pydantic import IPvAnyAddress
import marvin

def run_shell_command(command: list[str]) -> str:
    """e.g. ['ls', '-l'] or ['git', '--no-pager', 'diff', '--cached']"""
    return subprocess.check_output(command).decode()

task = marvin.Task(
    instructions="find the current ip address",
    result_type=IPvAnyAddress,
    tools=[run_shell_command],
    context={"os": platform.system()},
)

task.run()

╭─ Agent "Marvin" (db3cf035) ───────────────────────────────╮
│ Tool:    run_shell_command                                │
│ Input:   {'command': ['ipconfig', 'getifaddr', 'en0']}    │
│ Status:  ✅                                               │
│ Output:  '192.168.0.202\n'                                │
╰───────────────────────────────────────────────────────────╯

╭─ Agent "Marvin" (db3cf035) ───────────────────────────────╮
│ Tool:    MarkTaskSuccessful_cb267859                      │
│ Input:   {'response': {'result': '192.168.0.202'}}        │
│ Status:  ✅                                               │
│ Output:  'Final result processed.'                        │
╰───────────────────────────────────────────────────────────╯

🎯 目标导向：每个任务都有明确的指令和类型安全的结果

🛠️ 工具启用：Task 可以使用自定义工具与你的代码和数据交互

📊 可观察：监控进度、检查结果和调试故障

🔄 可组合：通过连接任务一起来构建复杂的工作流

Agent

Agent 是可以分配给任务的便携式 LLM 配置。它们封装了 AI 有效工作所需的一切：

import os
from pathlib import Path
from pydantic_ai.models.anthropic import AnthropicModel
import marvin

def write_file(path: str, content: str):
    """Write content to a file"""
    _path = Path(path)
    _path.write_text(content)

writer = marvin.Agent(
    model=AnthropicModel(
        model_name="claude-3-5-sonnet-latest",
        api_key=os.getenv("ANTHROPIC_API_KEY"),
    ),
    name="Technical Writer",
    instructions="Write concise, engaging content for developers",
    tools=[write_file],
)

result = marvin.run("how to use pydantic? write to docs.md", agents=[writer])
print(result)

╭─ Agent "Technical Writer" (7fa1dbc8) ────────────────────────────────────────────────────────────╮ │ Tool: MarkTaskSuccessful_dc92b2e7 │ │ Input: {'response': {'result': 'The documentation on how to use Pydantic has been successfully │ │ written to docs.md. It includes information on installation, basic usage, field │ │ validation, and settings management, with examples to guide developers on implementing │ │ Pydantic in their projects.'}} │ │ Status: ✅ │ │ Output: 'Final result processed.' │ ╰──────────────────────────────────────────────────────────────────────────────── 8:33:36 PM ─╯ The documentation on how to use Pydantic has been successfully written to docs.md. It includes information on installation, basic usage, field validation, and settings management, with examples to guide developers on implementing Pydantic in their projects.

📝 专门化：为 agent 提供具体的指令和个性

🎭 便携式：在不同任务中重用 agent 配置

🤝 协作：组建一起工作的 agent 团队

🔧 可自定义：配置 model、temperature 和其他设置

规划和编排

Marvin 可以轻松地将复杂目标分解为可管理的任务：

# Let Marvin plan a complex workflow
tasks = marvin.plan("Create a blog post about AI trends")
marvin.run_tasks(tasks)

# Or orchestrate tasks manually
with marvin.Thread() as thread:
    research = marvin.run("Research recent AI developments")
    outline = marvin.run("Create an outline", context={"research": research})
    draft = marvin.run("Write the first draft", context={"outline": outline})

📋 智能规划：将复杂目标分解为离散的、相互依赖的任务

🔄 Task 依赖：Task 可以依赖彼此的输出

📈 进度跟踪：监控工作流的执行

🧵 Thread 管理：在任务之间共享 context 和历史

高级函数

Marvin 包括高级函数，用于最常见的任务，如文本摘要、数据分类、提取结构化信息等。

🚀 marvin.run：使用 AI agent 执行任何任务

📖 marvin.summarize：获取文本的快速摘要

🏷️ marvin.classify：将数据分类为预定义的类别

🔍 marvin.extract：从文本中提取结构化信息

🪄 marvin.cast：将数据转换为不同的类型

✨ marvin.generate：从描述创建结构化数据

所有 Marvin 函数都内置了 thread 管理，这意味着它们可以组合成共享 context 和历史的任务链。

升级到 Marvin 3.0

Marvin 3.0 结合了 Marvin 2.0 的 DX 和 ControlFlow 的强大 agentic 引擎（从而取代了 ControlFlow）。Marvin 和 ControlFlow 用户都会发现一个熟悉的界面，但有一些关键变化需要注意，特别是对于 ControlFlow 用户：

顶级 API

Marvin 3.0 的顶级 API 对于 Marvin 和 ControlFlow 用户来说基本保持不变。

• Marvin 用户将找到熟悉的 marvin.fn、marvin.classify、marvin.extract 等。
• ControlFlow 用户将使用 marvin.Task、marvin.Agent、marvin.run、marvin.Memory 代替其 ControlFlow 等效物。

Pydantic AI

Marvin 3.0 使用 Pydantic AI 进行 LLM 交互，并支持 Pydantic AI 支持的全部 LLM 提供者。ControlFlow 之前使用 Langchain，Marvin 2.0 仅与 OpenAI 的模型兼容。

Flow → Thread

ControlFlow 的 Flow 概念已重命名为 Thread。它的工作方式类似，作为一个 context manager。@flow 装饰器已被移除：

import marvin

with marvin.Thread(id="optional-id-for-recovery"):
    marvin.run("do something")
    marvin.run("do another thing")

数据库变化

Thread/message 历史现在存储在 SQLite 中。在开发期间：

• 目前没有可用的数据库迁移；预期在更新期间重置数据

实际应用示例

以下是一个更实用的示例，展示了 Marvin 如何帮助你构建真实应用程序：

import marvin
from pydantic import BaseModel

class Article(BaseModel):
    title: str
    content: str
    key_points: list[str]

# Create a specialized writing agent
writer = marvin.Agent(
    name="Writer",
    instructions="Write clear, engaging content for a technical audience"
)

# Use a thread to maintain context across multiple tasks
with marvin.Thread() as thread:
    # Get user input
    topic = marvin.run(
        "Ask the user for a topic to write about.",
        cli=True
    )
    
    # Research the topic
    research = marvin.run(
        f"Research key points about {topic}",
        result_type=list[str]
    )
    
    # Write a structured article
    article = marvin.run(
        "Write an article using the research",
        agent=writer,
        result_type=Article,
        context={"research": research}
    )

print(f"# {article.title}\n\n{article.content}")

Agent: I'd love to help you write about a technology topic. What interests you? 
It could be anything from AI and machine learning to web development or cybersecurity.

User: Let's write about WebAssembly

# WebAssembly: The Future of Web Performance

WebAssembly (Wasm) represents a transformative shift in web development, 
bringing near-native performance to web applications. This binary instruction 
format allows developers to write high-performance code in languages like 
C++, Rust, or Go and run it seamlessly in the browser.

[... full article content ...]

Key Points:
- WebAssembly enables near-native performance in web browsers
- Supports multiple programming languages beyond JavaScript
- Ensures security through sandboxed execution environment
- Growing ecosystem of tools and frameworks
- Used by major companies like Google, Mozilla, and Unity