企业官网建设流程全解析

凌晨两点，一份自动生成的分析报告摆在屏幕上，营收增速、毛利率、经营性现金流三个数字彼此对不上。我把日志翻出来逐条比对，问题既不是模型"胡说"，也不是数据源坏了——它把上一季度的口径当成了本季度，并且在没有原始数值的情况下，自己"算"出了一个看起来很合理的复合增长率。那次之后，我把原来那条几百行的脚本拆了，改成一张状态图：谁负责取数、谁负责算数、谁负责解释、在哪一步停下来等人确认，全部写进图里，而不是靠注释和口头约定。

这套东西就是我一直在维护的智能金融分析工作流，底层用 DeepSeek 系列模型做解读与写作，用 LangGraph 把流程编成带检查点的状态图。它适合两类人参考：一类是手里已经有行情与财务数据源、想把"读数据—出观点—成稿"这段自动化掉的工程师；另一类是想搞清楚多节点工作流到底能干什么、不能干什么的产品侧同学。下面这些内容全部来自我自己的实现与联调过程，涉及具体数值的地方都做了脱敏，示例标的也是虚构的，只讲工程做法。

1. 从"一条Prompt跑到底"到状态图：我为什么拆掉旧脚本

最早那版实现非常朴素：拼一段超长提示词，把行情摘要、财务三张表的关键行、几条新闻标题一股脑塞进上下文，让模型直接输出一份分析。跑单个标的时效果还不错，一旦要批量跑几十上百个标的，问题就集中爆发了。

1.1 旧脚本翻车的三种典型现场

第一种是上下文挤压。财务数据本身字段多、位数长，塞进提示词之后动辄上万 token，模型注意力被稀释，越靠后的新闻内容越容易被忽略，最后输出的结论只反映了前半段数据。

第二种是中间结果不可见。脚本跑完只吐一份最终报告，中间到底取到了哪些字段、缺失字段是怎么处理的、模型有没有被要求"自行补齐"，全靠事后猜。金融场景里，结论错了还能忍，过程查不清是不能忍的。

第三种是出错只能整段重跑。某一次调用超时，前面二十分钟的取数和清洗全部作废。批量任务里这种浪费会被放大好几倍，尤其是数据接口本身有频率限制的时候，重跑意味着又要重新排队。

我把这三件事列在白板上，发现它们指向同一个根因：流程没有显式的结构。提示词是线性的、不可拆的、不可中断的，而分析这件事本身是分阶段、有分支、需要回头的。

1.2 LangGraph给的真正是"可中断、可回放"，不是"更聪明"

很多人第一反应是"用了图框架，模型是不是更聪明了"。实际并不是。LangGraph 提供的是编排能力：节点是普通函数，边是路由规则，状态在节点之间传递，每一步执行完可以落一个检查点。它不会提升模型本身的推理水平，但会把一件很麻烦的事变简单——把不可控的长对话，切成可控的短调用。

在金融分析这个场景里，这一点尤为关键：

注意：状态图解决的是"编排可控"，不是"内容正确"。模型该幻觉还是会幻觉，后面第 5 节专门讲这个。

1.3 DeepSeek在这套组合里承担哪一段活

我一开始想把所有环节都交给模型，试了两周就放弃了。现在的分工是这样的：

• 取数、清洗、对齐、指标计算：纯代码，模型不参与。复合增长率、同比环比、波动率、最大回撤这类东西，用几行代码算出来是确定的，交给模型去"算"，等于主动引入不确定性。
• 技术面与基本面的定性解读：交给模型。它擅长把一堆数字翻译成"这意味着什么"，比如毛利率环比走弱但绝对水平仍在区间内，这种需要语感的判断，写规则很难覆盖全。
• 风险清单的生成与归因：交给推理能力更强的那个模型版本。它需要把多个来源的矛盾点串起来，属于多步推理。
• 最终成稿与格式约束：交给响应更快、对结构化参数支持更宽松的通用版本。

这里有个踩过的坑值得单独提醒：我最初图省事，把结构化输出（要求返回固定字段的 JSON）也交给了推理型模型，结果那段时间解析失败率明显偏高。后来查下来是不同模型版本对结构化输出参数的支持程度不一致。我的做法是把"需要严格结构"的活儿统一收敛到通用版本，推理版本只产出自然语言段落，再由汇总节点统一抽取。你接入时务必先跑一遍你的目标模型版本，确认它到底支持哪些参数，别照抄别人的配置。

