How to Apply AI Graph Engineering Across the Development Pipeline—from Requirements to Deployment

Many teams adopt AI tools in an ad‑hoc way, ending up with scattered utilities that add little efficiency and increase maintenance overhead. Graph engineering solves this by stitching AI capabilities into a single, coordinated pipeline that spans the entire development lifecycle.

Typical Development Pipeline and Pain Points

Requirement Review : Inconsistent document quality leads to ambiguous specifications.

Technical Specification : Experience‑based reviews miss edge cases and security risks.

Code Implementation : Repetitive code generation with unstable AI output quality.

Code Review : Bottleneck as senior engineers spend excessive time on low‑value PRs.

Testing : Incomplete coverage and overlooked boundary conditions.

Release : Manual pre‑deployment checks prone to omission.

Graph nodes are introduced for the stages that can be split into specialized sub‑tasks requiring independent verification.

Node Selection Criterion

A stage is suitable when the work can be divided into sub‑tasks with distinct expertise and when human validation is still required. For example, code review is split into three parallel nodes—security, performance, and style—so each expert can focus on a single dimension without distraction.

Overall Architecture

The pipeline graph contains two Human‑in‑the‑Loop (HITL) nodes highlighted in yellow: the technical‑spec review and the test‑result confirmation. AI generates suggestions, and humans make the final decision.

State Design

from typing import TypedDict, Annotated, Optional

class DevPipelineState(TypedDict):
    # ── 需求信息 ──────────────────────────────────
    requirement_doc: str  # 原始需求文档
    parsed_requirements: dict  # 解析后的结构化需求
    acceptance_criteria: list[str]  # 验收标准
    # ── 技术方案 ──────────────────────────────────
    tech_spec: str  # 技术方案文档
    tech_risks: list[str]  # 技术风险点
    human_spec_feedback: Optional[str]  # 人工评审意见
    # ── 代码实现 ──────────────────────────────────
    code_files: dict  # 生成的代码文件 {filename: content}
    implementation_notes: str  # 实现说明
    # ── 代码审查（并行，用 Reducer 追加） ──────────
    review_results: Annotated[list[dict], operator.add]
    # ── 质量评估 ──────────────────────────────────
    quality_gate_passed: bool
    quality_issues: list
    revision_count: int
    # ── 测试 ──────────────────────────────────────
    test_cases: list
    test_results: Optional[dict]
    human_test_feedback: Optional[str]
    # ── 发布 ──────────────────────────────────────
    release_checklist: list[str]
    release_notes: str
    deploy_ready: bool

Requirement Parsing Node

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0.2)

def parse_requirement_node(state: DevPipelineState) -> dict:
    """需求解析节点：把需求文档转化成结构化需求"""
    prompt = f"""
    你是一个资深产品经理，擅长需求分析。
    请分析以下需求文档，输出结构化需求：
    【需求文档】{state['requirement_doc']}
    输出格式为 JSON：{{
      "functional_requirements": [...],
      "non_functional_requirements": [...],
      "acceptance_criteria": [...],
      "ambiguities": [...],
      "edge_cases": [...]
    }}
    """
    result = llm.invoke(prompt)
    parsed = parse_json_safely(result.content)
    return {
        "parsed_requirements": parsed,
        "acceptance_criteria": parsed.get("acceptance_criteria", [])
    }

Technical Specification Node

def generate_tech_spec_node(state: DevPipelineState) -> dict:
    """技术方案节点：基于需求生成技术方案"""
    requirements = state["parsed_requirements"]
    prompt = f"""
    你是一个资深架构师。
    基于以下需求，生成技术方案：
    【功能需求】{format_list(requirements.get('functional_requirements', []))}
    【非功能需求】{format_list(requirements.get('non_functional_requirements', []))}
    【边界情况】{format_list(requirements.get('edge_cases', []))}
    {f"【上一版本的评审意见，请针对性修改】
{state['human_spec_feedback']}" if state.get('human_spec_feedback') else ''}
    输出包括整体架构、核心数据模型、API 设计、技术选型说明、技术风险点、实现步骤。
    """
    result = llm.invoke(prompt)
    risks = extract_tech_risks(result.content)
    return {"tech_spec": result.content, "tech_risks": risks}

