跳到主要内容

Marvin — 总览：这是什么·全景·主线·阅读地图

30 秒导读： Marvin 是一个 Python 框架，让你用一句自然语言指令 + 一个 Python 类型，就能从 LLM 拿到经过校验、类型安全的结构化结果，并把这些调用编排成 agentic 工作流。它建立在 Pydantic 生态之上（用 pydantic-ai 真正跟模型对话），最巧的一招是：把"我要一个 int / 一个 Article"这件事，翻译成让模型去调用一个叫 MarkTaskSuccessful 的工具——拿结果 = 一次工具调用。

本章只做导览与全景：讲清"这是什么、大盘怎么转、一次 marvin.run 走完全程发生了什么"，并给出阅读地图。机制细节一律留给各章（下钻链接在每处点出）。

1. 这是什么（零基础也能懂）​

一句话定义： Marvin 把"向 LLM 要一个特定 Python 类型的答案"变成一个可声明、可观测、结果被校验过的任务（Task），再由一个编排器驱动模型直到任务完成。

解决什么问题 / 给谁用：

假设你在写 Python，想让 LLM 帮你"从这段乱七八糟的文字里抽出所有金额"，或者"把这句话分类成某个部门"，或者"研究一个主题、写一篇有标题和要点的文章"。你不想拿回一坨字符串然后自己解析、自己校验——你想直接拿到 list[int]、拿到一个 Enum 成员、拿到一个 Article 的 Pydantic 模型实例。Marvin 就是给这类人用的：需要把 LLM 的自由文本可靠地落成结构化、类型安全数据的 Python 工程师。

它能做什么（功能）：

  • 结构化输出工具：cast（转成某类型）、classify（归类）、extract（抽取）、generate（按描述造数据）、summarize、fn（把普通函数变成 AI 函数）。
  • agentic 控制流：Task（声明一个目标 + 结果类型）、Agent（可复用的 LLM 配置 + 工具 + 记忆）、run（跑一个任务）。
  • 编排与放大：Thread（共享上下文与历史，落到 SQLite）、plan / run_tasks（把复杂目标拆成有依赖的多任务）、Team / Swarm（多智能体协作，已标记弃用，见 04）、Memory（持久记忆）。

用起来什么样： 最小示例都在顶层包里（见 README.md）。

import marvin

# 1) 一句话跑一个任务，默认拿字符串
poem = marvin.run("Write a short poem about artificial intelligence")

# 2) 要一个具体类型——这是 Marvin 的核心卖点
answer = marvin.run("the answer to the universe", result_type=int)
print(answer)  # 42（一个真正的 int，不是 "42" 字符串）

# 3) 高层门面：转类型 / 归类 / 抽取 / 造数据
marvin.cast("the place with the best bagels", Location)   # -> {'lat':..., 'lon':...}
marvin.classify("shut up and take my money", SupportDepartment)  # -> SupportDepartment.SALES
marvin.extract("i found $30 ... bought 5 bagels for $10", int)   # -> [30, 10]

# 4) 显式 Task / 复用的 Agent
task = marvin.Task(instructions="Write a limerick about Python", result_type=str)
poem = task.run()

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

一句话直觉/类比： 把 Marvin 想成 LLM 版的"带类型标注的函数调用"。你写 result_type=int，就像给函数写返回类型注解；Marvin 负责让模型"填对返回值"，并在拿回来时用 Pydantic 校验。区别只是：中间那层不是编译器，而是一个被反复驱动的 LLM。

2. 顶层全景（它大概怎么转）​

