如何用 LangGraph 搭建 awesome-llm-apps 代码库迁移 Agent：文件级重构计划、人工审批门与并行 Worker

awesome-llm-apps 仓库中的 ai_codebase_migration_agent 目录提供了一套可以直接运行的 LangGraph 多 Agent 参考实现：输入一个仓库目标和迁移目标，Planner 先生成文件级重构计划，图在审批节点暂停等待人工确认，确认后按文件扇出并行 Worker 产出 diff，最后汇总成带风险图表的 markdown 迁移报告。它的产物是报告和 diff，不会自动改动你的仓库。本文按 app.py 的实际代码，把这条链路拆成可对照搭建的步骤，适合想复用这套「计划 → 审批 → 并行执行 → 汇总」结构的开发者。

先看整体链路：六段式图结构

README 和 app.py 头部注释给出的图流程一致：

Repo Target + Goal → Validator → Planner (Strategy + File Tasks) → HITL Approval
→ Parallel Refactor Workers → Aggregator → Migration Guide & Diffs

app.py 中的边定义等价于：

START -> query_validator -> planner -> approval (HITL interrupt)
       -> [Send() fan-out] -> refactor_worker xN -> aggregator -> END

目录内文件构成：

文件	作用
app.py	图节点实现、图构建与 Streamlit UI
requirements.txt	依赖清单
test_app_security.py	针对图表工具与代码执行的测试

图本身不依赖 Streamlit：app.py 提供 build_migration_graph()，可以脱离 UI 单独导入调用，这也是二次开发时的主要入口。

准备：依赖与模型配置

进入 advanced_ai_agents/multi_agent_apps/ai_codebase_migration_agent 目录后安装依赖：

pip install -r requirements.txt

requirements.txt 中的核心依赖（版本要求）：

langgraph>=1.2.6
langgraph-checkpoint>=4.1.1
langchain>=1.3.11
langchain-core>=1.4.8
langchain-openai>=1.3.3
pydantic>=2.13.4
python-dotenv>=1.2.2
streamlit
matplotlib

模型凭据支持两种配置方式，README 给出 .env 形式（sk-your-key-here 需替换为你自己的 key）：

OPENAI_API_KEY=sk-your-key-here

app.py 在启动时执行 load_dotenv()，读取的 key 优先级为 LLM_API_KEY 或 OPENAI_API_KEY；应用侧边栏也可以输入 key、MODEL_FAST、MODEL_PRO 和 LLM_BASE_URL 并点击 Save Settings 在运行时覆盖。默认值如下（app.py 源码常量）：

DEFAULT_MODEL_FAST = "gpt-5-mini"  # validation, planning, per-file refactor workers
DEFAULT_MODEL_PRO = "gpt-5.5"  # final report synthesis + chart generation
LLM_TIMEOUT = 300  # seconds for each provider request

也就是说：校验、规划、逐文件 Worker 都走 MODEL_FAST（默认 gpt-5-mini），最终报告汇总走 MODEL_PRO（默认 gpt-5.5）；额外设置 LLM_BASE_URL 可以指向任意 OpenAI 兼容端点。

第一步：定义图状态与文件级任务结构

图状态 MigrationState 决定了每个节点的输入输出契约，其中 status 的五个取值（planning / awaiting_approval / refactoring / completed / error）就是后续判断流程走到哪一步的依据：

class MigrationState(TypedDict):
    repo_target: str
    migration_goal: str
    strategy: str
    plan: List[dict]
    plan_approved: bool
    user_feedback: Optional[str]
    status: Literal[
        "planning", "awaiting_approval", "refactoring", "completed", "error"
    ]
    results: Annotated[List[str], operator.add]
    final_answer: Optional[str]
    error: Optional[str]

results 使用 operator.add 归约器，多个并行 Worker 写回的逐文件结果会被追加合并，而不是互相覆盖。

文件级任务用 Pydantic 模型 MigrationTask 描述，risk_level 被限定为四档枚举：

class MigrationTask(BaseModel):
    file_path: str = Field(
        description="Relative path of the target file to migrate (e.g. 'models/user.py', 'src/api.js')"
    )
    action: str = Field(
        description="Specific migration action (e.g., 'Update Pydantic v1 BaseModel & @validator to v2 Field & @field_validator')"
    )
    risk_level: Literal["Low", "Medium", "High", "Critical"] = Field(
        description="Estimated risk level of refactoring this file"
    )
    risk_reasoning: str = Field(
        description="Detailed explanation of why this risk level was assigned"
    )
    code_context: Optional[str] = Field(
        default=None,
        description="Optional existing code snippet or context for the file",
    )