Implementation Node

def implement_node(state: DevPipelineState) -> dict:
    """代码实现节点：基于技术方案生成代码"""
    llm_coder = ChatOpenAI(model="gpt-4o", temperature=0.1)
    review_context = ""
    if state.get("quality_issues"):
        review_context = f"【上一版本的代码审查问题，必须修复】
{chr(10).join(state['quality_issues'])}"
    prompt = f"""
    你是一个资深工程师。
    基于以下技术方案，生成实现代码（前3000字符）：
{state['tech_spec'][:3000]}
    【验收标准】{format_list(state['acceptance_criteria'])}
    {review_context}
    要求：代码完整可运行，函数注释清晰，处理边界情况，遵循 SOLID 原则。
    输出格式：```python
# 文件名：xxx.py
[代码内容]```"""
    result = llm_coder.invoke(prompt)
    code_files = parse_code_files(result.content)
    return {
        "code_files": code_files,
        "implementation_notes": extract_implementation_notes(result.content),
        "revision_count": state.get("revision_count", 0) + 1
    }

Parallel Code Review Nodes

def security_review_node(state: DevPipelineState) -> dict:
    """安全审查节点：专注安全漏洞"""
    llm_security = ChatOpenAI(model="gpt-4o", temperature=0)
    code_content = format_code_files(state["code_files"])
    prompt = f"""
    你是一个专注安全的代码审查专家。
    只关注安全问题，不评价代码风格或性能。
    审查以下代码的安全性：
{code_content}
    检查项：SQL 注入、XSS、身份验证、敏感数据、输入验证、不安全依赖。
    输出 JSON：{{"review_type": "security", "issues": [...], "passed": true/false}}
    """
    result = llm_security.invoke(prompt)
    review = parse_json_safely(result.content)
    review["review_type"] = "security"
    return {"review_results": [review]}

def performance_review_node(state: DevPipelineState) -> dict:
    """性能审查节点：专注性能问题"""
    llm_perf = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    code_content = format_code_files(state["code_files"])
    prompt = f"""
    你是一个专注性能优化的代码审查专家。
    只关注性能问题，不评价安全或代码风格。
    审查以下代码的性能：
{code_content}
    检查项：N+1 查询、循环嵌套、缺少缓存、内存泄漏、同步阻塞、大数据量处理。
    输出 JSON：{{"review_type": "performance", "issues": [...], "passed": true/false}}
    """
    result = llm_perf.invoke(prompt)
    review = parse_json_safely(result.content)
    review["review_type"] = "performance"
    return {"review_results": [review]}

def style_review_node(state: DevPipelineState) -> dict:
    """代码风格审查节点：专注代码规范"""
    llm_style = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    code_content = format_code_files(state["code_files"])
    prompt = f"""
    你是一个代码规范审查专家。
    只关注代码风格和可维护性，不评价安全或性能。
    审查以下代码的规范性：
{code_content}
    检查项：命名规范、函数长度、注释质量、重复代码、错误处理、代码复杂度。
    输出 JSON：{{"review_type": "style", "issues": [...], "passed": true/false}}
    """
    result = llm_style.invoke(prompt)
    review = parse_json_safely(result.content)
    review["review_type"] = "style"
    return {"review_results": [review]}

Quality Gate Node

def quality_gate_node(state: DevPipelineState) -> dict:
    """质量门禁：汇总三个审查结果，决定是否通过"""
    reviews = state["review_results"]
    all_issues = []
    has_high_severity = False
    for review in reviews:
        for issue in review.get("issues", []):
            all_issues.append(f"[{review['review_type'].upper()}] {issue['severity'].upper()}: {issue['description']}")
            if issue.get("severity") == "high":
                has_high_severity = True
    passed = (not has_high_severity and len(all_issues) <= 10) or state.get("revision_count", 0) >= 3
    return {
        "quality_gate_passed": passed,
        "quality_issues": [] if passed else all_issues
    }

