拆 DAMA 里的 LangGraph 生图工作流：从识别商品到 Critic 重试，这条链是怎么串起来的

1. 我为什么会把这条链迁到 LangGraph

这个项目前面还比较简单时，手写串行流程其实完全能跑。

无非就是：

• 识别商品
• 生成场景
• 生成 prompt
• 出图
• 评价
• 不行就再来一轮

但 DAMA 这种链路一旦往电商图智能体方向走，复杂度很快就上来了。

因为中间不再只有“出图”一个动作，还会多出很多必须保留的中间状态：

• 当前任务是卖点图还是种草图
• 商品事实卡和电商解读卡长什么样
• 展示策略是 product_only 还是 human_tryon
• 这一轮用的是什么策略
• 上一轮 Critic 具体给了什么修改建议
• 这一轮到底该继续改 prompt，还是直接换策略重来

如果这些东西继续塞在一大段函数里，短时间还能顶，后面很快就会乱。

所以我后面把它迁到 LangGraph，主要还是为了把这条链里那些有状态、会分支、会重试的部分摆清楚，不想再把它们糊在一段串行逻辑里。

2. 真正的入口不在前端，而是在 run_feature_image_graph(...)

这条生图工作流，真正的入口在：

def run_feature_image_graph(
    agent,
    feature_name: str,
    product_image_url: str,
    user_text: str,
    max_attempts: int = 3,
    output_dir: str | None = None,
    session_state: dict[str, Any] | None = None,
    context_bundle: dict[str, Any] | None = None,
    resume_state: dict[str, Any] | None = None,
) -> dict[str, Any]:

这个函数一上来并不会直接跑节点，它会先把 initial_state 组出来。

这里我比较在意两个字段：

• feature_name
• resume_state

前者决定这次任务到底按卖点图还是种草图走。

后者很关键，它意味着这条图不是只能从头跑。后面如果用户在已有结果上继续提优化意见，系统是可以带着已有中间结果，从某个节点重新接着跑的。

这也是我现在很喜欢它的一点：

这条 graph 不是一次性的，它是可续跑的。

3. 先看 FeatureImageGraphState，这基本就是整条链的运行现场

LangGraph 这类东西，先看 state 比先看节点更重要。

DAMA 这里的状态定义在 graph/state.py：

class FeatureImageGraphState(TypedDict, total=False):
    feature_name: str
    entry_node: str
    user_text: str
    product_image_url: str
    max_attempts: int
    output_dir: str | None
    session_state: dict[str, Any]
    context_bundle: Any

    product: Any
    interpretation: Any
    brief: Any
    presentation_plan: Any
    scene: Any
    strategy: Any
    prompt_package: Any
    latest_prompt: str
    latest_image: Any
    latest_critique: Any

    critique_feedback: str | None
    attempt_index: int
    attempts: list[Any]
    best_attempt: Any
    approved: bool
    retry_action: str
    protect_best_attempt: bool
    final_result: Any

我第一次看到这个 state 时，脑子里一下就清楚很多了。

因为它已经把这条链里真正重要的东西摊开了：

第一类：任务输入

• feature_name
• user_text
• product_image_url

第二类：中间理解结果

• product
• interpretation
• brief
• presentation_plan
• scene
• strategy
• prompt_package

第三类：运行态

• attempt_index
• attempts
• best_attempt
• latest_critique
• retry_action
• approved

也就是说，这个 graph 不是只在组织“节点顺序”，它其实也在组织：

一张图从输入到最终出图之间，哪些中间判断必须被显式保存。

4. START 不是死板地从头进，它会先判断应该从哪一步接上

feature_image_graph.py 里，我觉得很值钱的一段是这个：

graph.add_conditional_edges(
    START,
    _decide_entry_step,
    {
        "recognize_product": "recognize_product",
        "build_interpretation": "build_interpretation",
        "build_brief": "build_brief",
        "build_presentation_plan": "build_presentation_plan",
        "create_scene": "create_scene",
        "select_strategy": "select_strategy",
        "build_prompt_package": "build_prompt_package",
    },
)

一开始我以为 graph 每次都会从商品识别开始重走。

但这里不是。

它会先调用 _decide_entry_step(state)，根据当前 state 里已经有什么结果，决定从哪一步进。

比如：

• 如果已经有 product，可以直接跳过识别
• 如果已经有 brief，可以直接去补 presentation_plan
• 如果已经有 scene 和 strategy，可以直接重建 prompt
• 如果显式传了 entry_node，那就按指定节点恢复