怎么读这张图： 从上到下是"抽象层级"由高到低。你从最上面的一个便利函数进去，它层层归约成一个 Task，交给 Orchestrator 反复驱动一个 pydantic-ai 的 agentlet（一次性小 agent），中间的对话与结果落到 Thread/SQLite，途中的每个事件广播给 Handlers。

 高层门面  run / cast / classify / extract / generate / fn / summarize   （src/marvin/fns/*）
     │   全都归约成……
     ▼
   Task（目标 + result_type + tools）                          （src/marvin/tasks/task.py）
     │   run_tasks_async 交给……
     ▼
   Orchestrator —— 回合循环（while 未完成: run_once）           （src/marvin/engine/orchestrator.py）
     │   每回合向 Actor 要一个……
     ▼
   agentlet = pydantic_ai.Agent（临时构造，含工具 + EndTurn 输出类型） （src/marvin/agents/agent.py:get_agentlet）
     │   .iter() 驱动模型，事件流经……
     ├──► Handlers（打印 / 队列 / 自定义）                      （src/marvin/handlers/*）
     ▼
   Thread —— 对话历史 + LLM 调用记录，持久化到 SQLite            （src/marvin/thread.py, database.py）

部件一句话职责：

部件干什么在哪个文件（符号）
高层门面便利函数，每个都把"我要 X 类型"包成一个 Task 再跑src/marvin/fns/run.py（run/run_async）、fns/cast.py 等
Task声明式的一个目标：instructions + result_type + tools + 状态src/marvin/tasks/task.py（Task）
Orchestrator回合循环：收集就绪任务、装配工具、渲染系统提示、跑 agentlet、处理结束src/marvin/engine/orchestrator.py（Orchestrator.run / run_once）
Actor / Agent可复用的 LLM 配置；get_agentlet 现造一个 pydantic-ai agentsrc/marvin/agents/agent.py（Agent.get_agentlet）
agentlet真正跟模型对话的 pydantic-ai Agent，一回合用完即弃由 pydantic_ai.Agent(...) 构造，见 agent.py:211
EndTurn 工具把"结束回合并交结果"做成类型化工具，如 MarkTaskSuccessfulsrc/marvin/engine/end_turn.py（create_mark_task_successful）
Thread共享上下文/历史，with 作用域，持久化到 SQLitesrc/marvin/thread.py（Thread）、src/marvin/database.py
Handlers消费执行途中的事件（打印、入队、自定义）src/marvin/handlers/*、src/marvin/engine/events.py
defaults默认 agent / 模型 / 记忆提供方（可临时覆盖）src/marvin/defaults.py（defaults, override_defaults）

一句话记住两条暗线（详见 §4）：Marvin 自己几乎不跟模型说话——那是 pydantic-ai 干的，Marvin 是它的薄壳 + 编排层；而"当前是哪个 Thread / 哪个 Actor / 哪个 Orchestrator"这些不靠参数层层传递，靠 contextvars 隐式作用域。

3. 端到端主线走一遍（一次 marvin.run 从入口到结果）​

下面把 marvin.run("the answer to the universe", result_type=int) 从入口追到拿回 42。每步只点关键动作与源码位置，展开细节见对应章。

① 入口：便利函数 → 建一个 Task。 run 只是把同步壳套在 run_async 外面（src/marvin/fns/run.py:144 run → run_async）。run_async 做的第一件事就是把参数打成一个 Task：task = Task[result_type](instructions=..., result_type=result_type, ...)（fns/run.py:127），然后调 run_tasks_async([task], ...)（fns/run.py:134）。→ 细节见 01-task-and-result-type.md。

② 分流：独立多任务并发，否则进编排器。 run_tasks_async 判断任务之间有没有依赖：多个且互相独立就 asyncio.gather 并发；否则（单任务或有依赖）交给 Orchestrator（src/marvin/fns/run.py:34 run_tasks_async，独立性判定见 _tasks_are_independent）。单个 run 走的是后者。→ 见 02-orchestrator-turn-loop.md。

③ 回合循环开始。 Orchestrator.run 把自己塞进 _current_orchestrator 这个 contextvar，进入 while incomplete_tasks and turns < max_turns: 循环，每轮调一次 run_once（src/marvin/engine/orchestrator.py:240 run，循环体在 :277）。

④ 一回合内：装配工具 + 结果工具 + 系统提示。 run_once 挑出就绪任务、选一个 Actor，然后收集两类工具（src/marvin/engine/orchestrator.py:136 run_once）：

  • 普通工具：t.get_tools()；
  • 结束回合工具：t.get_end_turn_tools()——这里生成了那个关键的 MarkTaskSuccessful（src/marvin/tasks/task.py:517 get_end_turn_tools → mark_successful_tool）。

再渲染系统提示、把最近消息取出来当历史（Orchestrator._get_messages）。

⑤ 现造一个 agentlet，把"要什么类型"编码进它的 output_type。 actor.get_agentlet(tools=..., end_turn_tools=...) 构造一个临时的 pydantic_ai.Agent（src/marvin/agents/agent.py:139 get_agentlet）。关键：它把结束回合工具塞进 pydantic-ai 的 ToolOutput，作为 agent 的 output_type（agent.py:192）。"拿到 int 结果"于是变成"模型调用 MarkTaskSuccessful(result=42)"。 → 精华在 03-end-turn-and-agentlet.md。

⑥ 驱动模型，事件流经 handlers。 async with agentlet.iter(user_prompt, message_history=...) as run: 真正跟模型对话，handle_agentlet_events 把过程中的事件逐个 yield 给 orchestrator 的 handlers（src/marvin/engine/orchestrator.py:198）。这一步是 pydantic-ai 在干活，Marvin 只是转发事件。

⑦ 收尾：结果落进 task.result，历史落库。 模型调了 MarkTaskSuccessful 后，run.result.output 就是那个 EndTurn 实例；Orchestrator.end_turn 调它的 .run()（orchestrator.py:233），最终触发 mark_task.mark_successful(result) → self.result = 校验后的结果（src/marvin/engine/end_turn.py:54；src/marvin/tasks/task.py:549 mark_successful，self.result 在 :558）。同一回合还把新消息写进 Thread 并记一条 DBLLMCall（orchestrator.py:210-222）。

⑧ 返回。 循环发现任务已完成，退出。run_async 最后一行 return task.result（src/marvin/fns/run.py:141）把校验过的 42 交回给你。

一句话总结主线：便利函数 → Task（带 result_type）→ Orchestrator 回合循环 → 每回合现造 agentlet（把 result_type 变成 MarkTaskSuccessful 工具）→ pydantic-ai 驱动模型 → 模型调工具交结果 → 落进 task.result → 返回。

4. 全库最巧妙的一点，与两条暗线​

4.1 精华：把"拿类型结果"编码成一次 MarkTaskSuccessful 工具调用​

多数框架拿结构化输出，靠的是"让模型输出 JSON、再解析校验"。Marvin 换了个角度：为每个 Task 动态生成一个工具类，它的唯一参数 result 的类型，就是这个 Task 声明的 result_type。

# 摘自 create_mark_task_successful（end_turn.py:36），示意其骨架
@dataclass(kw_only=True)
class _MarkTaskSuccessful(MarkTaskSuccessful):
    result: mark_task.get_result_type()          # ← 参数类型 = 你的 result_type
    def __post_init__(self):
        mark_task.validate_result(self.result)   # ← 构造即用 Pydantic 校验
    async def run(self, thread, actor):
        await mark_task.mark_successful(self.result, thread=thread)  # ← 结果落库

于是"任务完成 + 交出类型正确的结果"这件事，被统一成模型主动调用一个工具。类型约束、Pydantic 校验、结果存储、回合结束，全挂在这一个工具上（真实实现 src/marvin/engine/end_turn.py:36 create_mark_task_successful；接进 agentlet 的 output_type 见 src/marvin/agents/agent.py:192）。这也解释了 README.md 里那些输出面板为什么总出现 MarkTaskSuccessful_xxxx —— 那就是模型在"交卷"。展开见 03-end-turn-and-agentlet.md。

4.2 暗线一：Marvin 是 pydantic-ai 的薄壳​

Marvin 不自己实现"给模型发消息、解析工具调用、跑工具"这套。真正跟模型对话的是 pydantic-ai 的 Agent（Marvin 里叫 agentlet），由 Agent.get_agentlet 现场构造（src/marvin/agents/agent.py:139），agentlet.iter(...) 驱动（src/marvin/engine/orchestrator.py:198）。Marvin 的增值在外面那一圈：Task 抽象、回合循环、EndTurn 工具、Thread 持久化、事件/handler。理解这条暗线，你就知道"某段逻辑该去 marvin 还是去 pydantic-ai 找"。

4.3 暗线二：用 contextvars 做隐式作用域​

"当前是哪个 Thread、哪个 Actor、哪个 Orchestrator、哪个 Task"这些上下文，Marvin 不靠参数一层层传，而是各自存在一个 ContextVar 里，靠 with 作用域压栈/弹栈：

上下文ContextVar进入方式位置
Thread_current_threadwith Thread():src/marvin/thread.py:98、__enter__ 在 :410
Actor_current_actorwith actor:src/marvin/agents/actor.py:25、:72
Orchestrator_current_orchestratorrun() 内 set/resetsrc/marvin/engine/orchestrator.py:45、:252
Task_current_taskTask 上下文src/marvin/tasks/task.py:48

这就是为什么 README.md 里 with marvin.Thread(): marvin.run(...); marvin.run(...) 能让两次调用自动共享历史——第二次 run 里 get_current_thread()（thread.py:430）拿到的正是外层那个 Thread。展开见 04-actors-threads-scaling.md。

5. 阅读地图（按顺序读，由浅入深）​

顺序章节讲什么什么时候读
101-task-and-result-type.mdTask 怎么把"要什么类型"声明出来、结果如何被校验想懂 Marvin 的核心抽象
202-orchestrator-turn-loop.mdOrchestrator 的回合循环：agentlet 怎样被一次次驱动想懂"控制流"怎么转
303-end-turn-and-agentlet.md精华：EndTurn 工具化 + agentlet 事件流想懂那个最巧的设计
404-actors-threads-scaling.md多智能体、依赖调度、记忆、Thread 持久化、contextvars想放大到复杂工作流
505-high-level-fns.mdcast/classify/extract/generate/fn 如何都归约成一个 Task想懂便利函数背后

建议路径： 只想会用 → 读本章 §1–§3 就够。想懂原理 → 顺着 01 → 02 → 03 读，这三章覆盖了主线上的每一步。想搭多智能体/长会话 → 加读 04。想扩展或看便利函数底层 → 05。

6. 代码地图（导航索引）​

用符号名 grep 定位比行号更稳（上游更新后行号会漂）。

主题文件路径符号名
顶层导出（所有公共 API）src/marvin/__init__.py__all__
便利函数入口 runsrc/marvin/fns/run.pyrun / run_async
多任务分流（并发 vs 编排）src/marvin/fns/run.pyrun_tasks_async / _tasks_are_independent
任务声明与结果类型src/marvin/tasks/task.pyTask / get_result_type / validate_result
结果落库src/marvin/tasks/task.pymark_successful
生成结束回合工具src/marvin/tasks/task.pyget_end_turn_tools / mark_successful_tool
回合循环src/marvin/engine/orchestrator.pyOrchestrator.run / run_once
收尾与结果处理src/marvin/engine/orchestrator.pyOrchestrator.end_turn
现造 agentlet（pydantic-ai）src/marvin/agents/agent.pyAgent.get_agentlet
精华：类型化结果工具src/marvin/engine/end_turn.pycreate_mark_task_successful / MarkTaskSuccessful
其它 EndTurn（失败/跳过/委派/规划）src/marvin/engine/end_turn.pycreate_mark_task_failed / create_delegate_to_actor / create_plan_subtasks
Thread + contextvar 作用域src/marvin/thread.pyThread / _current_thread / get_current_thread
持久化src/marvin/database.pyDBLLMCall / ensure_db_tables_exist
默认 agent / 模型src/marvin/defaults.pydefaults / override_defaults
事件与 handlersrc/marvin/engine/events.py、src/marvin/handlers/Event / Handler / AsyncHandler