2. State字段怎么定，决定了后面能不能查错

状态设计是整个工作流里最容易被低估的一环。我第一版状态字典只有五个字段，跑到第三版的时候膨胀到二十多个，还出现了序列化失败、字段互相覆盖、恢复后拿到旧值等一系列问题。回头看，如果一开始就把结构定清楚，能省下大量返工。

2.1 用TypedDict把状态约束成可序列化的扁平结构

检查点要落盘，就意味着状态必须是可序列化的。我见过有人往状态里塞了数据库连接对象和自定义类实例，内存版跑得好好的，一换持久化检查点就报错。我的建议是只放基础类型：字符串、数字、布尔、列表、字典，嵌套层级控制在三层以内。

from typing import TypedDict, Optional, Literal class AnalysisState(TypedDict, total=False): run_id: str # 本次运行标识，同时作为检查点的 thread_id target: str # 分析标的（示例中为虚构代号） as_of: str # 数据截止日，用于防止口径错位 schema_version: str # 状态结构版本号 raw_quotes: list # 原始行情，清洗前 fundamentals: dict # 清洗后的财务字段 metrics: dict # 代码计算出的指标，模型不可写 tech_view: str # 技术面解读 fund_view: str # 基本面解读 news_view: str # 舆情面解读 risk_flags: list # 风险标记，结构化 draft: str # 汇总草稿 retry_count: int # 重试计数，防死循环 human_decision: Optional[str] final_report: str

total=False是有意为之。金融数据经常缺字段，比如某标的当季没有披露某项明细。如果强制全部必填，缺一个就抛异常，整个流程走不下去。允许可选之后，缺失字段在清洗节点被显式标记为None并写入risk_flags，下游节点看到标记就知道该怎么说——是"该项未披露"，而不是默认为零。

2.2 数值归数值、文本归文本：计算与解释分家

这条是我在踩了开篇那个坑之后立的规矩：metrics字典只允许计算节点写入，任何模型输出都不得写入该字段。

具体做法是，在计算节点里把所有需要除法、需要同比、需要年化的指标一次性算完，保留原始精度，同时把计算口径也一起写进去：

def compute_metrics(fund: dict, quotes: list) -> dict: out = {} rev_now, rev_prev = fund.get("revenue"), fund.get("revenue_ly") if rev_now is not None and rev_prev: out["revenue_yoy"] = round((rev_now - rev_prev) / abs(rev_prev) * 100, 2) out["revenue_yoy_basis"] = "本季对上年同期，未年化" else: out["revenue_yoy"] = None out["revenue_yoy_basis"] = "缺失上年同期数据，未计算" # 波动率、最大回撤等同理，全部在此处算完 return out

然后把metrics以只读快照的形式塞进提示词，并在系统提示里明确写死："以下指标由系统计算得出，你不得修改、不得自行推算新的数值；如需引用，原样抄录。"这句话看着像废话，但实测下来，加上之后数值篡改的比例明显下降。

2.3 加上时间戳与版本号，避免"用了旧数据算新结论"

as_of和schema_version这两个字段是我在第二次翻车后补的。as_of记录数据截止日，每次调用模型时都把它写进提示词，并要求模型在输出中复述；schema_version用来处理状态结构升级——老检查点里的字段名和新代码对不上时，读取阶段先做一次迁移，缺的字段补默认值，多的字段忽略。

这两个字段看起来是小事，但在批量回放历史数据做评估的时候价值极大。我曾经因为没记录截止日，用三个月前的数据跑出了"本期业绩超预期"的结论，而那个"超预期"其实是拿更晚的预期值去比更早的实际值，方向完全反了。加上时间戳之后，这类错误在评估阶段就能被自动筛出来。

3. 五个节点各守一段：把脏活从模型手里摘出去

节点划分的原则只有一条：一个节点只做一件事，且这件事的输入输出能被单独测试。我现在这套流程是五个节点加两条条件边，跑一轮大约六到九次模型调用。

3.1 数据准备节点：清洗、对齐、缺失标记

这个节点里没有一次模型调用，全部是代码。它干四件事：拉取行情与财务原始数据、按as_of对齐时间口径、剔除明显异常值（比如价格为负、成交量为零但价格变动）、把缺失字段写进risk_flags。