这套设计对 refine 特别有用。

因为用户很多时候不是要从零再来一张，而是会说：

• 背景换一下
• 再自然一点
• 不要那么像硬广
• 上身感更强一点

这时系统没必要再重新识别商品、重新理解类目。它只需要从 build_prompt_package 甚至 select_strategy 之后继续走。

这也是 dispatcher.py 里 refine 那段 resume_state 的意义：

resume_state = {
    "entry_node": "build_prompt_package",
    "product": product,
    "interpretation": interpretation,
    "brief": brief,
    "presentation_plan": presentation_plan,
    "scene": scene,
    "strategy": strategy,
    "critique_feedback": combined_feedback,
}

我觉得这比“重新跑一遍完整链”实用得多。

5. 主链其实很清楚：理解商品，再定义任务，再组织画面

整条主链在 _build_feature_image_graph(agent) 里基本写死了：

recognize_product
-> build_interpretation
-> build_brief
-> build_presentation_plan
-> create_scene
-> select_strategy
-> build_prompt_package
-> generate_image
-> critique_image
-> update_attempt_record

如果把这几步翻成人话，大概就是：

1）recognize_product

先把商品识别出来，拿到商品事实卡。

这是后面所有东西的底。

2）build_interpretation

在事实之上，再补一层电商解读。

这一步开始回答“这件商品适合强调什么”。

3）build_brief

这里开始把任务正式定型。

重点是确定当前到底是：

• selling_point_image
• seeding_image

也就是卖点图还是种草图。

4）build_presentation_plan

这一步是我自己很看重的一层。

它不直接生成 prompt，而是先决定：

• 该怎么展示
• 应不应该有人
• 使用场景该落在哪
• 哪些场景明确不合理

5）create_scene

拿着展示策略再生成营销场景，不是凭空飘。

6）select_strategy

这里决定本轮走哪种生成策略，比如是不是强依赖参考图。

7）build_prompt_package

把 prompt 拆成：

• fidelity_prompt
• composition_prompt
• final_prompt

8）generate_image

这一轮真正出图。

9）critique_image

看图到底过不过。

10）update_attempt_record

把本轮结果记下来，并决定下一步是继续还是收口。

这条链的好处就在这里：每一步只管自己那一层，不把所有判断都挤进 prompt 生成里。

6. 重试闭环真正收口的地方，在 update_attempt_record

这一段我这次看得比较认真。

很多项目会把“重试”理解成：

• 出图
• 分数不够
• 再来一张

但 DAMA 这里不是这么粗暴。

真正控制重试语义的，是 update_attempt_record(...) 加 flow_runtime.py。

_update_attempt_record(state) 会先把当前 attempt 收起来：

attempt = AttemptRecord(
    attempt_index=state["attempt_index"],
    prompt=state["latest_prompt"],
    image=state["latest_image"],
    critique=state["latest_critique"],
)

然后调用：

progress = update_attempt_state(state.get("attempts", []), attempt, state.get("best_attempt"))

这里不是简单 append 一下就完了，后面还会继续做一轮判断。

flow_runtime.py 里会一起处理几件事：

• 当前 attempt 追加进历史
• 跟 best_attempt 做比较
• 看当前 Critic 给的 retry_action
• 决定要不要停
• 决定是否保护当前最好结果
• 把下一轮要用的 critique_feedback 抽出来

所以这套重试其实已经不是“随机再抽一次”。

它更像是：

Critic 会先给出任务级判断，runtime 再根据这个判断决定下一轮到底怎么走。

7. 分支也不复杂，重试时只有两种主要回退路径

update_attempt_record 之后，graph 会走一个条件分支：

graph.add_conditional_edges(
    "update_attempt_record",
    _decide_next_step,
    {
        "retry_prompt_only": "build_prompt_package",
        "retry_with_strategy_shift": "select_strategy",
        "finalize": "finalize_result",
    },
)

这个设计我挺喜欢，原因是它很克制。

它没有搞出太多复杂分支，而是只保留了三种最核心的结果：

1）retry_prompt_only

策略不用变，直接回到 build_prompt_package

说明当前问题更像是 prompt 表达层问题。

2）retry_with_strategy_shift

回到 select_strategy

说明问题不是一句 prompt 能修好的，得换一套生成策略再试。

3）finalize

流程到头，直接收口。

这三种语义已经够用了。

至少从工程上看，它把“为什么重试”“要退回到哪层重试”这件事说清楚了。

