用 LangGraph 搭一套企业级 Coding Workflow，聊聊我的思路

公司前一阵子在折腾 LangGraph，想把团队里从需求到上线的整个编码流程串起来。传统的 CI/CD 管道大家都熟，但如果把 LLM 能力嵌进去，让每个环节都带上"智能判断"，体验完全不一样。这里分享下我的方案和踩坑记录。

为什么选 LangGraph

LangChain 的 Chain 模式是线性的，而真实的开发流程充满分支——代码写完可能要打回重写，合规审查不过要回到开发阶段。LangGraph 天然支持有向图 + 条件路由，节点之间可以灵活跳转，非常适合这种有状态、有循环的工作流。

整体架构

我把流程拆成五个核心节点：

需求分析 → 代码开发 → Code Review → 合规审查 → 部署运维
              ↑            |              |
              └── 打回 ────┘              |
              ↑                           |
              └────── 合规不通过 ─────────┘

关键在于两个回路：Review 不通过回开发，合规不通过也回开发。这就是 LangGraph 比纯 Chain 强的地方。

核心代码

from langgraph.graph import StateGraph, END
from typing import TypedDict

class WorkflowState(TypedDict):
    requirement: str
    code: str
    review_passed: bool
    compliance_passed: bool
    deploy_status: str
    retry_count: int

def develop(state: WorkflowState) -> WorkflowState:
    """代码开发节点，基于需求生成/修改代码"""
    # 实际项目里这里调 LLM 生成代码
    state["code"] = f"generated code for: {state['requirement']}"
    return state

def code_review(state: WorkflowState) -> WorkflowState:
    """Code Review 节点，检查代码质量、规范、安全漏洞"""
    # LLM 做静态分析 + 规则引擎兜底
    state["review_passed"] = True  # 简化示意
    return state

def compliance_check(state: WorkflowState) -> WorkflowState:
    """合规审查，检查许可证、数据隐私、行业法规"""
    state["compliance_passed"] = True
    return state

def deploy(state: WorkflowState) -> WorkflowState:
    """部署上线"""
    state["deploy_status"] = "deployed"
    return state

def monitor(state: WorkflowState) -> WorkflowState:
    """运维监控，异常时可触发回滚"""
    state["deploy_status"] = "running_stable"
    return state

# 条件路由
def after_review(state: WorkflowState):
    return "compliance_check" if state["review_passed"] else "develop"

def after_compliance(state: WorkflowState):
    return "deploy" if state["compliance_passed"] else "develop"

# 构建图
graph = StateGraph(WorkflowState)
graph.add_node("develop", develop)
graph.add_node("code_review", code_review)
graph.add_node("compliance_check", compliance_check)
graph.add_node("deploy", deploy)
graph.add_node("monitor", monitor)

graph.set_entry_point("develop")
graph.add_edge("develop", "code_review")
graph.add_conditional_edges("code_review", after_review)
graph.add_conditional_edges("compliance_check", after_compliance)
graph.add_edge("deploy", "monitor")
graph.add_edge("monitor", END)

workflow = graph.compile()

# 执行
result = workflow.invoke({
    "requirement": "实现用户数据导出接口",
    "code": "", "review_passed": False,
    "compliance_passed": False, "deploy_status": "",
    "retry_count": 0
})