金融中的图工程：设计稳健的Agent图

✍ zostaff🕐 2026-07-25📦 28.2 KB 🟢 已读 𝕏 文章列表

本文探讨了金融领域中Agent图的设计与实现。文章指出，Agent图解决了脚本在等待、重启和并行分支上的缺陷，并通过状态图而非DAG来实现条件路由和循环。重点讨论了状态管理、并行拓扑模式（如Fan-out、Supervisor）以及在金融任务中的实际应用。

Agent 图工程状态管理并行计算金融 LangChain 拓扑模式

# Graph Engineering in Finance: Designing Agent Graphs That Do Not Fall Apart **作者**: zostaff **日期**: 2026-07-24T14:05:49.000Z **来源**: [https://x.com/zostaff/status/2080655698619142274](https://x.com/zostaff/status/2080655698619142274) --- ![Image](/doc/3424/image/img_80920a01389f.jpg) A script made of several agents breaks on three things: when one agent waits on another, when state has to survive a restart, and when you need six branches running in parallel. A graph solves these three not by being prettier, but by having a different execution model. By 2026 the industry converged on a specific definition, and it is no longer about "nodes and edges." An agent graph is an explicit directed graph with typed state, conditional routing, checkpointed execution at each node, and layered observability. Every word there carries weight, and most articles about graphs unpack none of them. This article is about the mechanics: how state works, what happens with parallel branches, why checkpointing is not the same as fault tolerance, and how all of it maps onto financial tasks. With code on an open stack, not on one proprietary runner. ## Why a graph, not a script and not a DAG Three levels that are often confused. A script is a sequence of calls. State lives in process variables, parallelism is done by hand, and a failure in one step takes down the whole pipeline. It works while you have two agents and they do not argue. A DAG in the classic sense (Airflow, Prefect) is an acyclic graph: nodes and dependencies, but cycles are forbidden by definition. For ETL that is correct, for agents it is not, because an agent loop contains a return by nature: checked, did not converge, went back and redid it. A state graph is what agents need. Nodes are functions over shared state, edges are transitions, and transitions can be conditional and cyclic. More than 70 percent of production agents use a graph structure precisely because real processes rarely run straight: they need branching, returns, and points of human intervention. ## State: where most systems break I will start with what usually comes last in graph articles, or not at all. According to LangChain's 2026 report on the state of agent engineering, more than 60 percent of production incidents are tied to state management. Not model quality, not prompts. State. State in a graph is a typed structure passed between nodes. Each node receives it, returns a partial update, and the framework applies the update by rules you defined. Those rules are called reducers, and the first trap hides in them. ``` from typing import Annotated, TypedDict from operator import add class FactorState(TypedDict): # without a reducer: the last write overwrites the previous one universe: list[str] # with the add reducer: results of parallel branches ACCUMULATE instead of overwriting factor_results: Annotated[list[dict], add] # error log, also accumulated errors: Annotated[list[str], add] ``` The difference between a field with a reducer and one without is the difference between working parallelism and lost data. If seven nodes compute seven factors at once and all write to a field without a reducer, you get the result of one random node and lose six. And silently: the graph completes, the report arrives, it simply lacks six factors. This is the classic race condition on parallel state updates, and it produces errors nearly impossible to catch from a stack trace, because there is no stack trace. Second thing about state: it must be narrow. The temptation to stuff everything into state ends with every node hauling megabytes and checkpoints bloating. The rule is simple: state holds what several nodes need. Everything else goes to storage, and state holds a reference. ## Topology: what to parallelize and what not to Graph topology is determined by data dependencies, not by a desire to go faster. Nodes that do not depend on each other by data run in parallel. Nodes where the next one reads the previous one's result run in sequence. This sounds trivial until you start working out what actually depends on what. Take a financial task: multi-factor analysis. Individual factor computations are independent: momentum does not read the value result, each computes its own slice from shared data. So they are parallel. But what follows is a chain with nothing to parallelize: validation reads all factors at once, the portfolio constructor reads the survivors of validation, risk decomposition reads the finished portfolio. Each next one cannot start until the previous is ready. ``` from langgraph.graph import StateGraph, START, END builder = StateGraph(FactorState) # parallel section: nodes are independent by data for name in ["momentum", "value", "quality", "low_vol"]: builder.add_node(name, make_factor_node(name)) builder.add_edge(START, name) # all start at once # synchronization point: the validator waits for ALL parallel branches builder.add_node("validator", validate_factors) for name in ["momentum", "value", "quality", "low_vol"]: builder.add_edge(name, "validator") # convergence = a sync barrier # sequential section: each reads the previous one's result builder.add_node("portfolio", construct_portfolio) builder.add_node("risk", decompose_risk) builder.add_edge("validator", "portfolio") builder.add_edge("portfolio", "risk") builder.add_edge("risk", END) graph = builder.compile(checkpointer=checkpointer) ``` Note the edges converging into validator. That is not just a connection, it is a synchronization barrier: the node will not start until every incoming branch finishes. This is exactly why the factor_results field must have the add reducer, otherwise six of seven results vanish at that barrier. ## Four topology patterns, and when to use each Topologies are not infinite. Four base patterns show up in production, and the choice between them is determined by who decides the next step. Show Image Fan-out with a barrier. One entry branches into independent nodes, all converge into an aggregator node. No routing decision is made at all, the structure is static. This is the cheapest pattern to debug, because the order is known in advance. It fits when the set of subtasks is fixed: compute seven factors, gather five data sources, check a document against four lists. Supervisor. A central node decides which worker to call next, and does so at every step by looking at state. Workers do not know about each other and return results to the supervisor. It fits when the set of steps is unknown in advance and depends on intermediate results: an analyst may request additional data if the initial set was insufficient. ``` def supervisor(state: FactorState) -> str: """Returns the next node name or END. The decision is made at every step.""" done = {r["name"] for r in state["factor_results"]} if "momentum" not in done: return "momentum" if state.get("needs_fundamentals") and "quality" not in done: return "quality" # branch arose from an intermediate result if len(done) >= state["min_factors"]: return "validator" return "widen_search" builder.add_conditional_edges("supervisor", supervisor, { "momentum": "momentum", "quality": "quality", "validator": "validator", "widen_search": "widen_search", }) # every worker returns to the supervisor instead of moving on by itself for worker in ["momentum", "quality", "widen_search"]: builder.add_edge(worker, "supervisor") ``` Hierarchy. A supervisor of supervisors: the top node distributes work between teams, each team being a subgraph with its own supervisor. Needed when there are more than a dozen nodes and a flat supervisor stops coping with the choice, because a description of all workers no longer fits its prompt. In finance this looks like a research team, a risk team, and an execution team, each with its own internal graph. Network. Any node can hand control to any other. Maximum flexibility and maximum unpredictability: such a graph is hard to debug and nearly impossible to cost out in advance. It is rarely used in production, and usually with a hard global step counter. The practical rule for choosing: start with a fan-out if the set of steps is fixed. Move to a supervisor when branching on intermediate results appears. Move to a hierarchy only when a flat supervisor has genuinely stopped choosing correctly. Do not take the network until you have proven the other three do not fit. ## Conditional routing: where a graph differs from a pipeline Direct edges give you a static pipeline. The real value of a graph is in conditional edges: the next node is chosen by a function of state at runtime. ``` def route_after_validation(state: FactorState) -> str: survivors = [f for f in state["factor_results"] if f["passed"]] if not survivors: return "no_signal" # nothing survived, empty day if len(survivors) < 3: return "widen_search" # too few signals, go back and widen return "portfolio" # enough, go build the portfolio builder.add_conditional_edges( "validator", route_after_validation, {"no_signal": END, "widen_search": "momentum", "portfolio": "portfolio"}, ) ``` The widen_search branch hands control back to a factor node. That is a cycle inside the graph, the thing forbidden in a classic DAG and necessary for agents. Here the graph does not merely execute a fixed order, it decides the route based on what actually came out. An important detail people forget: any cycle in a graph must have a counter and a cap. Without one, widen_search will spin until the budget runs out. The counter lives in state and is checked in the routing function. ## The human point: how to halt a graph and resume it Around 60 percent of production agent systems contain human intervention points, and in finance the share is higher, because a decision about capital or about refusing a client cannot be left to an opaque automaton. Technically this is not "the graph posted to Slack and waits" but an interruption mechanism built into execution. It works like this: a node raises an interrupt, the graph saves state to a checkpoint and ends execution. The process can die, the server can restart. When the human answers, the graph resumes from the same place with the answer injected. ``` from langgraph.types import interrupt, Command def approval_gate(state: FactorState) -> dict: portfolio = state["portfolio"] # execution STOPS here, state goes to the checkpoint decision = interrupt({ "question": "Approve the portfolio for execution?", "gross_exposure": portfolio["gross"], "largest_position": portfolio["max_weight"], "factors_used": [f["name"] for f in state["factor_results"] if f["passed"]], }) if decision["approved"]: return {"status": "approved", "approved_by": decision["user"]} return {"status": "rejected", "reason": decision.get("reason", "")} # later, possibly in another process and an hour later: graph.invoke( Command(resume={"approved": True, "user": "pm@fund"}), config={"configurable": {"thread_id": "factors-2026-07-24"}}, ) ``` The key point is that resumption goes by thread identifier, not by an object in memory. The interrupt survives a process restart because state lives in the checkpointer. This is exactly what separates a real human point from a blocking wait in code, which survives neither a deploy nor a crash. Three places where a human point is mandatory in financial graphs: before executing a trade or moving capital, before refusing a client in a regulated process, and before any action whose rollback costs more than waiting. Everything else gets automated, but not those three. Separately, on what to show the human. The interrupt should carry not a raw state dump but a compressed summary of the decision: what is proposed, on what grounds, which alternatives were discarded. A human shown ten kilobytes of JSON will click approve without looking, and the control point becomes a rubber stamp. ## Checkpointing: recovery, not prevention Checkpointing is saving state after each node to durable storage. It gives you three things: recovery from a crash at the last point, a pause for human intervention, and the ability to rewind execution for debugging. ``` from langgraph.checkpoint.postgres import PostgresSaver checkpointer = PostgresSaver.from_conn_string(DB_URL) graph = builder.compile(checkpointer=checkpointer) # each run lives in its own thread; you can resume by it config = {"configurable": {"thread_id": "factors-2026-07-24"}} result = graph.invoke({"universe": tickers}, config=config) ``` And here is the honest distinction almost nobody states. Checkpointing is a mechanism for recovering after a failure, not for preventing one. It returns you to the state before the node fell over, but does nothing to stop the node from falling over again for the same reason. If a node fails on a hit rate limit, resuming from the checkpoint reproduces the failure. Prevention is separate work: concurrency limits, retries with exponential backoff, degradation to cache. Hence the practical rule: checkpointing is mandatory but insufficient. Failure has to be handled at the node level, and the graph must be able to continue without the failed branch when the task allows it. ``` def make_factor_node(name: str): def node(state: FactorState) -> dict: try: result = compute_factor(name, state["universe"]) return {"factor_results": [{"name": name, "passed": True, **result}]} except RateLimitError as e: # degradation: the branch does not kill the graph, it marks itself as failed return { "factor_results": [{"name": name, "passed": False, "reason": "rate_limit"}], "errors": [f"{name}: {e}"], } return node ``` The key point: the node does not raise outward, it returns a flagged result. The graph continues with six factors instead of seven, and the validator sees from the flag that one is missing and accounts for it. The failure is scoped to the node instead of taking down the pipeline, and that is exactly the property graphs exist for. ## Debugging: how to fix a graph that broke Debugging a graph differs from debugging a script in that there is often no stack trace. A node ran, returned something wrong, the graph moved on, and the error surfaced three nodes later. Three tools make this tractable. First: reading checkpoints. Since state is saved after every node, you can pull the run history and see what state looked like at each point. This answers "where exactly did the data go wrong." ``` # state history for the thread, latest first for snapshot in graph.get_state_history(config): print(snapshot.next, len(snapshot.values.get("factor_results", []))) # you can see at which node the factor count stopped growing ``` Second: rewinding and replaying. You can take a checkpoint from mid-run, modify state, and continue from there with different data. This lets you test the hypothesis "if the validator had received this input, it would have decided correctly" without rerunning the whole graph from scratch and without paying for all preceding nodes again. ``` # take a specific checkpoint and replay from it with a state edit target = list(graph.get_state_history(config))[3] graph.update_state(target.config, {"min_factors": 2}) # change the condition graph.invoke(None, target.config) # continue from this point ``` Third: node tracing. Every node records input, output, duration, and cost. Without it the cost of a run stays a mystery until the bill, and a slow node cannot be found. Separately, a class of error caught only by checkpoints: races on parallel branches. The symptom is characteristic - the run completed without a single error, but data is missing from the result, and a repeat run gives a different result on the same inputs. Nondeterminism in the absence of exceptions is almost always a lost state update, that is, a forgotten reducer. ## Testing the topology before production A graph can and should be tested without a single model call. Topology is structure, and its correctness is verified separately from node contents. ``` def fake_node(name, payload): """A stub instead of a real node: makes no model call, returns a fixed answer.""" def node(state): return {"factor_results": [{"name": name, "passed": True, **payload}]} return node def test_barrier_collects_all_branches(): """Check that the barrier collects ALL parallel branches, not just the last.""" b = StateGraph(FactorState) names = ["momentum", "value", "quality", "low_vol"] for n in names: b.add_node(n, fake_node(n, {"sharpe": 1.0})) b.add_edge(START, n) b.add_node("collect", lambda s: {}) for n in names: b.add_edge(n, "collect") b.add_edge("collect", END) out = b.compile().invoke({"universe": [], "factor_results": [], "errors": []}) assert len(out["factor_results"]) == 4, "branch loss: check the reducer" def test_cycle_terminates(): """Check that a cycle with a return does not spin forever.""" state = {"attempts": 0, "factor_results": [], "errors": [], "universe": []} for _ in range(100): route = route_after_validation(state) if route in ("portfolio", "no_signal"): break state["attempts"] += 1 assert state["attempts"] < 10, "cycle without a cap" ``` The first test catches a forgotten reducer, the most expensive error in a graph, in milliseconds and with no model calls. The second verifies the cycle terminates. Both run in ordinary CI on every commit, and this is the only way to learn about branch loss before it happens on a real run with real money. What else to test: that conditional routing covers every possible value (there is no state for which the function returns the name of a nonexistent node), that the graph reaches the end when one node fails, and that state serializes into the checkpointer without loss. ## Splitting models across nodes Graph nodes do not have to run on one model, and this is a direct lever on cost and quality. A cheap fast model on the many uniform nodes, a strong model with high reasoning effort on the judge nodes. ``` FAST = ChatAnthropic(model="claude-sonnet-4-5", temperature=0) STRONG = ChatAnthropic(model="claude-opus-4-5", temperature=0) # seven uniform computations: the cheap model def make_factor_node(name): ... # uses FAST # the validator delivers a verdict: strong model, different instructions def validate_factors(state): ... # uses STRONG ``` The logic here is not that different models catch different errors, though that is true. The logic is that the judge node must not be the same agent that produced the work, otherwise it confirms its own conclusions. Splitting across nodes makes that separation structural rather than declarative: the judge is physically a different node, with a different prompt, on a different model, reading only the output and not the producer's reasoning. ## Data: what separates a financial graph from any other Here begins the domain specificity that makes a financial graph impossible to assemble from a generic tutorial. First and foremost: a node has no right to see the future relative to the decision point. In an ordinary pipeline you simply read a table. In a financial node, reading a table without a time filter is look-ahead bias, and it will not surface as an error, it will surface as an excellent backtest and a loss in production. The defense is built at the state level: the cutoff timestamp lives in state, and every node is obliged to apply it. ``` class FinState(TypedDict): as_of: datetime # cutoff point, shared across the whole run universe: list[str] factor_results: Annotated[list[dict], add] data_gaps: Annotated[list[str], add] def load_prices(state: FinState) -> dict: # the as_of filter is applied IN THE QUERY, not after loading df = db.query( "SELECT ts, ticker, close FROM prices WHERE ts <= :cutoff AND ticker IN :u", cutoff=state["as_of"], u=tuple(state["universe"]), ) return {"prices": df} ``` The filter sits in the query deliberately. If you load everything and trim in memory, sooner or later someone adds a node that reads the untrimmed frame, and the future leaks back in. A time cutoff must be impossible to bypass by carelessness. Second: filings are published with a delay and later revised. A quarterly report becomes known not on the quarter-end date but weeks later, and then the numbers get corrected. A node that takes the final version of a metric and attributes it to the quarter-end date is also looking into the future, just less obviously. The correct approach stores not only the value but the moment it became publicly known, and filters on that. Third: data arrives partially, and this is normal rather than a failure. One source responded, another hit a limit, a third returned gaps in the middle. The graph must distinguish three states: no data, data present, data partially present. The third is the most dangerous, because a node will compute a factor on an incomplete sample and return a confident number. ``` def make_factor_node(name: str, min_coverage: float = 0.9): def node(state: FinState) -> dict: data = state["prices"] coverage = data["ticker"].nunique() / len(state["universe"]) if coverage < min_coverage: # an honest refusal instead of a confident number on holey data return { "factor_results": [{"name": name, "passed": False, "reason": "coverage"}], "data_gaps": [f"{name}: coverage {coverage:.0%} below threshold"], } return {"factor_results": [{"name": name, "passed": True, **compute(name, data)}]} return node ``` The coverage threshold is the case where an explicit refusal is more useful than an answer. A factor computed on sixty percent of the universe is not merely less accurate, it is systematically biased, because missing data is almost never missing at random: what drops out tends to be illiquid, newly listed, or distressed names, precisely the ones carrying signal. Fourth: the cutoff point must be part of the run identifier. A checkpointer thread named by date and cutoff gives reproducibility: the same run on the same data can be pulled up and repeated, and that is what gets asked for in any review of a disputed decision. ## Applied to finance: three real graphs Now the specifics of where this pays off in financial tasks. Three topologies seen in production. > The research fan-out. Many parallel nodes gather different slices (news, filings, price series, macro) and converge into a synthesizer node. Topology: a wide parallel layer plus a barrier. Main risk: races in state, solved by reducers. The pipeline with a debate. Analysts prepare theses, then two nodes argue from opposing positions (bulls against bears), an arbiter node decides, then risk control. This is the architecture of TradingAgents, an open framework on LangGraph where the pipeline runs analysts - researchers - trader - risk management - portfolio manager. Main risk: an endless debate, solved by a round counter in state. The compliance pipeline. Extraction from documents, watchlist screening, adverse-media check, validation, a human decision point. Here what matters most is not speed but auditability: each node writes its justification into state, checkpoints are retained, and any decision can be pulled up and explained. In regulated tasks an opaque system is unusable regardless of accuracy. The common denominator of the three: parallelism where there is no data dependency, a barrier before the node that reads everything, and a mandatory human point where the cost of error is high. Around 60 percent of production agent systems add human intervention points, and in finance that share is higher for regulatory reasons. ## Observability and cost A graph without tracing is a black box where debugging reduces to guessing. The minimum you need: a trace per node (input, output, duration, cost), metrics for the run as a whole, and an alert on budget overrun. On cost, honestly: a parallel fan-out of seven nodes is seven simultaneous model calls, and a cycle with a return multiplies that by the number of iterations. The budget has to be computed before the run, not after. The practical approach: a hard iteration cap on every cycle, a cost ceiling per run, and degradation to a cheaper model as the limit approaches. A graph that can spin without a cap will one day spin without a cap. ## The two-layer architecture for long processes The last thing separating production from a demo. A graph framework is good when a run takes minutes and lives inside one service. When a process runs for hours, involves external services, and must survive an infrastructure restart, the industry uses two layers: a durable execution layer on the outside (Temporal and similar) and the graph layer inside. The division of duties is simple. The outer layer is responsible for the process reaching the end despite crashes, restarts, and network failures, and for coordinating several services. The inner graph is responsible for logic: routing, state, conditional transitions within one task. For financial processes that run overnight and touch several systems, this is the standard pairing. If a run fits in minutes and lives in one process, the second layer is unnecessary and built-in checkpointing is enough. Adding it in advance is complexity without benefit. ## Anti-patterns: how not to design a graph Five design errors that come up most often. Each looks reasonable at the moment the decision is made. > A graph where a function would do. If the steps are fixed, there is no branching, and state does not outlive the run, a graph adds a checkpointer, serialization, and a layer of abstraction for nothing. The tell: the graph has not a single conditional edge and not a single cycle. The cure: an ordinary function. The state dump. Everything gets stuffed into state, including raw dataframes and full model responses. Checkpoints bloat, serialization slows, and every node receives megabytes from which it reads three fields. The cure: references and metadata in state, heavy objects in storage. False parallelism. Nodes are spread across parallel branches, but each reads its neighbor's result through shared storage. Formally parallel, actually a race with a nondeterministic result. The tell: node order affects the outcome even though they are marked independent. The cure: if there is a data dependency, it is a sequential edge, not a parallel branch. A node that does three things. One node pulls data, computes a factor, and writes a report. On failure it is unclear which part broke, and on resume from a checkpoint all three operations repeat, including the ones that already succeeded. The cure: a node is one operation with one clear output. The judge inside the maker. The same node first generates, then checks itself against its own criteria. Formally a check exists, in practice it does not, because a producer is too generous toward its own output. The cure: the judge is a separate node reading only the result. A shared diagnostic for the first three: if you cannot draw the graph on paper in a minute, the topology is too complex for the task. A graph is first of all a model you hold in your head, and if it cannot be held, it cannot be debugged either. ## Build order The technical order that works: first define state with explicit types and reducers for parallel fields, then build topology from data dependencies, then add conditional routing with cycle caps, then the checkpointer, then node-level failure handling, then tracing and budget. Each step catches its class of problems: typed state catches races, data-driven topology catches false parallelism, caps catch infinite cycles, the checkpointer gives recovery, failure handling stops one node from taking down the run, and tracing makes debugging possible. ## 相关链接 - [zostaff](https://x.com/zostaff) - [@zostaff](https://x.com/zostaff) - [8.5K](https://x.com/zostaff/status/2080655698619142274/analytics) - [Upgrade to Premium](https://x.com/i/premium_sign_up) - [10:05 PM · Jul 24, 2026](https://x.com/zostaff/status/2080655698619142274) - [8,566 Views](https://x.com/zostaff/status/2080655698619142274/analytics) - [View quotes](https://x.com/zostaff/status/2080655698619142274/quotes) --- *导出时间: 2026/7/25 16:35:56* --- ## 中文翻译 # 金融领域的图工程：设计不会崩溃的智能体图 **作者**: zostaff **日期**: 2026-07-24T14:05:49.000Z **来源**: [https://x.com/zostaff/status/2080655698619142274](https://x.com/zostaff/status/2080655698619142274) --- ![Image](/doc/3424/image/img_80920a01389f.jpg) 一个由多个智能体组成的脚本会在三件事上崩溃：当一个智能体等待另一个智能体时，当状态需要在重启后存活时，以及当你需要六个分支并行运行时。图解决这三个问题不是因为它更漂亮，而是因为它有不同的执行模型。到了 2026 年，行业对特定的定义达成了一致，它不再关乎“节点和边”。智能体图是一个带有类型化状态、条件路由、每个节点检查点执行以及分层可观测性的显式有向图。那里的每一个词都举足轻重，而大多数关于图的文章都没有对它们进行拆解。本文是关于机制的：状态如何工作，并行分支会发生什么，为什么检查点不等于容错，以及所有这些如何映射到金融任务。使用开源技术栈上的代码，而不是某个专有的运行器。 ## 为什么是图，而不是脚本，也不是 DAG 经常被混淆的三个层次。脚本是调用的序列。状态存在于进程变量中，并行性是手工完成的，一个步骤中的故障会导致整个管道崩溃。当你有两个智能体且它们不争吵时，它就能工作。经典意义上的 DAG（Airflow, Prefect）是一个无环图：节点和依赖关系，但根据定义循环是被禁止的。对于 ETL（提取、转换、加载）这是正确的，对于智能体则不然，因为智能体循环本质上包含一个返回：已检查，未收敛，回去重做。状态图正是智能体所需要的。节点是对共享状态的函数，边是转换，而转换可以是条件性的和循环的。超过 70% 的生产环境智能体使用图结构，正是因为真实的过程很少直线运行：它们需要分支、返回和人为干预的点。 ## 状态：大多数系统崩溃的地方我将从在图文章中通常排在最后或根本不出现的内容开始。根据 LangChain 2026 年关于智能体工程状态的报告，超过 60% 的生产事故与状态管理有关。不是模型质量，不是提示词。是状态。图中的状态是在节点之间传递的类型化结构。每个节点接收它，返回部分更新，框架根据你定义的规则应用更新。这些规则被称为 reducers（规约器），第一个陷阱就隐藏在其中。 ```python from typing import Annotated, TypedDict from operator import add class FactorState(TypedDict): # 没有 reducer：最后一次写入覆盖前一次 universe: list[str] # 使用 add reducer：并行分支的结果会累积而不是覆盖 factor_results: Annotated[list[dict], add] # 错误日志，也是累积的 errors: Annotated[list[str], add] ``` 带有 reducer 的字段和不带有 reducer 的字段之间的区别，在于正常工作的并行性和数据丢失之间的区别。如果七个节点一次计算七个因子并且都写入一个没有 reducer 的字段，你会得到一个随机节点的结果并丢失另外六个。而且是静默地：图完成了，报告到了，只是少了六个因子。这是并行状态更新中的经典竞态条件，它产生的错误几乎不可能从堆栈跟踪中捕获，因为没有堆栈跟踪。关于状态的第二点：它必须狭窄。将所有东西塞进状态的诱惑最终会导致每个节点拖动兆字节的数据，检查点膨胀。规则很简单：状态保存几个节点需要的东西。其他所有东西都去存储，状态只保存引用。 ## 拓扑：什么要并行化，什么不要图拓扑由数据依赖决定，而不是由加快速度的愿望决定。彼此之间没有数据依赖的节点并行运行。下一个节点读取上一个节点结果的节点按顺序运行。这听起来微不足道，直到你开始弄清楚实际上什么依赖于什么。以一项金融任务为例：多因子分析。单个因子的计算是独立的：动量不读取价值因子的结果，每个都从共享数据中计算自己的部分。所以它们是并行的。但是接下来是一个没有任何东西可以并行化的链：验证一次性读取所有因子，投资组合构建器读取验证的幸存者，风险分解读取完成的组合。每一个下一步直到前一步准备好才能开始。 ```python from langgraph.graph import StateGraph, START, END builder = StateGraph(FactorState) # 并行部分：节点在数据上是独立的 for name in ["momentum", "value", "quality", "low_vol"]: builder.add_node(name, make_factor_node(name)) builder.add_edge(START, name) # 全部同时开始 # 同步点：验证器等待所有并行分支 builder.add_node("validator", validate_factors) for name in ["momentum", "value", "quality", "low_vol"]: builder.add_edge(name, "validator") # 收敛 = 一个同步屏障 # 顺序部分：每个都读取前一个的结果 builder.add_node("portfolio", construct_portfolio) builder.add_node("risk", decompose_risk) builder.add_edge("validator", "portfolio") builder.add_edge("portfolio", "risk") builder.add_edge("risk", END) graph = builder.compile(checkpointer=checkpointer) ``` 注意汇聚到 validator 的边。这不仅仅是一个连接，它是一个同步屏障：直到每个传入的分支完成，节点才会开始。这正是 factor_results 字段必须具有 add reducer 的原因，否则七个结果中有六个会在该屏障处消失。 ## 四种拓扑模式以及何时使用每种模式拓扑不是无限的。生产中出现了四种基本模式，它们之间的选择由谁来决定下一步决定。显示图片带有屏障的扇出。一个入口分支到独立的节点，全部汇聚到一个聚合器节点。根本不做路由决策，结构是静态的。这是调试起来最便宜的模式，因为顺序是预先知道的。它适用于子任务集是固定的情况：计算七个因子，收集五个数据源，根据四个列表检查文档。主管。一个中心节点决定下一步调用哪个工作者，并且每一步都通过查看状态来做。工作者彼此不了解并将结果返回给主管。它适用于步骤集预先未知且取决于中间结果的情况：如果初始集不足，分析师可能会请求额外数据。 ```python def supervisor(state: FactorState) -> str: """返回下一个节点名称或 END。决策在每一步做出。""" done = {r["name"] for r in state["factor_results"]} if "momentum" not in done: return "momentum" if state.get("needs_fundamentals") and "quality" not in done: return "quality" # 分支源于中间结果 if len(done) >= state["min_factors"]: return "validator" return "widen_search" builder.add_conditional_edges("supervisor", supervisor, { "momentum": "momentum", "quality": "quality", "validator": "validator", "widen_search": "widen_search", }) # 每个工作者都返回主管而不是自己继续移动 for worker in ["momentum", "quality", "widen_search"]: builder.add_edge(worker, "supervisor") ``` 层次结构。主管的主管：顶层节点在工作组之间分配工作，每个工作组是一个带有自己主管的子图。当有超过十几个节点且扁平主管停止能够做出选择时需要，因为对所有工作者的描述不再适合放入其提示词中。在金融领域，这看起来像一个研究团队、一个风险团队和一个执行团队，每个都有自己的内部图。网络。任何节点都可以将控制权交给任何其他节点。最大的灵活性和最大的不可预测性：这样的图很难调试，而且几乎不可能预先计算成本。它在生产中很少使用，通常带有一个硬性的全局步数计数器。选择的实用规则：如果步骤集是固定的，从扇出开始。当出现基于中间结果的分支时，转移到主管。只有当扁平主管真的停止正确选择时，才转移到层次结构。直到你证明其他三个不适合之前，不要采用网络。 ## 条件路由：图与管道的区别直接边给你一个静态管道。图的真正价值在于条件边：下一个节点由运行时的状态函数选择。 ```python def route_after_validation(state: FactorState) -> str: survivors = [f for f in state["factor_results"] if f["passed"]] if not survivors: return "no_signal" # 没有任何东西幸存，空日子 if len(survivors) < 3: return "widen_search" # 信号太少，回去拓宽 return "portfolio" # 足够了，去构建投资组合 builder.add_conditional_edges( "validator", route_after_validation, {"no_signal": END, "widen_search": "momentum", "portfolio": "portfolio"}, ) ``` widen_search 分支将控制权交回给一个因子节点。这是图中的一个循环，这是经典 DAG 中禁止的事情，但对于智能体来说是必要的。在这里，图不仅仅是执行固定的顺序，它根据实际出来的东西决定路线。人们忘记的一个重要细节：图中的任何循环必须有一个计数器和一个上限。如果没有一个，widen_search 将旋转直到预算耗尽。计数器存在于状态中并在路由函数中检查。 ## 人工介入点：如何暂停图并恢复它大约 60% 的生产智能体系统包含人工介入点，在金融领域这个比例更高，因为关于资本或拒绝客户的决策不能交给不透明的自动机。在技术上，这并不是“图发布到 Slack 并等待”，而是一个内置到执行中的中断机制。它的工作原理是这样的：一个节点引发中断，图将状态保存到检查点并结束执行。进程可能会死，服务器可能会重启。当人类回答时，图从同一个地方恢复，并注入答案。 ```python from langgraph.types import interrupt, Command def approval_gate(state: FactorState) -> dict: portfolio = state["portfolio"] # 执行在此停止，状态进入检查点 decision = interrupt({ "question": "批准投资组合执行吗？", "gross_exposure": portfolio["gross"], "largest_position": portfolio["max_weight"], "factors_used": [f["name"] for f in state["factor_results"] if f["passed"]], }) if decision["approved"]: return {"status": "approved", "approved_by": decision["user"]} return {"status": "rejected", "reason": decision.get("reason", "")} # 稍后，可能在另一个进程中并且一小时后： graph.invoke( Command(resume={"approved": True, "user": "pm@fund"}), config={"configurable": {"thread_id": "factors-2026-07-24"}}, ) ``` 关键点是恢复是通过线程标识符进行的，而不是通过内存中的对象。中断可以在进程重启中存活，因为状态存在于检查点中。这正是将真实的人工介入点与代码中的阻塞等待区分开来的原因，后者既无法在部署中存活，也无法在崩溃中存活。金融图中强制人工介入的三个地方：在执行交易或移动资本之前，在受监管流程中拒绝客户之前，以及在任何回滚成本高于等待的操作之前。其他所有东西都是自动化的，但这三个不是。另外，关于给人类看什么。中断应该携带的不是原始状态转储，而是决策的压缩摘要：提出了什么，基于什么理由，哪些替代方案被丢弃了。一个看到十千字节 JSON 的人类会不看就点击批准，控制点变成了橡皮图章。 ## 检查点：恢复，而不是预防检查点是在每个节点之后将状态保存到持久存储。它给你三样东西：从最后一点崩溃的恢复、暂停以进行人工干预，以及回滚执行以进行调试的能力。 ```python from langgraph.checkpoint.postgres import PostgresSaver checkpointer = PostgresSaver.from_conn_string(DB_URL) graph = builder.compile(checkpointer=checkpointer) # 每次运行都在它自己的线程中；你可以通过它恢复 config = {"configurable": {"thread_id": "factors-2026-07-24"}} result = graph.invoke({"universe": tickers}, config=config) ``` 这里是几乎没有人陈述的诚实区别。检查点是一种在失败后恢复的机制，而不是预防失败的机制。它让你回到节点倒下之前的状态，但没有任何作用来阻止节点因相同原因再次倒下。如果一个节点在命中率限制上失败，从检查点恢复会重现该失败。预防是单独的工作：并发限制、指数退避重试、降级到缓存。因此，实用规则是：检查点是强制性的但不够充分。故障必须在节点级别处理，并且当任务允许时，图必须能够在没有失败分支的情况下继续。 ```python def make_factor_node(name: str): def node(state: FactorState) -> dict: try: result = compute_factor(name, state["universe"]) return {"factor_results": [{"name": name, "passed": True, **result}]} except RateLimitError as e: # 降级：分支不会杀死图，它将自己标记为失败 return { "factor_results": [{"name": name, "passed": False, "reason": "rate_limit"}], "errors": [f"{name}: {e}"], } return node ``` 关键点：节点不会向外引发错误，它返回一个标记的结果。图用六个因子而不是七个继续，验证器从标记中看到有一个缺失并将其计入。故障被限定在节点而不是拉下管道，这正是图存在的属性。 ## 调试：如何修复崩溃的图调试图与调试脚本的不同之处在于，通常没有堆栈跟踪。一个节点运行，返回了错误的东西，图继续了，错误在三个节点后浮出水面。三个工具使这变得可处理。首先：读取检查点。由于状态在每个节点之后都被保存，你可以拉取运行历史并查看每个点的状态是什么样的。这回答了“数据究竟在哪里出了问题”。 ```python # 线程的状态历史，最新的在前 for snapshot in graph.get_state_history(config): print(snapshot.next, len(snapshot.values.get("factor_results", []))) # 你可以看到因子计数停止增长是在哪个节点 ``` 其次：回滚和重放。你可以从中途运行的检查点中取出，修改状态，并使用不同的数据从那里继续。这让你测试假设“如果验证器收到了这个输入，它会正确决定”，而无需从头开始重新运行整个图，也无需为所有前面的节点再次付费。 ```python # 从特定检查点获取并使用状态编辑从中重放 target = list(graph.get_state_history(config))[3] graph.update_state(target.config, {"min_factors": 2}) # 更改条件 graph.invoke(None, target.config) # 从这一点继续 ``` 第三：节点跟踪。每个节点记录输入、输出、持续时间和成本。没有它，运行的成本在账单到来之前一直是个谜，并且找不到慢节点。另外，一类只能通过检查点捕获的错误：并行分支上的竞态。症状是特征性的——运行完成没有一个错误，但数据从结果中丢失了，并且在相同输入上的重复运行给出不同的结果。没有异常的不确定性几乎总是一个丢失的状态更新，也就是说，一个被遗忘的 reducer。 ## 生产前测试拓扑图可以也应该在没有任何模型调用的情况下进行测试。拓扑是结构，其正确性与节点内容分开验证。 ```python def fake_node(name, payload): """一个代替真实节点的存根：不进行模型调用，返回固定答案。""" def node(state): return {"factor_results": [{"name": name, "passed": True, **payload}]} return node def test_barrier_collects_all_branches(): """检查屏障是否收集了所有并行分支，而不仅仅是最后一个。""" b = StateGraph(FactorState) names = ["momentum", "value", "quality", "low_vol"] for n in names: b.add_node(n, fake_node(n, {"sharpe": 1.0})) b.add_edge(START, n) b.add_node("collect", lambda s: {}) for n in names: b.add_edge(n, "collect") b.add_edge("collect", END) out = b.compile().invoke({"universe": [], "factor_results": [], "errors": []}) assert len(out["factor_results"]) == 4, "branch loss: check the reducer" def test_cycle_terminates(): """检查带有返回的循环不会永远旋转。""" state = {"attempts": 0, "factor_results": [], "errors": [], "universe": []} for _ in range(100): route = route_after_validation(state) if route in ("portfolio", "no_signal"): break state["attempts"] += 1 assert state["attempts"] < 10, "cycle without a cap" ``` 第一个测试在毫秒内且没有模型调用的情况下捕获了一个被遗忘的 reducer，这是图中最昂贵的错误。第二个验证循环终止。两者都在每次提交的普通 CI 中运行，这是在涉及真金白银的真实运行之前了解分支丢失的唯一方法。还有什么要测试的：条件路由覆盖每一个可能的值（没有状态使得函数返回不存在节点的名称），当一个节点失败时图到达终点，以及状态序列化到检查点没有丢失。 ## 跨节点拆分模型图节点不必在一个模型上运行，这是对成本和质量的直接杠杆。在许多统一节点上使用便宜快速的模型，在具有高推理能力的判断节点上使用强大的模型。 ```python FAST = ChatAnthropic(model="claude-sonnet-4-5", temperature=0) STRONG = ChatAnthropic(model="claude-opus-4-5", temperature=0) # 七个统一计算：便宜模型 def make_factor_node(name): ... # 使用 FAST # 验证器给出裁决：强大模型，不同的指令 def validate_factors(state): ... # 使用 STRONG ``` 这里的逻辑并不是说不同的模型捕获不同的错误，尽管这是真的。逻辑是，判断节点不应该是生成工作的同一个智能体，否则它确认自己的结论。跨节点拆分使这种分离成为结构性的而不是声明性的：判断在物理上是一个不同的节点，具有不同的提示词，在不同的模型上，只读取输出而不读取生产者的推理。 ## 数据：区分金融图和其他图的东西这里开始了使金融图无法从通用教程组装的领域特异性。首先也是最重要的：相对于决策点，节点无权看到未来。在普通管道中，你只是读取一个表。在金融节点中，没有时间过滤器地读取表是前瞻偏差，它不会作为错误出现，它会作为出色的回测和生产中的损失出现。防御是在状态级别构建的：截止时间戳存在于状态中，并且每个节点都有义务应用它。 ```python class FinState(TypedDict): as_of: datetime # 截止点，在整个运行中共享 universe: list[str] factor_results: Annotated[list[dict], add] data_gaps: Annotated[list[str], add] def load_prices(state: FinState) -> dict: # as_of 过滤器应用在查询中，而不是加载后 df = db.query( "SELECT ts, ticker, close FROM prices WHERE ts <= :cutoff AND ticker IN :u", cutoff=state["as_of"], u=tuple(state["universe"]), ) return {"prices": df} ``` 过滤器故意放在查询中。如果你加载所有内容并在内存中修剪，迟早有人会添加一个节点读取未修剪的帧，未来就会泄漏回来。时间截止必须是不可能通过疏忽绕过的。其次：文件发布有延迟，后来被修订。季度报告不是在季度末日期被知晓，而是几周后，然后数字得到更正。一个采用指标最终版本并将其归因于季度末日期的节点也是在展望未来，只是不太明显。正确的方法不仅存储值，还存储它变得公开已知的时刻，并对此进行过滤。第三：数据部分到达，这是正常的而不是失败。一个源响应了，另一个达到限制，第三个在中间返回缺口。图必须区分三种状态：没有数据，存在数据，部分存在数据。第三种是最危险的，因为节点将在不完整的样本上计算因子并返回一个自信的数字。 ```python def make_factor_node(name: str, min_coverage: float = 0.9): def node(state: FinState) -> dict: data = state["prices"] coverage = data["ticker"].nunique() / len(state["universe"]) if coverage < min_coverage: # 诚实的拒绝而不是在漏洞数据上的自信数字 return { "factor_results": [{"name": name, "passed": False, "reason": "coverage"}], "data_gaps": [f"{name}: coverage {coverage:.0%} below threshold"], } return {"factor_results": [{"name": name, "passed": True, **compute(name, data)}]} return node ``` 覆盖率阈值是显式拒绝比答案更有用的情况。在 60% 的范围内计算的因子不仅不太准确，而且系统上有偏差，因为缺失的数据几乎绝不是随机缺失的：掉出的往往是不流动的、新上市的或陷入困境的名称，正是那些承载信号的名称。第四：截止点必须是运行标识符的一部分。由日期和截止命名的检查点线程提供了可重复性：相同数据上的相同运行可以被拉出并重复，这就是在对有争议的决策的任何审查中要求的。 ## 应用于金融：三个真实的图现在具体说明这在金融任务中哪里得到回报。生产中看到的三种拓扑。 > 研究扇出。许多并行节点收集不同的切片（新闻、文件、价格序列、宏观）并汇聚到一个合成器节点。拓扑：一个宽的并行层加上一个屏障。主要风险：状态中的竞态，由 reducers 解决。带有辩论的管道。分析师准备论题，然后两个节点从对立位置争论（多头对阵空头），仲裁节点决定，然后是风险控制。这是 TradingAgents 的架构，这是 LangGraph 上的一个开放框架，其中管道运行分析师 - 研究人员 - 交易员 - 风险管理 - 投资组合经理。主要风险：无休止的辩论，由状态中的轮次计数器解决。合规管道。从文档中提取、观察列表筛选、不良媒体检查、验证、人工决策点。这里最重要的不是速度而是可审计性：每个节点将其理由写入状态，检查点被保留，并且可以拉出和解释任何决策。在受监管的任务中，不透明的系统无论准确性如何都是不可用的。三者的共同点：在没有数据依赖的地方并行化，在读取所有内容的节点之前有一个屏障，以及在错误成本高的情况下有一个强制性的人工介入点。大约 60% 的生产智能体系统增加了人工介入点，而在金融领域，出于监管原因，这一比例更高。 ## 可观测性和成本没有跟踪的图是一个黑匣子，调试变成了猜测。你需要的最低限度：每个节点的跟踪（输入、输出、持续时间、成本）、整个运行的指标以及预算超支警报。关于成本，诚实地说：七个节点的并行扇出是七次同时的模型调用，一个带有返回的循环将其乘以迭代次数。预算必须在运行之前计算，而不是之后。实用方法：每个循环都有一个硬性的迭代上限，每次运行都有一个成本上限，以及在接近限制时降级到更便宜的模型。一个可以无限旋转的图总有一天会无限旋转。 ## 用于长进程的两层架构最后一件事将生产与演示区分开来。当运行需要几分钟并且生活在一个服务内部时，图框架是好的。当一个进程运行数小时，涉及外部服务，并且必须在基础设施重启中存活时，行业使用两层：外部的持久执行层（Temporal 和类似）和内部的图层。职责划分很简单。外层负责尽管崩溃、重启和网络故障，进程仍能到达终点，并协调几个服务。内部图负责逻辑：路由、状态、一个任务内的条件转换。对于过夜运行并触及几个系统的金融流程，这是标准配对。如果运行适合几分钟并且生活在一个进程中，第二层是不必要的，内置检查点就足够了。提前添加它是没有收益的复杂性。 ## 反模式：如何不设计图出现最频繁的五个设计错误。每一个在做出决定的时刻看起来都是合理的。 > 一个本可以用函数解决的图。如果步骤是固定的，没有分支，并且状态不比运行长寿，图增加了一个检查点、序列化和一个没有任何作用的抽象层。迹象：图没有一个条件边也没有一个循环。疗法：普通函数。状态转储。所有东西都被塞进状态，包括原始数据帧和完整的模型响应。检查点膨胀，序列化减慢，每个节点都从中读取三个字段的兆字节。疗法：状态中的引用和元数据，存储中的重对象。虚假并行性。节点分布在并行分支中，但每个都通过共享存储读取邻居的结果。形式上并行，实际上是一个具有不确定结果的竞态。迹象：节点顺序影响结果，即使它们被标记为独立。疗法：如果有数据依赖，它是一个顺序边，而不是并行分支。一个做三件事的节点。一个节点拉取数据，计算因子，并写入报告。失败时，不清楚哪部分坏了，从检查点恢复时，所有三个操作都重复，包括已经成功的那些。疗法：一个节点是一个具有一个清晰输出的操作。制造者内部的判断者。同一个节点首先生成，然后根据它自己的标准检查自己。形式上存在检查，实际上不存在，因为生产者对自己的输出太宽容了。疗法：判断者是一个只读取结果的单独节点。前三个的共同诊断：如果你不能在一分钟内在纸上画出图，那么拓扑对于任务来说太复杂了。图首先是你在脑海中持有的模型，如果它不能被持有，它也不能被调试。 ## 构建顺序有效的技术顺序：首先定义具有显式类型和并行字段 reducers 的状态，然后从数据依赖构建拓扑，然后添加带有循环上限的条件路由，然后是检查点，然后是节点级故障处理，然后是跟踪和预算。每一步捕获它的一类问题：类型化状态捕获竞态，数据驱动的拓扑捕获虚假并行性，上限捕获无限循环，检查点提供恢复，故障处理阻止一个节点拉下运行，跟踪使调试成为可能。 ## 相关链接 - [zostaff](https://x.com/zostaff) - [@zostaff](https://x.com/zostaff) - [8.5K](https://x.com/zostaff/status/2080655698619142274/analytics) - [Upgrade to Premium](https://x.com/i/premium_sign_up) - [10:05 PM · Jul 24, 2026](https://x.com/zostaff/status/2080655698619142274) - [8,566 Views](https://x.com/zostaff/status/2080655698619142274/analytics) - [View quotes](https://x.com/zostaff/status/2080655698619142274/quotes) --- *导出时间: 2026/7/25 16:35:56*

🔗 相关文档