对齐时间口径这一步最容易出问题。不同数据源的日期字段含义不一样，有的是交易日，有的是报告期，有的是公告日。我的做法是统一转成一个内部日期格式，并且在状态里同时保留原始日期字符串，方便出问题时回溯到底用的是哪一条。

def prepare_node(state: AnalysisState) -> dict: raw = fetch_raw(state["target"], state["as_of"]) # 抽象接口，内部按你的数据源实现 fund, flags = clean_fundamentals(raw["fund"], state["as_of"]) quotes = clean_quotes(raw["quotes"]) metrics = compute_metrics(fund, quotes) if metrics.get("revenue_yoy") is None: flags.append("missing:revenue_yoy") return { "raw_quotes": quotes, "fundamentals": fund, "metrics": metrics, "risk_flags": flags, }

提示：清洗逻辑一定要写单元测试。我见过最隐蔽的一个 bug 是日期解析把"12/01"当成 12 月 1 日，而数据源实际含义是 1 月 12 日。这种错误在单条数据上完全看不出来，批量跑出来的结论却会整体偏移。

3.2 三个分析节点的提示词骨架

技术面、基本面、舆情面三个节点并行执行，互不依赖。它们的提示词结构高度相似，都是四段式：角色与边界 → 只读数据 → 输出要求 → 禁止事项。

以基本面节点为例，骨架大致是这样的：

FUND_SYSTEM = """你是一名财务分析助手，只做解读，不做预测，不给操作建议。 输入包含两部分：metrics（系统计算的指标，只读）与 fundamentals（原始字段）。 规则： 1. 只能引用 metrics 中存在的数值，不得自行计算新数值； 2. 引用时必须原样抄录，保留两位小数； 3. 字段缺失时写"该项未披露"，不得推定为零或行业均值； 4. 输出 3 到 5 段，每段不超过 120 字，不使用绝对化措辞； 5. 不评价股价高低，不出现"买入""卖出"等表述。"""

有三个细节值得说。第一，明确禁止自行计算，这一条比想象中重要，模型非常乐意"帮你算一下"。第二，限制段落数与字数，不限制的话它会写出八百字，汇总节点再压缩一遍，等于白花两次调用的钱。第三，禁用绝对化措辞，这既是合规要求，也让输出更接近专业分析的口径。

舆情面节点的提示词略有不同，它需要处理的是新闻标题列表，重点是去重、按时间排序、标注来源数量。我在这个节点里额外加了一条规则："同一事件被多条标题重复提及时，只保留信息量最大的一条，并注明出现次数。"不加这条，输出会变成标题罗列。

3.3 汇总节点的结构化输出与引用约束

汇总节点把三份解读合成一份草稿，并生成结构化的风险清单。这里的输出必须是可解析的，否则下游没法用。我用的是"提示词里给定 schema + 解析兜底"的方案：

import json, re def parse_structured(text: str) -> dict | None: m = re.search(r"\{.*\}", text, re.S) if not m: return None try: data = json.loads(m.group(0)) except json.JSONDecodeError: return None if not isinstance(data.get("risk_flags"), list): return None return data

为什么不直接用框架自带的结构化输出能力？两个原因。一是不同模型版本对相关参数的支持程度不一致，混用的时候容易出兼容问题；二是我需要对失败做分级处理——解析失败一次就重试一次，第二次还失败就走降级路径（把原始文本塞进草稿，风险清单留空并标记待人工补录），而不是让整个流程崩掉。

注意：解析失败不要静默丢弃。我在状态里专门留了retry_count，重试两次仍失败则触发条件边跳到人工复核，把原始文本一并交出去。静默失败是这类系统里最危险的行为。

4. 条件边与人工闸门：让流程在该停的地方停住

前面三个节点跑完，流程其实还没结束。真正让这套东西从"玩具"变成"能用的工具"的，是条件边和人工闸门。前者决定流程往哪走，后者决定什么时候停下来等人。

4.1 用条件边做质量闸与重试上限

我在汇总节点之后加了一个质量判定：检查草稿长度是否达标、风险清单是否为空、三个面的解读是否都提到了。判定不通过就回到汇总节点重写，但必须有上限。