第二步：输入校验与文件级重构计划

query_validator 节点分两层做校验：

1. 纯代码检查：repo_target 或 migration_goal 为空时直接返回 status: "error"，错误信息为 "Please provide both a target repository URL/path AND a clear migration goal."，图随后走条件边直接进入 END，不会消耗 LLM 调用。
2. LLM 结构化校验：用 QueryValidation（is_valid / error_message）判断请求是否实质、安全；若调用抛出异常且错误信息包含 api key、401、403 等关键字，UI 会显示 "API key is missing or invalid. Open the sidebar to configure your API keys."

planner_node 使用 MODEL_FAST 的 with_structured_output(MigrationPlanResponse, method="json_mode")。其系统提示要求：给出 1–3 句的架构策略摘要，并生成 3–6 个文件迁移任务，每个任务含 file_path、action、risk_level、risk_reasoning；若状态里带有上一轮的 user_feedback，planner 会把反馈和旧计划一起拼进提示词重新规划。输出约束由模型类兜底：

class MigrationPlanResponse(BaseModel):
    strategy: str = Field(description="High-level migration strategy summary")
    tasks: List[MigrationTask] = Field(
        description="List of file migration tasks with risk assessments",
        min_length=1,
        max_length=8,
    )

注意提示词要求 3–6 个任务，而 Pydantic 约束是 1–8 个，后续 Worker 扇出也按 8 个封顶（见下文）。计划生成成功后节点把 status 置为 awaiting_approval。

第三步：人工审批门——用 interrupt() 暂停图

审批节点 plan_approval 是整条链路里唯一需要人的地方。核心是 LangGraph 的 interrupt()：

def plan_approval(state: MigrationState) -> dict:
    user_response = interrupt("Waiting for migration plan approval/feedback")

    feedback = (
        user_response.get("message", "")
        if isinstance(user_response, dict)
        else str(user_response)
    )
    # ... 用 APPROVAL_SYSTEM_PROMPT 对 feedback 做结构化分类 ...
    status = "refactoring" if result.plan_approved else "planning"
    return {
        "plan_approved": result.plan_approved,
        "user_feedback": feedback,
        "status": status,
    }

工作流程是：interrupt() 挂起图并弹出审批消息；恢复时传入的 resume 值是一个含 message 键的字典（Streamlit UI 中即 Command(resume={"message": message})）。随后 APPROVAL_SYSTEM_PROMPT 把这句话分类为「明确批准」（如 approve、proceed、yes）或「要求修改」（如增删文件、调整风险级别），前者进入 refactoring，后者带着 user_feedback 回到 planning 重新走 planner。

路由逻辑还处理了一个边界：

def route_after_approval(state: MigrationState):
    # An approved-but-empty plan would fan out to zero workers and strand the
    # graph before the aggregator, so send it back to the planner instead.
    if state.get("plan_approved") and state.get("plan"):
        return dispatch_workers(state)
    return "planner"

已批准但计划为空的情况会打回 planner，避免零 Worker 扇出把图卡死在聚合之前。图以 MemorySaver() 编译（builder.compile(checkpointer=MemorySaver())），README 说明这一内存 checkpoint 使中断状态能跨 Streamlit 的 rerun 存活。

第四步：Send() 并行 Worker 扇出

批准后的分发函数为每个计划任务发一个 Send，每任务一个 Worker：

def dispatch_workers(state: MigrationState) -> List[Send]:
    """Fan-out to one refactor worker per file in the approved migration plan."""
    return [
        Send("refactor_worker", {**state, "file_task": task})
        for task in state["plan"][:8]
    ]

state["plan"][:8] 表明并行分支最多 8 条，与 planner 的 max_length=8 对应。

refactor_worker_node 开头有一段错峰逻辑，源码注释写明目的：

    # Stagger worker execution to avoid API rate limit spikes
    stagger_time = random.uniform(0.3, 1.5)
    time.sleep(stagger_time)

每个 Worker 用 MODEL_FAST 针对单个文件生成 markdown 输出（重构拆解、unified diff 或 before/after 代码块、破坏性变更与缓解措施、建议的单测），封装成 ### File: <file_path> 段落写入 results。单个文件出错不会中断整体：异常被捕获后，该文件在 results 里写入一段 "Refactoring Failed" 块，聚合节点照常汇总。

第五步：汇总节点与只收数据的图表工具

aggregator_node 使用 MODEL_PRO（llm.pro().bind_tools([generate_matplotlib_chart])），最多循环 MAX_AGGREGATOR_ROUNDS = 6 轮工具调用：

