预计阅读时间：13 分钟
大型遗留系统的迁移，一直是工程团队最头疼的事——动辄数年、牵一发动全身、风险高到没人敢拍板。David Stein 在 ServiceTitan 的实践中，把这件事彻底换了一种打法：不靠少数架构师苦战，而是用 AI 把迁移拆成标准化工序，像工厂流水线一样大规模并行推进。关键在于两件事——任务分解的粒度要足够细、足够标准，以及验证回路必须程序化且刚性，不给 LLM 的幻觉留任何缝隙。
遗留迁移的真正瓶颈在哪
传统迁移的慢，不是因为代码多，而是因为每一步都依赖人的判断。架构师需要读懂旧系统、设计新结构、手动验证每一处改动——串行推进，瓶颈永远在少数人身上。更致命的是，人在高压下容易漏判，一次回归缺陷就可能让整个迁移倒退数周。
ServiceTitan 的核心洞察是：迁移本身可以像制造业一样工业化。不是让 AI 替你思考架构，而是让 AI 替你执行那些已经被定义清楚、可重复的搬砖动作。
流水线模式：把迁移拆成标准工序
"Assembly line" 模式的精髓是分解。把一个庞大的迁移目标，拆成大量同构的小任务：
- 每个任务只做一件事：比如把一个旧框架的 Controller 改写成新框架的 Handler
- 每个任务的输入输出格式固定：旧代码片段 → 新代码片段 + 元数据
- 每个任务的验证规则固定：编译通过、单元测试通过、接口契约不变
这样一来，100 个同类任务可以同时扔给 100 个 LLM 调用并行处理，而不是排队等一个架构师逐个手改。
刚性验证回路：锁死幻觉的出口
LLM 最危险的不是生成慢，而是生成"看起来对但实际错"的代码。Stein 强调，必须用程序化的刚性验证回路来兜底：
- 编译验证：生成的新代码必须能编译，零容忍语法错误
- 测试验证：必须通过已有的单元测试，或至少通过自动生成的对照测试
- 契约验证：接口签名、输入输出类型必须与旧系统严格一致
- 差异验证：关键路径的行为差异必须被自动检测并标记
任何一步失败，该任务直接打回重做，不进入下一环节。这不是"建议性检查"，而是硬性门禁——只有全部通过才算完成。
实战：搭建一条最小迁移流水线
下面用一个简化但可运行的示例，展示如何把上述思路落地。假设我们要把一批旧版 Django View 函数迁移到 FastAPI Handler。核心流程：任务定义 → LLM 生成 → 刚性验证 → 通过/打回。
# migration_pipeline.py — 最小可运行示例
# 前置依赖：pip install openai fastapi uvicorn pytest
# 假设你已有旧 Django 代码片段和对应的单元测试
import os, json, subprocess, textwrap
from openai import OpenAI
from pathlib import Path
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
# ---- 1. 定义标准任务格式 ----
TASK_SCHEMA = {
"old_code": "str", # 旧代码片段
"old_tests": "str", # 旧单元测试代码
"target_framework": "str", # 目标框架名
"interface_contract": "dict", # 接口契约：路径、方法、参数、返回类型
}
# ---- 2. LLM 生成环节 ----
SYSTEM_PROMPT = textwrap.dedent("""\
你是一个代码迁移工程师。将给定的旧框架代码改写为目标框架代码。
规则：
- 保持接口契约完全一致（路径、HTTP方法、参数名、返回类型）
- 只输出代码，不输出解释
- 输出格式：{"new_code": "...", "new_tests": "..."}
""")
def generate(task: dict, retries: int = 3) -> dict:
"""调用 LLM 生成新代码，失败则重试"""
for attempt in range(retries):
resp = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": json.dumps(task, ensure_ascii=False)},
],
temperature=0.1, # 低温度减少随机性
)
try:
result = json.loads(resp.choices[0].message.content)
return result
except json.JSONDecodeError:
print(f"[attempt {attempt+1}] LLM 输出非合法 JSON，重试")
raise RuntimeError("LLM 多次输出格式错误，任务终止")
# ---- 3. 刚性验证回路 ----
def validate_compilation(code: str, filename: str) -> bool:
"""验证生成的代码能否被 Python 编译"""
try:
compile(code, filename, "exec")
return True
except SyntaxError as e:
print(f"编译失败: {e}")
return False
def validate_tests(test_code: str, test_filename: str) -> bool:
"""将测试代码写入临时文件并运行 pytest"""
tmp = Path(f"/tmp/migration_{test_filename}")
tmp.write_text(test_code)
result = subprocess.run(
["pytest", str(tmp), "-x", "--tb=short"],
capture_output=True, text=True, timeout=30,
)
if result.returncode != 0:
print(f"测试失败:\n{result.stdout}\n{result.stderr}")
return False
return True
def validate_contract(new_code: str, contract: dict) -> bool:
"""验证新代码的接口契约与旧系统一致（简化版：正则检查）"""
import re
path_ok = contract["path"] in new_code
method_ok = contract["method"].lower() in new_code
# 更严格的实现可以用 AST 解析，这里用正则做快速检查
return path_ok and method_ok
# ---- 4. 流水线主循环 ----
def run_pipeline(task: dict, max_rounds: int = 5) -> dict:
"""完整流水线：生成 → 验证 → 打回重做 → 通过"""
for round_num in range(max_rounds):
print(f"\n=== 第 {round_num+1} 轮 ===")
result = generate(task)
new_code = result["new_code"]
new_tests = result["new_tests"]
# 刚性验证，全部通过才算完成
checks = [
("编译", validate_compilation(new_code, "handler.py")),
("契约", validate_contract(new_code, task["interface_contract"])),
("测试", validate_tests(new_tests, "test_handler.py")),
]
all_pass = all(passed for _, passed in checks)
for name, passed in checks:
status = "✅" if passed else "❌"
print(f" {name} 验证: {status}")
if all_pass:
print("\n🎉 任务通过所有验证，迁移完成")
return result
else:
print(" 打回重做...")
raise RuntimeError(f"任务在 {max_rounds} 轮内未通过验证")
# ---- 5. 执行示例 ----
if __name__ == "__main__":
# 这是一个示例任务，实际使用时替换为你的旧代码和契约
sample_task = {
"old_code": textwrap.dedent("""\
from django.http import JsonResponse
def get_user(request, user_id):
user = User.objects.get(id=user_id)
return JsonResponse({"id": user.id, "name": user.name})
"""),
"old_tests": textwrap.dedent("""\
import pytest
def test_get_user(client):
resp = client.get('/users/1/')
assert resp.status_code == 200
data = resp.json()
assert data['id'] == 1
"""),
"target_framework": "fastapi",
"interface_contract": {
"path": "/users/{user_id}",
"method": "GET",
"params": ["user_id"],
"return_type": "dict",
},
}
result = run_pipeline(sample_task)
print("\n生成的新代码:\n", result["new_code"])
运行前需要设置 OPENAI_API_KEY 环境变量，并确保 pytest 可用。这个示例做了几件关键的事：
- 任务格式标准化——每个迁移任务都用同一套 schema 描述，LLM 的输入输出都被约束
- 低温度生成——
temperature=0.1大幅减少 LLM 的随机发散 - 三层刚性验证——编译、契约、测试，任何一层失败就打回，不给幻觉留后门
- 自动重试上限——最多 5 轮，防止无限循环
实际生产中，你还需要补充 AST 级别的契约校验、并行调度器（比如用 asyncio.gather 或任务队列）、以及人工审查环节。但核心骨架就是这个：标准化任务 → LLM 执行 → 刚性验证 → 通过/打回。
并行调度：从一条线到一百条线
单条流水线跑通后，真正的加速来自并行。假设你有 300 个同类 Controller 要迁移，不需要排队：
import asyncio
async def run_pipeline_async(task: dict) -> dict:
# 与 run_pipeline 逻辑相同，但用 async OpenAI client
# 这里省略实现，核心是替换同步调用为异步
pass
async def batch_migrate(tasks: list[dict], concurrency: int = 20):
"""并行调度多条流水线，控制并发数"""
sem = asyncio.Semaphore(concurrency)
async def guarded(task):
async with sem:
return await run_pipeline_async(task)
results = await asyncio.gather(*[guarded(t) for t in tasks])
return results
20 个并发意味着 300 个任务理论上可以在 15 批内完成——如果每个任务平均 2 分钟，整批迁移大约 30 分钟，而不是一个架构师苦干数周。
落地前的取舍与检查清单
这套模式不是万能药，有几个边界需要提前想清楚：
- 任务同构性要求高：如果旧系统里混杂着各种风格、各种特殊逻辑，标准化分解本身就很困难。先对旧代码做分类和统计，同构比例低于 60% 时收益会打折
- 验证回路的构建成本：编译和测试验证相对容易，但契约验证（尤其是行为等价性）可能需要你先花时间写对照测试或快照
- LLM 的上下文窗口限制：单个任务的旧代码不能太长，否则 LLM 会丢失关键细节。控制在 2000 行以内比较稳妥
- 人工审查不可省：流水线通过所有刚性验证后，仍然需要人工做最终审查——尤其是涉及业务语义的部分
检查清单：
- 旧代码是否已按类型/框架分组，同构任务占比是否足够高？
- 每类任务是否已有明确的接口契约文档？
- 是否已有可自动运行的单元测试覆盖旧代码核心路径？
- 验证回路是否覆盖编译、测试、契约三层？
- 并行调度的并发数是否根据 LLM API 限速做了调整？
- 是否预留了人工审查环节和回滚机制？
把迁移从"架构师苦战"变成"流水线量产"，核心不是 AI 多聪明，而是你多严格。刚性验证回路越硬，AI 的产出就越可靠，迁移的速度就越快。