8. finalize_result(...) 这里不只是返回最后一轮，它还要选最终可交付结果

我一开始还以为 finalize 就是把最后一轮结果吐回去。

后来看到这里，发现不是。

_finalize_result(state) 会先拿：

• latest_attempt
• best_attempt
• approved
• retry_action

然后通过 select_final_attempt(...) 决定最后到底返回哪一轮。

这个点挺重要。

因为真实场景里经常会有一种情况：

• 后一轮没完全通过
• 但前一轮其实更稳
• 如果硬返回最后一轮，结果反而更差

所以这里的收口逻辑，不是机械地拿最后一次，而是从当前链路里挑一轮最合适的结果。

这也解释了为什么 state 里要一直保留 best_attempt。

9. context_bundle 这层，让 graph 跑节点时也能把任务现场一起带上

这个项目里还有一层我现在越来越看重，就是 context_bundle。

它相关的处理主要在 runtime_state.py。

这里有两个动作：

1）从 session state 构建 context bundle

build_context_bundle_from_state(...)

它会把这些东西整理进去：

• 用户最近要求
• 明确提到的场景和人物要求
• 商品身份锁信息
• 当前 task plan
• attempt runtime
• critic policy

2）每过一个节点就做一次 snapshot

snapshot_context_bundle(...)

比如 interpretation、brief、presentation plan、strategy、prompt package、latest critique 这些更新后，都会同步写回快照。

这个设计对我来说挺有感觉。

因为它说明这里的上下文已经不是“前面几轮对话文本”那么简单了，而是在往一个更像执行现场快照的方向走。

后面前端如果要展示 thinking flow、后端如果要做 resume、历史回放如果要还原过程，这层都会很有用。

10. dispatcher.py 才是这条 graph 和真实会话状态接起来的那层胶水

如果只看 graph，很容易觉得这条链已经完了。

其实还差一层：graph 跑出来的东西，怎么回到真实 session 里。

这块主要在 dispatcher.py。

生成时，入口是：

flow = session.agent.run_selling_point_image(...)

或者：

flow = session.agent.run_seeding_image(...)

跑完以后，不是直接把结果丢给前端，而是先走：

self._update_state_from_selling_point_flow(session, flow)

然后内部再调用：

apply_flow_result_to_state(session.state, flow, file_url_builder=...)

这一步会把 graph 产出的东西重新写回 session state，比如：

• product_analysis
• ecommerce_interpretation
• selling_point_brief
• presentation_plan
• latest_strategy_decision
• latest_prompt_package
• latest_generated_image
• latest_critique
• attempt_history

到这里，这条 graph 才算真正进入项目主状态。

所以如果从工程角度看，我现在会把它拆成两层理解：

• feature_image_graph.py：组织一次任务内部怎么跑
• dispatcher.py + runtime_state.py：把这次运行接回会话系统

两层缺一不可。

11. 现在回头看，这条 LangGraph 工作流真正帮我解决了三件事

这篇拆到这里，我自己最明确的感受是，它帮我把三件以前很容易混在一起的东西慢慢分开了。

第一件：把“商品理解”和“图像生成”分开

不再一上来就写 prompt，而是先理解商品、理解任务。

第二件：把“表达问题”和“策略问题”分开

不是所有失败都靠多改一句 prompt 来救。

第三件：把“这次生成结果”和“整轮任务状态”分开

最终结果只是结果，真正决定系统能不能持续优化的，是过程状态有没有被保住。

这也是我现在越来越喜欢 LangGraph 这类东西的原因。

我喜欢它，不是冲着名字新去的。主要还是因为这种项目一旦进入多阶段、多状态、多轮修正，它确实比手写串行逻辑更容易把问题摆清楚。

12. 这条链后面还能继续往哪补

当然，这套工作流现在也不是完全收口了。

我自己接下来还会继续盯几块：

• run / attempt / event 和持久化链路继续打通
• SSE 把中间步骤更完整地投给前端
• durable worker / queue 把后台执行从 demo 态再往前推一步
• context bundle 和历史恢复继续统一

但至少到这一步，我已经能比较明确地说：

DAMA 这套 LangGraph 工作流，已经不只是“模型出图流程”了。现在看起来，它更像是一条围绕商品理解、任务分型、策略决策、结果评价和状态续跑组织起来的执行链。

这也是我现在更愿意把它当成“智能体工作流”来看，而不是“多调了几个模型接口”的原因。