from langgraph.graph import StateGraph, END def quality_gate(state: AnalysisState) -> Literal["revise", "human_review", "abort"]: if state.get("retry_count", 0) >= 2: return "abort" if not state.get("draft") or len(state["draft"]) < 300: return "revise" if not state.get("risk_flags"): return "revise" return "human_review" builder = StateGraph(AnalysisState) builder.add_node("prepare", prepare_node) builder.add_node("fundamental", fundamental_node) builder.add_node("technical", technical_node) builder.add_node("news", news_node) builder.add_node("summarize", summarize_node) builder.add_node("human_review", human_review_node) builder.set_entry_point("prepare") builder.add_edge("prepare", "fundamental") builder.add_edge("prepare", "technical") builder.add_edge("prepare", "news") builder.add_edge(["fundamental", "technical", "news"], "summarize") builder.add_conditional_edges("summarize", quality_gate, { "revise": "summarize", "human_review": "human_review", "abort": END, })

这里有个必须注意的点：revise回到summarize之前，一定要在节点内部把retry_count加一。我最早忘了加，结果质量闸判定不通过、节点重写、还是不通过，来回打转，直到触发框架的递归上限才停。虽然框架有兜底，但那一轮把额度烧掉了不少。后来我习惯在编译时显式设置recursion_limit，一般是节点数的三到四倍，超了就报错，比默默烧钱好。

并行那一层也要留意。三个分析节点虽然写成并行边，但实际的模型调用是不是并发，取决于你的执行配置。批量跑的时候我一般是串行执行、多进程并行处理不同标的，这样限流更好控制，出问题也好定位是哪个标的。

4.2 人工复核节点的interrupt写法与版本差异

人工复核是这套流程里我最坚持保留的一环。全自动出稿在技术演示里很酷，在真实使用里风险太高——至少第一版必须有人签字。

新版本的写法是在节点内调用挂起函数，把需要人看的内容抛出去：

from langgraph.types import interrupt def human_review_node(state: AnalysisState) -> dict: payload = { "target": state["target"], "as_of": state["as_of"], "draft": state["draft"], "risk_flags": state["risk_flags"], "metrics": state["metrics"], } decision = interrupt(payload) # 流程在此挂起，等待外部恢复 return {"human_decision": decision}

恢复的时候带上同一个thread_id和恢复指令：

config = {"configurable": {"thread_id": state_run_id}} graph.invoke(Command(resume="approve"), config=config)

老版本框架用的是编译期参数interrupt_before=["human_review"]，配合graph.update_state()手动改状态再续跑。两种方式都能用，区别在于：新版把"要给人看什么"写在节点内部，逻辑更内聚；老版在编译期声明，节点函数本身很干净。如果你在跟别人的教程对照，发现写法完全对不上，先确认框架版本，别怀疑自己写错了。

4.3 检查点存储：内存版跑通，落盘版上线

联调阶段用内存检查点就够了，重启即清空，方便反复试。上线必须换持久化实现，否则服务一重启，所有挂在人工复核的流程全部丢失——这个坑我踩过，一次发版丢了十几条待审任务，只能重新跑。

切换方式很简单，编译时换掉检查点对象即可。要注意两件事：第一，持久化后状态必须可序列化，前面强调的扁平结构在这里体现价值；第二，thread_id要设计成稳定可查的，我用的是"标的 + 截止日 + 批次号"拼出来的字符串，便于按标的检索待审任务。

提示：检查点表会随着运行次数持续增长，记得定期归档。我现在的做法是按批次清理，超过保留期的记录导出到冷存储后删除，避免表无限膨胀拖慢查询。

5. 联调阶段真正咬人的四个问题

代码写完到能稳定跑，中间隔了大概三周。这三周里遇到的问题，比写代码本身更值得记录。

5.1 幻觉数字：看起来合理的财务比率最危险

模型编造数字有个特点：它不会编一个离谱的数，而是编一个"合理区间内"的数。比如把 12.4% 写成 12.7%，把"增长"说成"小幅增长"。这种偏差人工扫读时几乎发现不了，但结论方向可能就变了。

我的应对是三层：提示词层禁止自行计算、结构层把数值锁在只读字段、校验层做事后比对。第三层最关键——在最终成稿前跑一遍扫描，把文中出现的所有百分数与metrics里的值做匹配，匹配不上的直接标红进入人工复核。

def scan_numbers(text: str, metrics: dict) -> list: allowed = {f"{v}" for v in metrics.values() if isinstance(v, (int, float))} hits = re.findall(r"\d+\.?\d*%", text) return [h for h in hits if h.rstrip("%") not in {a.rstrip("%") for a in allowed} - {""}]

