1. 为什么要把“分析流水线”画成一张图1.1 从脚本堆叠到状态机一个踩过坑的起点最早做金融分析自动化很多人的做法跟我一样一个main.py里从上到下写着拉数据、算指标、调模型、拼报告中间靠几个if判断要不要重新跑。第一版能跑通第二版就开始失控——数据源偶发超时后脚本直接崩溃、报告生成到一半断了只能从头再来、某个分析维度想换个提示词结果整条链路都得重测。这类问题的根源不在模型而在于把流程逻辑写进了代码顺序里流程一变代码就得大改。LangGraph解决的就是这件事。它把整条分析流水线抽象成一张有向图节点是“一次可独立重试的计算或模型调用”边是“状态如何在节点之间流转”条件边是“下一步走哪”检查点则是“走到一半能不能存档续跑”。落到智能金融分析工作流这个具体场景这条流水线天生就有并行、有分支、有人工卡点、有失败重试用图来表达比用函数嵌套自然得多。至于为什么选DeepSeek作为推理内核不是因为它“最强”而是因为在金融文本这类任务上它的性价比结构特别合适中文财报、公告、研报的行文习惯它吃得比较准长上下文能一次性塞进多期对比材料而且deepseek-chat和deepseek-reasoner两档模型给了我们做分级路由的空间——便宜的跑抽取和分类贵的只留给最后的深度研判。这套组合下来一份包含 5~8 个分析维度的日报纯模型成本能压到可接受区间。说明本文所有数据、标的、公司一律使用模拟代称内容仅讨论工程实现与信息整理方法不构成任何投资建议也不涉及具体交易决策。1.2 LangGraph 里到底有哪几个东西值得用我把这套框架里真正高频用到的能力归成四类其余的子图、流式、持久化后端切换属于锦上添花能力API 形态在金融分析里解决什么问题状态图编排StateGraphadd_node/add_edge把流程从代码顺序中抽离改成显式声明条件路由add_conditional_edges数据是否完整、风险等级是否越线、是否需要重写并行扇出Send多标的、多维度同时分析而不是串行排队检查点与中断compile(checkpointer...)interrupt()报告出具前的人工复核、断点续跑、失败重放这里最容易低估的是条件路由和检查点的组合价值。举个例子财务数据节点跑完后如果发现某标的的营收与现金流方向严重背离我们并不直接让它进报告而是路由到一个“异常复核”节点标记该维度为“数据存疑不进入结论”。这类业务判断如果写成硬编码的if-else半年后没人记得为什么这么写写成图上的边一眼就能看懂。1.3 DeepSeek 在这套流程里的角色边界这一点必须说清楚否则整条流水线会做歪大模型在金融分析里负责“解释和归纳”不负责“计算和判定”。我见过太多翻车案例让模型自己去算同比增长率、算市盈率、算波动率结果就是数字忽对忽错而且错得毫无规律。正确分工是所有数值计算交给pandas和确定性的函数模型只拿到“已经算好的指标 JSON”去做因果解释、风险提示、文本组织和跨维度一致性检查。模型输出里出现的每一个数字都必须能回溯到工具返回的原始值这一条后面我会专门讲怎么做校验。2. 工作流骨架与状态设计2.1 四层结构采集、计算、研判、成稿整张图我拆成四层层与层之间只通过 State 传递数据不共享全局变量采集层行情快照、财务指标、公告与新闻、宏观与行业数据四个节点互相独立可并行。计算层把原始数据压成结构化指标字典波动率、回撤、均线偏离、同比环比、现金流质量等全部是纯函数。研判层多个分析角色并行各自读同一份指标字典输出固定 schema 的结论对象观点、证据、置信度、风险点。成稿层汇总研判结果做冲突消解、生成报告草稿、数字回溯校验最后进人工复核节点。这四层不是拍脑袋分的而是按失败的影响半径分的。采集层失败只影响单维度重试成本低计算层失败是代码 bug必须显式报错不能静默研判层失败是模型问题可以换提示词重跑成稿层失败影响最大所以要加人工卡点。分层之后每一层的重试策略、超时设置、日志粒度都可以单独调。2.2 State 该怎么定什么进状态什么不进状态设计是这套东西里最容易埋雷的地方。我的原则是State 只放“需要跨节点流转的轻量结论”不放原始大对象。from typing import TypedDict, Annotated from operator import add class AnalysisState(TypedDict, totalFalse): task_id: str # 任务唯一标识 as_of: str # 分析基准日 universe: list[str] # 本次覆盖的模拟标的池 snapshot_ref: dict # 原始数据的存储引用路径/键不是数据本身 metrics: dict # 计算层产出的指标字典 findings: Annotated[list[dict], add] # 研判结论多节点写入需要 reducer conflicts: list[dict] # 结论冲突记录 report_draft: str # 报告草稿 review_notes: str # 人工复核意见 revision: int # 重写轮次用于熔断两个关键决策值得展开。第一findings用了Annotated[..., add]这个 reducer。因为研判层是并行扇出的多个节点会同时往同一个 key 写值没有 reducer 就会互相覆盖报错形态是“只有最后一个节点的结论活下来”而且不报错极难发现。第二原始行情和公告文本不进 State只存一个引用对象存储的 key 或本地路径节点需要时自己去取。原因很实在检查点是要序列化的。如果把几十万字的公告原文塞进 State每走一步就序列化一次跑一轮下来磁盘和延迟都顶不住而且datetime、DataFrame这类对象序列化还会直接抛异常。我最早没注意这点跑到第四步就报TypeError: Object of type Timestamp is not JSON serializable排查了半小时才发现是某个节点顺手把df.index[0]写进了状态。2.3 并行扇出与汇聚Send 的用法和坑多标的、多维度并行是这条流水线提速的核心。Send的语义是“动态生成若干个下游任务”每个任务带着自己的入参进入同一个节点from langgraph.types import Send def fan_out_analysts(state: AnalysisState): tasks [] for ticker in state[universe]: for role in (fundamental, technical, event): tasks.append(Send(analyst, { task_id: state[task_id], ticker: ticker, role: role, metrics: state[metrics].get(ticker, {}), as_of: state[as_of], })) return tasks这里有个必须提醒的细节Send的第二个参数会成为目标节点看到的完整输入状态而不是和主状态做合并。也就是说如果目标节点里读了state[findings]而你没在Send的参数里带上它就会KeyError。我的处理方式是让研判节点只依赖传入的子集绝不回读全局字段同时目标节点的返回值会通过 reducer 合并回主图所以findings用add是必需的。扇出之后的汇聚有两种做法。一种是隐式汇聚所有并行分支跑完自动进入下一个节点适合“必须全部分析完才能成稿”的场景。另一种是显式汇聚节点我在里面做冲突消解——比如基本面角色给出“景气度上行”事件角色给出“存在重大不确定性”汇聚节点就要把这两条标成冲突交给后续环节处理而不是简单拼在一起。3. 核心节点实现细节3.1 采集节点重试、限流与幂等三件套采集节点看起来最无聊但线上出问题最多的就是它。我的写法是把“一次取数”包成一个带退避的重试函数节点本身只做调度import time, random, httpx def fetch_with_retry(url: str, params: dict, tries: int 4, timeout: float 8.0): last None for i in range(tries): try: r httpx.get(url, paramsparams, timeouttimeout) if r.status_code 200: return r.json() if r.status_code in (429, 500, 502, 503, 504): raise RuntimeError(fretryable status {r.status_code}) raise ValueError(fnon-retryable status {r.status_code}) except Exception as e: last e # 指数退避 抖动避免同一时刻集体重试把上游打穿 sleep min(2 ** i, 8) random.uniform(0, 0.5) time.sleep(sleep) raise RuntimeError(ffetch failed after {tries} tries: {last})几点实操心得。退避一定要加抖动纯指数退避在并行扇出的场景下会让所有分支在同一秒重试反而更容易触发限流。并发要卡住用asyncio.Semaphore把同时对某个数据源的请求压到 3~5 个比无脑并发稳得多。最重要的是幂等落盘的文件名用{数据源}_{标的}_{日期}_{版本}.json节点开头先检查文件是否存在存在就直接读。这样一来重跑整条流水线时采集层几乎零成本调试效率天差地别。注意采集层永远不要假设上游字段稳定。我给每个数据源都写了一份最小的 schema 校验字段缺失时直接抛错并标明缺失字段名而不是让None一路流到模型那里最后生成一段“该指标暂缺”的废话报告。3.2 指标计算节点把算术彻底从模型手里拿走计算节点是纯函数输入是原始数据引用输出是指标字典。这里给出我实际在用的核心几项参数和口径都写死在代码里import numpy as np, pandas as pd TRADING_DAYS 252 def compute_metrics(df: pd.DataFrame, rf_annual: float 0.02) - dict: close df[close].astype(float).reset_index(dropTrue) ret close.pct_change().dropna() n len(ret) # 年化波动率 vol ret.std(ddof1) * np.sqrt(TRADING_DAYS) # 年化夏普无风险利率先折算到日 rf_daily rf_annual / TRADING_DAYS excess ret - rf_daily sharpe excess.mean() / ret.std(ddof1) * np.sqrt(TRADING_DAYS) if ret.std(ddof1) 0 else np.nan # 最大回撤 cum (1 ret).cumprod() mdd float((cum / cum.cummax() - 1).min()) # 均线偏离MA20 ma20 close.rolling(20).mean() bias20 float((close.iloc[-1] / ma20.iloc[-1] - 1)) if len(close) 20 else np.nan # RSI14Wilder 平滑 diff close.diff() up diff.clip(lower0).ewm(alpha1/14, adjustFalse).mean() down (-diff.clip(upper0)).ewm(alpha1/14, adjustFalse).mean() rs up / down.replace(0, np.nan) rsi14 float(100 - 100 / (1 rs.iloc[-1])) if down.iloc[-1] ! 0 else 100.0 return { sample_size: int(n), annual_vol: round(float(vol), 6), sharpe: round(float(sharpe), 4) if np.isfinite(sharpe) else None, max_drawdown: round(mdd, 6), ma20_bias: round(bias20, 6) if np.isfinite(bias20) else None, rsi14: round(rsi14, 2), data_ok: bool(n 60), }为什么坚持用代码算因为夏普比率这种东西让模型算十次能出五种结果而代码算出来的每一次都一样还能写单元测试。我给自己定的门槛是报告中出现的任何数值型结论都必须能对应到指标字典里的一个键。同时加了data_ok这种质量位——样本不足 60 个交易日时后面所有基于统计量的解释都要降级处理不能理直气壮地下结论。另外一个常被忽略的点是口径一致性。年化波动率到底是按 252 还是 250 折算、夏普用不用无风险利率、回撤用复权价还是原始价这些必须在代码里写死并在报告里注明否则同一份数据今天出 18.2%、明天出 17.9%用户第一反应就是“这系统不可信”。3.3 研判节点结构化输出与提示词分层研判节点的核心不是提示词写得多花哨而是输出必须可解析、可校验。我给每个角色定义一个固定的输出 schema要求模型返回 JSON并在代码里做严格解析ANALYST_SYSTEM 你是一名{role}分析员。你只能基于用户提供的指标字典和材料片段做归纳 不得自行计算任何数值不得引入未提供的数字。 输出严格的 JSON字段如下 {{ ticker: string, stance: positive | neutral | negative | unclear, confidence: 0~1 之间的数字, key_points: [{{claim: string, evidence_key: string}}], risks: [string], data_gaps: [string] }} evidence_key 必须是输入指标字典中真实存在的键名找不到依据时 stance 填 unclear。三个设计要点。第一强制evidence_key指向指标字典里真实存在的键这给了我们一个机械化的幻觉检测手段——解析后遍历key_points凡是不在metrics.keys()里的直接判定为编造并触发重试。第二unclear这个选项必须给否则模型在信息不足时会被迫瞎猜反而更危险。第三data_gaps让模型主动暴露“我缺什么”这个字段在后续成稿时非常有用可以生成一段诚实的局限性说明。不同角色的提示词差异不在语气而在关注范围和禁止项。基本面角色禁止讨论价格走势技术面角色禁止引用财报科目事件角色只允许基于给定的材料片段做归纳且必须标注片段编号。这种“互相不许越界”的约束比让每个角色都“全面分析”效果更好因为后面做冲突消解时你能清楚知道分歧来自哪个视角。3.4 成稿与数字回溯校验给报告加一道机械闸门成稿节点把findings汇总让模型写一份结构化报告。这里最大的风险不是文笔是模型在改写过程中把数字改了——指标字典里明明是 0.1832报告里写成 18.3% 甚至 18% 都可能发生。所以我在成稿之后加了一个纯代码的校验节点import re NUM_PAT re.compile(r-?\d(?:\.\d)?) def verify_numbers(report: str, facts: dict, rel_tol: float 0.005): allowed [] for v in facts.values(): if isinstance(v, (int, float)): allowed.append(abs(float(v))) allowed.append(abs(float(v)) * 100) # 百分比写法 bad [] for token in NUM_PAT.findall(report): x abs(float(token)) if x 1: # 小于 1 的数字多为年份、序号、页码跳过 continue if not any(abs(x - a) rel_tol * max(a, 1e-9) for a in allowed): bad.append(token) return bad这个校验器不完美会误报比如报告里写“近 5 年”“3 个维度”这类计数所以我把它设计成只报警不阻断把bad列表塞进状态由成稿节点带着这份“疑似数字清单”重写一轮最多重写两次。实测下来加入这道闸门后报告数字与指标字典不一致的比例从肉眼可察的个位数百分比降到了接近零。同理报告里所有定性结论我要求带上括号标注依据比如“景气度边际改善依据营收同比、订单数据”。不是为了好看而是为了让人工复核时有据可查也让模型自己不敢乱写。4. 人工介入、断点续跑与成本控制4.1 用 interrupt 做合规复核卡点金融类内容天然需要“机器产出、人来签字”。LangGraph的中断机制正好对应这个需求图跑到复核节点时挂起把当前状态持久化等人看完给出意见后再恢复执行。from langgraph.types import interrupt def human_review(state: AnalysisState): decision interrupt({ report: state[report_draft], flagged_numbers: state.get(suspicious_numbers, []), question: approve / revise / reject, }) if decision.get(action) revise: return {review_notes: decision.get(notes, ), revision: state.get(revision, 0) 1} return {review_notes: approved}配套的路由函数根据review_notes和revision决定是回到成稿节点重写还是直接结束。这里必须加熔断revision 2时无论意见如何都强制结束并标记为“需人工接手”否则出现过审-重写-再审的死循环recursion_limit报错时你都不知道卡在哪一轮。提示中断恢复依赖检查点。如果你用的是内存版检查点进程一重启挂起的任务就全丢了。生产环境我建议直接上 SQLite 或数据库后端thread_id用task_id重启后按 id 捞回来继续跑。4.2 检查点选型与状态版本管理检查点这件事选型不复杂但版本管理很容易翻车。开发阶段用内存版足够速度快、无副作用一旦涉及人工中断就必须换持久化后端。后端适用阶段注意点内存版本地调试、单元测试进程退出即丢失不能用于中断场景SQLite单机部署、小规模并发写要注意锁状态别太大数据库/对象存储多实例、长期归档状态 schema 变更要做迁移或版本前缀状态 schema 变更是个大坑。你今天给 State 加了个字段昨天挂起的任务恢复时就可能缺这个 key节点里直接KeyError。我的做法是所有新增字段在 State 定义里都标totalFalse节点内部一律用state.get(x, 默认值)取值同时在task_id之外再记一个schema_version恢复时版本不匹配就直接重跑而不是硬续。4.3 Token 预算与模型分级路由成本这块我给一组实测的量级参考不同供应商价格会变但比例关系是稳定的环节单次输入量级单次输出量级建议模型档位公告/新闻抽取3k~8k tokens300~600轻量档单维度研判1.5k~3k400~800轻量档为主冲突消解2k~4k500中等档报告成稿4k~8k1.5k~3k高能力档数字复核0纯代码0不调用模型一份覆盖 6 个标的、每标 3 个维度的日报粗略算下来是 18 次研判调用加 1 次成稿输入总量约 6 万 tokens、输出约 1.2 万 tokens。如果全部用最高档模型成本是没有分级路由时的 3~4 倍而那些抽取类任务用最高档模型纯属浪费。分级路由的实现很简单在节点内部按角色查表选模型即可。但有个细节值得注意推理型模型不适合放在并行扇出里。它单次响应慢18 个分支并行等待时整体延迟会被最慢的那个拖死而且它对结构化 JSON 输出的遵从度不如通用对话模型。我的做法是研判层全部用通用模型只在最后的冲突消解环节对确实存在矛盾的少数案例调用一次推理型模型做深度权衡——调用量小、价值高、延迟可接受。另外两个省钱手段一是提示词缓存把不变的系统提示和指标字典键说明放在前面把变化的数值放后面配合供应商的上下文缓存能力重复调用能省掉一大截输入费用二是结果缓存用hash(role ticker as_of 指标字典)做键同一交易日内重跑整个流水线时未变更的维度直接命中缓存。开发调试阶段这个缓存能把一轮耗时从几分钟压到几十秒。5. 常见故障与排查实录5.1 高频报错速查表下面这些是我在把这条流水线从能跑到跑稳的过程中真实撞过的坑按出现频率排序现象大概率原因处理方式KeyError: findingsSend入参没带全或并行节点回读全局字段研判节点只依赖传入子集禁止回读并行后只剩一个节点的结论该 key 缺 reducer用Annotated[list, add]声明合并逻辑Recursion limit超限重写/复核形成死循环加revision熔断限制最多两轮Object of type Timestamp is not serializable把 DataFrame 或时间对象写进了 State只存引用和基础类型时间统一转字符串JSON 解析失败模型输出带了 markdown 代码块或说明文字解析前剥离 包裹失败则带错误信息重试一次某维度结论空泛该维度输入信息不足模型被迫作答引入data_gaps字段不足时允许unclear整体耗时异常长推理型模型进了并行分支或未设并发上限分级路由 信号量限流恢复后状态错乱检查点版本与当前 schema 不匹配带schema_version不匹配则重跑这张表里最值得说的是第一行。它不报错的时候最可怕——并行四路只有一路的结果被采纳报告看起来“完整”实际上少了三个视角。我是靠对同一标的做人工复核时才发现的当时还以为模型变蠢了。后来养成一个习惯在汇聚节点里断言len(findings) 预期分支数不等就直接抛错。5.2 输出质量的隐蔽问题比起报错更麻烦的是“能跑通但内容不对”。我遇到过三种典型情况都不是代码 bug而是设计缺陷第一种是过度自信。指标字典里data_ok是False样本不足但模型依然给出了“positive”和 0.85 的置信度。修法是把质量位直接写进提示词并强制要求data_okFalse时confidence不得超过 0.5stance只能是unclear或neutral代码里做二次校验违反就重写。第二种是证据错配。结论说的是“盈利能力改善”但evidence_key指向的是波动率指标。这是模型对键名的语义理解偏差。修法是在指标字典里给每个键附一句人话解释比如annual_vol注明“年化波动率衡量价格波动幅度不代表盈利水平”让模型没有误解空间。第三种是跨维度结论打架。基本面说“估值合理”技术面说“短期承压”如果成稿节点只是简单拼接读者会觉得这报告精神分裂。正确做法是在成稿前插入冲突消解把矛盾点显式写成“不同视角的分歧点”并且说明各自依据什么。这不是和稀泥反而比强行统一口径更诚实、更有信息量。注意别指望靠调提示词解决所有质量问题。判断标准应该是——这个问题能不能用确定性代码检查出来能就写成校验节点比改提示词可靠一百倍。5.3 上线前我必查的一份清单跑过几轮完整流程之后我形成了一个固定的自检清单基本能覆盖八成以上的翻车风险状态瘦身打开检查点文件看一眼单个状态超过几百 KB 就要重新审视往里放了什么。重放验证故意在第三个节点抛异常确认恢复后能从中断点继续而不是从头再跑一遍采集。幂等验证同一天连跑三次完整流程产出的报告除了时间戳应当完全一致。降级验证关掉其中一个数据源确认流程不中断只是报告里多出一段“该维度数据缺失”的说明。限流验证把标的池从 6 个扩到 30 个观察是否有大量 429以及延迟是否线性劣化。成本看板按节点维度统计 token 消耗我一般要求研判层不超过总成本的 60%成稿层不超过 25%剩下的留给重试和意外。最后分享一个我在实际项目里体会最深的事这条流水线真正的价值不在于“自动化写报告”而在于把分析过程变成了可审计的对象。每一步的输入输出、每一次模型调用、每一条结论的依据全都被状态和检查点记录下来了。当有人质疑某段结论时我能直接把对应的指标字典和证据片段调出来对照而不是去翻聊天记录里当时的判断。这个能力比报告本身重要得多。