Test Generation Node

def generate_tests_node(state: DevPipelineState) -> dict:
    """测试节点：基于验收标准和代码生成测试用例"""
    llm_tester = ChatOpenAI(model="gpt-4o", temperature=0.2)
    prompt = f"""
    你是一个测试工程师，擅长编写全面的测试用例。
    基于以下信息生成测试用例：
    【验收标准】{format_list(state['acceptance_criteria'])}
    【代码实现】{format_code_files(state['code_files'])[:2000]}
    覆盖：正常流程、边界值、异常情况、 安全、性能。
    每个用例包含名称、前置条件、步骤、预期结果、pytest 代码。
    输出 JSON 列表。
    """
    result = llm_tester.invoke(prompt)
    test_cases = parse_json_safely(result.content)
    return {"test_cases": test_cases if isinstance(test_cases, list) else []}

Release Checklist Node

def generate_release_checklist_node(state: DevPipelineState) -> dict:
    """发布节点：生成发布清单和说明"""
    llm_release = ChatOpenAI(model="gpt-4o-mini", temperature=0.1)
    prompt = f"""
    基于以下信息生成发布清单和说明：
    【需求摘要】{format_list(state['acceptance_criteria'][:5])}
    【技术风险点】{format_list(state['tech_risks'])}
    【代码审查问题（已修复）】{format_list(state.get('quality_issues', [])[:5])}
    生成：
    1. 至少 10 条检查清单（数据库迁移、配置变更、依赖更新、回滚方案）
    2. 100 字以内的发布说明
    3. 关键监控指标
    4. 回滚方案
    输出 JSON。
    """
    result = llm_release.invoke(prompt)
    release_info = parse_json_safely(result.content)
    return {
        "release_checklist": release_info.get("checklist", []),
        "release_notes": release_info.get("release_notes", ""),
        "deploy_ready": True
    }

Graph Assembly

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
from langfuse.callback import CallbackHandler

builder = StateGraph(DevPipelineState)
# 添加节点
builder.add_node("parse_req", parse_requirement_node)
builder.add_node("gen_spec", generate_tech_spec_node)
builder.add_node("implement", implement_node)
builder.add_node("security_review", security_review_node)
builder.add_node("performance_review", performance_review_node)
builder.add_node("style_review", style_review_node)
builder.add_node("quality_gate", quality_gate_node)
builder.add_node("gen_tests", generate_tests_node)
builder.add_node("gen_release", generate_release_checklist_node)
# 主流程
builder.set_entry_point("parse_req")
builder.add_edge("parse_req", "gen_spec")
builder.add_edge("gen_spec", END)  # HITL 暂停
# 实现后并行审查
builder.add_edge("implement", "security_review")
builder.add_edge("implement", "performance_review")
builder.add_edge("implement", "style_review")
# 汇聚审查结果
builder.add_edge("security_review", "quality_gate")
builder.add_edge("performance_review", "quality_gate")
builder.add_edge("style_review", "quality_gate")
# 条件路由
builder.add_conditional_edges(
    "quality_gate",
    route_after_quality_gate,
    {"generate_tests": "gen_tests", "fix_issues": "implement"}
)
builder.add_edge("gen_tests", END)  # HITL 暂停
builder.add_edge("gen_release", END)
# 编译并使用 SQLite 持久化
checkpointer = SqliteSaver.from_conn_string("dev_pipeline.db")
app = builder.compile(checkpointer=checkpointer, interrupt_after=["gen_spec", "gen_tests"])

Full Run Example (Human‑in‑the‑Loop)