这段逻辑不完美，会误报（比如文中引用了区间上下限），但误报可以人工忽略，漏报会直接进报告。宁可多标，不可漏标。

5.2 并发限流与单次全量扫描的耗时账

单个标的跑一轮，从取数到出草稿大概需要六到九次模型调用，加上数据接口往返，几十秒到两分钟不等。几十个标的串行跑，一晚上就过去了。

我试过直接开高并发，结果触发了接口的频率限制，一半请求返回错误。后来改成固定大小的并发池，加指数退避重试，并且把失败标的单独收集起来放到队列尾部重跑，而不是让整个批次失败。这个改动让全量扫描的成功率从七成左右提到了接近全量。

还有个不成文的经验：把耗时长的标的分批错峰。数据量大的标的放在前面跑，小的放后面填充，整体完成时间比严格顺序跑更短，因为尾部的小任务能快速填满并发空位。

5.3 时间窗口错位导致的结论反转

这个问题我在第 2 节提过一次，但值得单独展开，因为它是最容易反复踩的。表现是：同一个标的，同样的数据，不同批次跑出来的结论方向相反。查下来基本是三个原因之一——行情数据用了收盘后更新的版本、财务数据用了修订后的口径、舆情数据的时间范围没和前面两个对齐。

现在的做法是把as_of作为硬约束贯穿全流程：取数时按它过滤、提示词里写进去、模型输出要求复述、入库时作为主键的一部分。看着啰嗦，但批量和回放场景下，这是唯一可靠的办法。

5.4 恢复后拿到旧状态

用持久化检查点的时候，我遇到过一次很迷惑的现象：流程从人工复核恢复后，读到的draft还是上一版的内容。查了半天发现，是恢复时传入的thread_id和原来的不一致，框架把它当成了一次新运行，从最早的检查点开始重建。

这个问题的教训是：thread_id必须由外部统一生成并持久保存，不要依赖内存变量。我现在的做法是把run_id写进任务表，恢复时从表里读，而不是从进程内存里拿。

6. 上线前的两道保险：回放评估与降级链路

功能跑通和可以交付之间，还差两件事。少做任何一件，出问题的时候都会很被动。

6.1 拿历史样本做回放，评估集怎么攒

我没有现成的标注数据，所以评估集是攒出来的：挑一批历史时点，把当时的数据快照固化成样本，跑一遍工作流，人工看结果。重点不是看它说得漂不漂亮，而是看三件事——数字对不对、口径对不对、有没有越界表述。

样本量不需要很大，我最初只用了三十条，就发现了七八个问题，包括两处数值引用错误和一处时间口径错位。随着问题修复，样本逐步加到一百多条，覆盖不同类型标的和不同数据完整度的情况。每次改提示词或换模型版本，先跑一遍这个集合，比对通过率变化，再决定要不要上。

评估的时候有个细节：把同一份输入跑两遍，看两次输出的结论方向是否一致。金融分析允许措辞不同，但不允许方向相反。方向反转的比例，是我判断提示词稳不稳定的主要指标。

6.2 模型侧不可用时的降级与降级后的告知

模型接口会超时、会限流、会临时不可用，这在批量任务里是常态。我的降级链路分三级：

1. 重试级：单次调用失败，指数退避重试两次，成功则继续，用户无感。
2. 降级级：换用备用模型版本完成同一节点的任务，输出上加标记。
3. 跳过级：某个分析面彻底失败，把该面的解读置为"未生成"，风险清单里加一条degraded:xxx，其余部分照常出稿。

第三级最重要的一条是必须告知。降级后的报告如果和正常报告长得一模一样，那比直接失败更危险。我现在的做法是，任何降级都会在报告头部显式标注，并且强制进入人工复核，不允许自动通过。

注意：降级逻辑要在设计阶段就写进状态，而不是等到出事再加。事后补降级，往往要改动一大片节点的返回值结构，代价比一开始就规划高得多。

最后分享一个我自己的习惯：每改一次提示词或者换一次模型版本，我都会把那三十条基准样本重新跑一遍，把输出和上一版并排放在一起看。多数时候差别很小，但只要有一次出现结论反转或者数值引用错误，就说明这次改动不能直接上。这套流程从第一版到现在改了七版，真正省下时间的不是模型变强了，而是每一步都能查、能停、能重来——这一点，比任何提示词技巧都实在。