MAX_AGGREGATOR_ROUNDS = 6

图表工具 generate_matplotlib_chart 只接受数据（chart_type 为 bar/line、labels、values、title、y_label），不执行任何模型生成的 Python；它做长度、类型、有限数值等校验，通过后把 PNG 以 base64 data URI 形式返回。聚合流程用 <!--CHART_1--> 之类占位符接收工具返回，最后替换为图片标记。如果多轮结束后没有任何助手正文（例如第一轮就超时），节点返回 status: "error"，错误信息明确要求 "Re-run the approval step to try again"。

组装图并运行

五个节点与条件边组装如下（app.py 原文，仅省略空行）：

def build_migration_graph():
    """Build and compile the codebase migration graph (pure LangGraph logic)."""
    builder = StateGraph(MigrationState)

    builder.add_node("query_validator", query_validator)
    builder.add_node("planner", planner_node)
    builder.add_node("approval", plan_approval)
    builder.add_node("refactor_worker", refactor_worker_node)
    builder.add_node("aggregator", aggregator_node)

    def route_after_approval(state: MigrationState):
        if state.get("plan_approved") and state.get("plan"):
            return dispatch_workers(state)
        return "planner"

    builder.add_edge(START, "query_validator")
    builder.add_conditional_edges(
        "query_validator",
        lambda state: END if state.get("error") else "planner",
        ["planner", END],
    )
    builder.add_conditional_edges(
        "planner",
        lambda state: END if state.get("error") else "approval",
        ["approval", END],
    )
    builder.add_conditional_edges(
        "approval", route_after_approval, ["refactor_worker", "planner"]
    )
    builder.add_edge("refactor_worker", "aggregator")
    builder.add_edge("aggregator", END)

    return builder.compile(checkpointer=MemorySaver())

启动应用：

streamlit run app.py

界面操作路径：

1. 侧边栏填写 API Key、Fast/Pro 模型与 Base URL（或直接依赖 .env），点击 Save Settings；
2. 主区域输入 "Target Repository Path or URL"（占位示例：github.com/my-org/my-project or ./src/my_app）与 "Migration Goal & Scope"（占位示例：Migrate Pydantic v1 BaseModel to v2 Field & @field_validator across all services），点击 Plan Migration；
3. 页面进入审批视图：显示策略摘要和文件级风险矩阵（Critical/High/Medium/Low 徽章），下方是 "Safety & Approval Gate (HITL)"。在 "Revision instructions" 输入框中写修改意见（界面给出的示例：mark models/user.py as Critical risk、skip config/settings.py）后点击 Revise Migration Plan 会重新规划；点击 Approve & Execute Migration Plan 则恢复图继续执行；
4. 执行期间状态框逐条记录每个 refactor_worker 的完成片段和聚合器的汇总进度；
5. 完成后页面渲染迁移报告（含 base64 内嵌图表），并提供 "Download Migration Report (.md)" 下载按钮，文件名形如 migration_plan_<thread_id>.md。侧边栏的 "New Migration Plan" 按钮会清空会话状态开始新一轮。

验证方式与已知边界

判断任务是否按预期走通，可以对照这几处文档/代码中明确给出的信号：

• 状态值：status 依次经过 awaiting_approval、refactoring，成功时到 completed 并产出 final_answer；任何节点失败都会落 error 并带具体错误文本。
• API key 问题：校验节点捕获认证类异常后返回固定提示 "API key is missing or invalid. Open the sidebar to configure your API keys."，这是最常见的启动失败现象。
• 汇总为空：聚合器超时或只产生工具调用时，错误信息要求重新走审批步骤重试。
• 安全测试：test_app_security.py 含三条断言，可用 pytest 运行（注意 requirements.txt 未包含 pytest，需自行安装）：
  • generate_matplotlib_chart 的参数 schema 只含 chart_type、labels、values、title、y_label，不接受任何源码参数；
  • labels 与 values 长度不一致时返回 "Error rendering chart: labels and values must have the same length."；
  • app.py 源码经 AST 检查不含任何 exec 调用。

限制条件（均来自源码）：并行分支上限 8 个文件任务；每个 Worker 开始前随机 sleep 0.3–1.5 秒以错峰 API 调用；checkpoint 用内存版 MemorySaver，会话状态随进程存在。README 列出的适用场景包括 Pydantic v1 → v2、Flask 2 → 3、SQLAlchemy 1.4 → 2.0、JavaScript → TypeScript 转换、同步 I/O 改 async/await 以及废弃 API 清理，可作为你搭建同类 Agent 时的第一批迁移目标。