def run_dev_pipeline(requirement_doc: str, pr_id: str):
    """运行研发流水线"""
    config = {"configurable": {"thread_id": f"pr-{pr_id}"}, "callbacks": [CallbackHandler(public_key=os.getenv("LANGFUSE_PUBLIC_KEY", ""), secret_key=os.getenv("LANGFUSE_SECRET_KEY", ""))]}
    print(f"🚀 开始处理需求，PR: {pr_id}")
    # 阶段一：需求解析 + 技术方案
    state = app.invoke({"requirement_doc": requirement_doc, "review_results": [], "revision_count": 0}, config=config)
    print("
📋 技术方案已生成，等待评审…")
    print(state['tech_spec'][:500] + "…")
    for risk in state.get("tech_risks", []):
        print(f"  - {risk}")
    decision = input("
[a]批准方案 / [m]修改后批准 / [r]打回重写 > ").strip()
    if decision == "r":
        feedback = input("请输入修改意见：")
        app.update_state(config, {"human_spec_feedback": feedback})
        state = app.invoke(None, config=config)
    elif decision == "m":
        feedback = input("请输入修改意见（AI 会基于此修改）：")
        app.update_state(config, {"human_spec_feedback": feedback})
    # 阶段二：代码实现 + 审查
    print("
⚙️  开始代码实现和审查…")
    app.update_state(config, {}, as_node="gen_spec")
    state = app.invoke(None, config=config)
    print("
✅ 代码审查完成")
    print(f"质量门禁：{'通过' if state['quality_gate_passed'] else '未通过'}")
    if state.get("quality_issues"):
        print("
⚠️  已自动修复的问题：")
        for issue in state["quality_issues"][:5]:
            print(f"  - {issue}")
    print(f"
🧪 测试用例已生成：{len(state.get('test_cases', []))} 个")
    test_decision = input("
[a]确认测试通过 / [r]发现问题，打回修改 > ").strip()
    if test_decision == "r":
        feedback = input("请描述测试发现的问题：")
        app.update_state(config, {"human_test_feedback": feedback, "quality_issues": [feedback]})
    # 阶段三：生成发布清单
    app.update_state(config, {}, as_node="gen_tests")
    final_state = app.invoke(None, config=config)
    print("
🚀 发布清单已生成")
    print(f"
📝 发布说明：{final_state.get('release_notes', '')}")
    print(f"
✅ 发布前检查清单（{len(final_state.get('release_checklist', []))} 项）：")
    for item in final_state.get('release_checklist', [])[:5]:
        print(f"  □ {item}")
    return final_state

Practical Roll‑out Phases

Phase 1 (1‑2 weeks) : Deploy only the parallel code‑review graph. Integrate with GitHub Actions to auto‑comment PRs.

Phase 2 (≈1 month) : Add the requirement‑parsing node to produce structured requirements and acceptance criteria.

Phase 3 (after stabilization) : Introduce the test‑generation node so testers become reviewers of AI‑generated tests.

Real‑World Impact (10‑person team)

Average PR review wait time dropped from 4.2 hours to 8 minutes (AI initial review).

Senior engineer review time reduced from 2.5 hours/day to 45 minutes/day .

Security issues leaking to production fell from ≈15 % to ≈3 % .

Monthly OpenAI API cost ≈ ¥600, offset by saved engineering time.

Common Pitfalls & Solutions

Context overflow in code‑generation node : Pass only the core architecture and API design, keep data‑model separate.

Overly broad review prompts : Begin each prompt with “Only focus on X, do not evaluate other aspects.”

Too strict quality‑gate rules : Block only on high‑severity issues; treat medium/low as suggestions.

Key Takeaways

Graph engineering coordinates AI across the dev pipeline; it is not about automating every step.

Parallel security, performance, and style review nodes provide the easiest entry point.

Two HITL nodes—technical‑spec review and test‑result confirmation—remain essential.

Implement routing logic with explicit if/else rather than delegating decisions to LLMs.

Adopt a phased rollout to balance benefit against maintenance cost.