Samuel Colvin 9分钟 2026/10/13
以 Markdown 查看
TL;DR
图库是个错误。你的代码本身就是图。
代码通过类型提示提供类型安全，通过 async 提供并行执行，通过沙箱外的 Temporal 和沙箱内的 Monty 提供持久化执行。
引言
最近关于图和循环与提示词之争作为驱动智能体方式的讨论很多。有人认为提示词已经过时：改为构建图和循环，让智能体从那里自我构建，图工程将在六个月内成为严肃智能体系统的默认架构。
我同意其中一半。人类编写代码的时代正在结束。开发者转而承担更难也更有趣的工作：把问题表述清楚。选择智能体应优化的指标和必须遵守的约束，编写衡量这些指标的 评估 ，然后让智能体能够快速轻松地迭代。这比编写代码本身更有价值。
另一半我认为是错的，而且我是以一个构建过图库的人的身份这么说的。大多数人所理解的图（把节点和边交给一个库，由它替你遍历）是个错误。唯一有效的图就是代码本身。
很长一段时间里，我都在想这是否只是出于我对 LangGraph 及其无视类型安全的强烈反感。现在我确信并非如此：在节点边界实现持久化并支持并行节点执行的图与类型安全从根本上不兼容，但反对图的理由远不止类型安全。
但当智能体在定义图，而图就是代码时，问题就变成了如何运行这段代码？这就是我们构建 Monty的原因，一个正是为此而写的沙箱化 Python 解释器。
下面是我所说的“图即代码”的意思。 reviewer 和 summarize 是子智能体，每个都是一个带类型化结果的异步函数，而这是调用它们的函数：
async def review_comments(comments: list[str]) -> str | None:
reviews = await asyncio.gather(*(reviewer(task=c) for c in comments))
issues = [r.issues for r in reviews if r.issues]
if issues:
return await summarize(issues)
扇出、汇合、分支。这就是一个图，同时也是一个 Python 函数： ty 检查 summarize 接受 issues，调用栈就是运行状态，模型可以在一次工具调用中写出它。这是纯粹、规范的 Python，因此智能体很容易编写。Monty 安全地运行它，并可以在每次外部函数调用处（本例中是reviewer 和 summarize ）进行快照和恢复。
每个图都在与调用栈对抗
你的程序已经有状态了：变量的值、执行到达的位置，以及等待返回的调用。图库为这个状态增加了另一种表示，然后你必须让它与代码保持一致。
if 是分支。 for 是循环。函数调用是边， except 是失败边， for 包裹 try 是重试，而 await asyncio.gather(...) 是扇出加汇合。调用栈记录了你在哪里，堆栈跟踪告诉你如何到达那里。它有调试器、性能分析器、类型检查器，而且每个程序员和每个模型都已经了解它。它的正式名称是结构化控制流。其他人就叫它代码。
图库把这一切重新构建了一遍，而且更糟。节点是你无法调用的函数。边是注册调用。状态是由 reducer 合并的字典，或是在你无法控制的时刻被修改的数据类，而不是局部变量。而“运行现在在哪里”由库自己的迭代器回答，而不是你的可观测性工具中的跟踪。
下面是 我们自己的文档中的扇出示例，特意挑选的：
import asyncio
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext, reduce_list_append
@dataclass
class SimpleState:
pass
async def main():
g = GraphBuilder(state_type=SimpleState, output_type=list[int])
@g.step
async def generate_list(ctx: StepContext[SimpleState, None, None]) -> list[int]:
return [1, 2, 3, 4, 5]
@g.step
async def square(ctx: StepContext[SimpleState, None, int]) -> int:
return ctx.inputs * ctx.inputs
collect = g.join(reduce_list_append, initial_factory=list[int])
g.add(
g.edge_from(g.start_node).to(generate_list),
g.edge_from(generate_list).map().to(square),
g.edge_from(square).to(collect),
g.edge_from(collect).to(g.end_node),
)
graph = g.build()
result = await graph.run(state=SimpleState())
print(sorted(result))
#> [1, 4, 9, 16, 25]
asyncio.run(main())
以及同样的程序：
import asyncio
async def square(n: int) -> int:
return n * n
async def main():
squares = await asyncio.gather(*(square(n) for n in [1, 2, 3, 4, 5]))
print(sorted(squares))
#> [1, 4, 9, 16, 25]
asyncio.run(main())
第二个没有添加任何东西就有类型，你可以在其中放一个 breakpoint() ，而且扇出只有一行。人类几秒钟就能读懂，模型也一样。第一个是我去年引以为豪的那种东西。
图仍然有用，它是你画在代码 之上 的东西，这就是为什么 pydantic-graph 会渲染 Mermaid。
对智能体来说更糟，因为智能体所需的控制流在运行前很少是已知的。那个论点在这一点上是对的：工作单元是一个带检查的循环，围绕任务设计。由人预先接好的固定图，不过是多了几步的提示词。
类型化、快照、并行：三选二
图也有一个 CAP 定理。任何构建图智能体的人都想要三样东西：
- 类型安全，让检查器能发现节点 A 交给节点 B 它无法接受的东西；
- 并行执行，这样你可以扇出和汇合；
- 图状态的快照，这样崩溃的运行可以从最后一个节点恢复，而不是从头开始。
我还没有找到能同时提供这三者的图 API，而且在尝试构建一个之后，我认为这不可能做到。
类型化意味着节点以普通对象读写状态，而节点的出边就是它的返回类型，因此类型检查器会同时检查两者。并行意味着同时有多个节点在运行。快照意味着在节点边界，整个运行是一个可序列化的值，从它恢复得到的结果与从未停止过相同。
在最初的 pydantic-graph 中，快照字面上就是 NodeSnapshot(state, next_node)：状态，加上我们在哪里。它是有类型的且可恢复的，因为“我们在哪里”就是一个节点。
加上并行之后，“我们在哪里”就变成了一组在途节点和半填充的汇合，而其中几个节点正在修改同一个状态对象。假设其中一个做了三次更新。写快照时，这三次中有多少次已经发生？从它恢复，第二次更新会运行两次，或者永远不会运行。除了没有任何在途节点的屏障处，不存在状态是一个连贯值的时刻，因此节点边界不再是运行边界。你可以只在屏障处快照，但那样一步之内的崩溃会丢失这一步，而一个在崩溃前写了一半更改的节点已经破坏了你即将保存的状态，且无法撤销，因为这些写入是就地进行的。你也不能简单地重新运行那些在途节点。运行器不知道某个节点是否接触了外部世界，或者做两次是否无害。它发了邮件或没发，而图无法分辨。
So to snapshot under parallelism, nodes must stop writing the state. They return a description of a change and the runtime applies it at a commit point. That is the step that costs you the types. Two nodes' changes arrive in no fixed order, so applying them has to be order-independent, which means a merge rule per field, declared outside the node code. The checker can verify the rule's signature. It can't verify the rule is commutative, and it can't verify the state's cross-field invariants hold after two partial updates land. More generally, any join of parallel branches that both touched state needs a semantic merge, and no type system checks a semantic merge. Joining outputs is fine, gather returns a typed tuple. Joining state is the problem.
Each side of the triangle is a real system.
Parallel and snapshotted, so the merge leaves the type system. That's LangGraph. Nodes return updates instead of mutating state, and a checkpoint is written after every super-step. When two nodes in the same step write the same key, a reducer declared in an Annotated type hint merges the writes, and without one the graph raises. It works. The state can be typed, and so can the reducer's signature, but the merge itself is out of reach. A node's write used to be a statement a type checker could read; now it's data.
Typed and snapshotted, so sequential. That was the original pydantic-graph, where FullStatePersistence saved the state and the one next node, and only one node ever ran at a time.
Typed and parallel, so no snapshots. That's pydantic-graph today. We looked for a long time for an API that stayed typed and could still be snapshotted, and every version either turned into reducer annotations or needed more generics than anyone would put up with. So pydantic_graph.persistence was removed in V2 and durability went to engines built for it. As David Montague, who designed the graph builder API, put it: "durable execution is the practical solution to most practical end-user challenges". What's left is a typed state machine with no durability and no snapshots of its own. Those come from Temporal replaying the code around it, and the graph machinery is, if anything, one more thing that has to replay deterministically.
Temporal escapes the triangle by never snapshotting application state at all. It records the result of each activity as it completes and re-runs your code to rebuild the state, so "how many of the three updates had happened" is answered by running the code again. An activity interrupted mid-flight may run twice, and the activity boundary is where you, not the runner, declare what touches the outside world and whether it's safe to retry.
Code dissolves the whole thing. In code the join of state is something you write, in the language: a, b = await gather(fa(), fb()), then the line that combines them. The merge is typed and ordered because you wrote it, and durability comes from an engine replaying that code. The graph library's problem was that it tried to do the merge for you, generically, and that's the part that can't be typed.
Monty is how the same answer works inside the sandbox, where there is no Temporal. The interpreter holds no operating system resources: no sockets, no file descriptors, no threads. Every parallel branch of the script is a host function the sandbox is waiting on, so at the moment it pauses, everything in flight is outside the interpreter and the interpreter itself, heap and call stack included, is a coherent value a few kilobytes long. That is the barrier the graph libraries couldn't find, and code gets it for free, because the only state is local variables and the only places anything crosses the boundary are the calls you wrote. A gather over three sub-agents pauses as one snapshot with three pending calls, and you resume it, in another process if you like, by handing back their results. The host function is still where you decide what touches the outside world and whether running it twice is safe, exactly like an activity. The join is a line of Python, so it is typed, and the state is the interpreter, so it snapshots.
Long live graphs
Prime Intellect rewrote Prime Agent from TypeScript to Rust this month. More than 2,000 agents did the work over two weeks, and the result starts roughly 14x faster. The orchestration: a root agent sorted the work into a topological order of tasks and wrote no product code. Each task went through a planner, an implementer, a reviewer, and a verifier, and a failed review or verification sent the task back to the implementer. They describe it as a factory of finite state machines. Look at the shape of it: a for over tasks, and inside each one a while with a try.
The question is what the agent writes that loop in, and how you run it. If a model is going to write Python that calls your sub-agents and your tools, you need to execute it without a human reading it first, and without handing over your filesystem and your network. That's Monty's job: the only way out of the sandbox is a function the host registered. There is no container to start and no network call, so the graph can be rewritten on every turn.
This is already how Pydantic AI Harness does multi-agent coordination. Each sub-agent is an async function. The orchestrator gets one tool, run_workflow, and writes the choreography:
import asyncio
reports = await asyncio.gather(
reviewer(task="Review auth.py for bugs:\n<file contents>"),
reviewer(task="Review parser.py for bugs:\n<file contents>"),
)
await summarizer(task="Summarize these review findings:\n" + "\n\n".join(reports))
Retries and voting are the same Python as everywhere else: a for around a try, a list comprehension over the results. Only the last value returns to the model's context. The script is type-checked against the sub-agent signatures first, so a misspelled function or a wrong-typed argument is rejected before any sub-agent runs.
The graph you want to look at is the one the run actually took, not the one someone drew before it. Sub-agents are Pydantic AI agents, so each run and model call is an OpenTelemetry span in the orchestrator's trace, in Logfire or any other backend. The fan-out, the join and every retry show up nested the way the code was written, which is why agent observability needs no graph visualizer: the trace is the drawing.
Durability is the same answer as before. Attach Temporal (or any other durable execution engine) to the orchestrator and to the sub-agents whose model calls should be activities, and the script the model wrote is replayed against the recorded sub-agent results. The script itself doesn't change.
Keep drawing graphs
Draw graphs. Put them in design docs, render them from code, use them to explain a run after the fact. Do not execute them. The runtime for a graph already exists, it's the one your language ships with, and when the author of the graph is a model, give it Monty rather than a library of nodes.
The human side of this, formulating the problem and setting up feedback, like evals that score every run, so the model's randomness can only push in the direction you want, is my talk at FastAPI Conf in Amsterdam on October 28.
See more from Pydantic in Google Search
Add Pydantic as a preferred source so our latest articles are easier to find.
Choose Pydantic as a preferred source on Google (opens in a new tab)Ready to see what your agents are actually doing?
Start free Start free Talk to us Talk to usExplore our open source packages
