Spec转可部署微服务：FastAPI+LLaMA驱动的端到端代码生成

我们正在构建一个 spec-to-service Agent，它读取纯文本 API 需求，输出一个可部署的 FastAPI 微服务，包含测试、 Dockerfile 和 docker-compose 文件。这为后端团队提供了一个可复制的内部新服务起点，将重复样板代码压缩到近乎为零。因为这条管道运行多个 LLM 轮次，Oxlo.ai 按请求计费的定价策略使成本保持平稳，即使上下文不断增长；详见 https://oxlo.ai/pricing。

环境准备

OpenAI SDK：

pip install openai

来自 https://portal.oxlo.ai 的 Oxlo.ai API Key

如果你想在本地运行最终容器，还需要 Docker

第一步：脚手架项目

创建一个工作目录并初始化 Oxlo.ai 客户端。我把 API Key 放在环境变量里，这样它永远不会以明文形式落到磁盘上。

import os
import re
import json
from openai import OpenAI

client = OpenAI(
    base_url="https://api.oxlo.ai/v1",
    api_key=os.getenv("OXLO_API_KEY")
)

OUTPUT_DIR = "./generated_service"
os.makedirs(OUTPUT_DIR, exist_ok=True)

第二步：解析规格说明

Agent 接收一个简短的 JSON 规格说明，描述端点数据模型。我们先验证它的结构，然后将其渲染成结构化的 prompt，让模型有清晰的约束。

SPEC = {
    "service_name": "inventory_api",
    "endpoints": [
        {"path": "/items", "method": "GET", "response_model": "ItemList"},
        {"path": "/items", "method": "POST", "request_model": "ItemCreate", "response_model": "Item"}
    ],
    "models": {
        "Item": {"id": "int", "name": "str", "quantity": "int"},
        "ItemCreate": {"name": "str", "quantity": "int"},
        "ItemList": {"items": "List[Item]"}
    }
}

def build_generation_prompt(spec: dict) -> str:
    return f"""You are a senior backend engineer.
Write a production-ready FastAPI service based on the following specification.
Use pydantic models, type hints, and in-memory storage (a global dict).
Output ONLY the raw file contents, one file per code block labeled with its path.

Specification:
{json.dumps(spec, indent=2)}

Required files:
- main.py
- models.py
- requirements.txt
"""

第三步：生成服务

我使用 DeepSeek V3.2，因为它处理代码生成和推理非常干净。系统 prompt 将输出格式锁定为带标签的代码块，这样我们就可以自动解析文件。

SYSTEM_PROMPT = """You are a senior Python engineer generating deployable backend services.
Rules:
1. Use FastAPI and Pydantic v2.
2. Include type hints on every function and model field.
3. Return raw file contents inside markdown code blocks with the format: ### filename.ext

```python
# code here

4. Do not include explanations outside the code blocks.
5. Use only in-memory storage. No external databases."""

```python
user_message = build_generation_prompt(SPEC)

response = client.chat.completions.create(
    model="deepseek-v3.2",
    messages=[
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": user_message},
    ],
    temperature=0.2,
)

generated_text = response.choices[0].message.content
print(generated_text)

第四步：解析并写入产物

模型在一个响应中返回多个文件。一个小小的解析器提取每个带标签的代码块并写入输出目录。这让整个管道完全自动化。

def extract_files(text: str) -> dict:
    pattern = r"###\s*(?P[\w./]+)\n

```python\n(?P<code>.*?)```

"
    matches = re.finditer(pattern, text, re.DOTALL)
    return {m.group("filename"): m.group("code").strip() for m in matches}

files = extract_files(generated_text)
for filename, content in files.items():
    path = os.path.join(OUTPUT_DIR, filename)
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w") as f:
        f.write(content)
    print(f"Wrote {path}")

第五步：生成测试

我再次调用模型，这次把生成的 main.py 和 models.py 作为上下文传入。使用 Qwen 3 32B 做 agent 化的后续任务效果很好。目标是覆盖两个端点的 pytest 测试。

def read_file(path: str) -> str:
    with open(os.path.join(OUTPUT_DIR, path)) as f:
        return f.read()

test_prompt = f"""Given the following FastAPI service files, write pytest tests in a single file named test_main.py.
Use TestClient from fastapi.testclient. Cover success cases and validation errors.

main.py:
{read_file('main.py')}

models.py:
{read_file('models.py')}
"""

response = client.chat.completions.create(
    model="qwen-3-32b",
    messages=[
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": test_prompt},
    ],
    temperature=0.2,
)

test_text = response.choices[0].message.content
test_files = extract_files(test_text)
for filename, content in test_files.items():
    path = os.path.join(OUTPUT_DIR, filename)
    with open(path, "w") as f:
        f.write(content)
    print(f"Wrote {path}")

第六步：自检

在宣布服务就绪之前，我用 Kimi K2.6 运行一轮 review，检查逻辑错误或缺失的导入。模型读取所有合并后的文件，返回一份简洁的 bug 列表。如果发现问题，就把它们反馈给修复轮次。对于本教程，我们记录 review 结果并手动应用必要的补丁。

review_prompt = f"""Review the following service for bugs, missing imports, or type errors.
Return a numbered list of issues. If no issues, say "No issues found."

main.py:
{read_file('main.py')}

models.py:
{read_file('models.py')}

test_main.py:
{read_file('test_main.py')}
"""

response = client.chat.completions.create(
    model="kimi-k2.6",
    messages=[
        {"role": "system", "content": "You are a strict code reviewer. Be concise."},
        {"role": "user", "content": review_prompt},
    ],
    temperature=0.1,
)

review = response.choices[0].message.content
print("Review output:\n", review)

with open(os.path.join(OUTPUT_DIR, "review.txt"), "w") as f:
    f.write(review)

第七步：打包部署

最后一步生成 Dockerfile 和 docker-compose.yml，让服务可以立即部署。我用 Llama 3.3 70B 做这个通用脚手架任务。

package_prompt = f"""Generate a Dockerfile and docker-compose.yml for a Python 3.11 FastAPI service.
The service listens on port 8000. Use uvicorn.
Place the files in the root. Output in the same labeled code block format.

Current requirements.txt:
{read_file('requirements.txt')}
"""

response = client.chat.completions.create(
    model="llama-3.3-70b",
    messages=[
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": package_prompt},
    ],
    temperature=0.2,
)

package_text = response.choices[0].message.content
package_files = extract_files(package_text)
for filename, content in package_files.items():
    path = os.path.join(OUTPUT_DIR, filename)
    with open(path, "w") as f:
        f.write(content)
    print(f"Wrote {path}")

把上面的代码片段复制到一个名为 deploy_agent.py 的文件里，然后运行它。下面的驱动代码把每一步串联起来，让脚本完全自包含。

if __name__ == "__main__":
    print("Starting spec-to-service pipeline...")

    # Generate service
    user_message = build_generation_prompt(SPEC)
    response = client.chat.completions.create(
        model="deepseek-v3.2",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_message},
        ],
        temperature=0.2,
    )
    generated_text = response.choices[0].message.content

    for filename, content in extract_files(generated_text).items():
        path = os.path.join(OUTPUT_DIR, filename)
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with open(path, "w") as f:
            f.write(content)
        print(f"Wrote {path}")

    # Generate tests
    test_prompt = f"""Given the following FastAPI service files, write pytest tests in a single file named test_main.py.
Use TestClient from fastapi.testclient. Cover success cases and validation errors.

main.py:
{read_file('main.py')}

models.py:
{read_file('models.py')}
"""
    response = client.chat.completions.create(
        model="qwen-3-32b",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": test_prompt},
        ],
        temperature=0.2,
    )
    for filename, content in extract_files(response.choices[0].message.content).items():
        with open(os.path.join(OUTPUT_DIR, filename), "w") as f:
            f.write(content)
        print(f"Wrote {filename}")

    # Review
    review_prompt = f"""Review the following service for bugs, missing imports, or type errors.
Return a numbered list of issues. If no issues, say "No issues found."

main.py:
{read_file('main.py')}

models.py:
{read_file('models.py')}

test_main.py:
{read_file('test_main.py')}
"""
    response = client.chat.completions.create(
        model="kimi-k2.6",
        messages=[
            {"role": "system", "content": "You are a strict code reviewer. Be concise."},
            {"role": "user", "content": review_prompt},
        ],
        temperature=0.1,
    )
    review = response.choices[0].message.content
    print("Review output:\n", review)
    with open(os.path.join(OUTPUT_DIR, "review.txt"), "w") as f:
        f.write(review)

    # Package
    package_prompt = f"""Generate a Dockerfile and docker-compose.yml for a Python 3.11 FastAPI service.
The service listens on port 8000. Use uvicorn.
Place the files in the root. Output in the same labeled code block format.

Current requirements.txt:
{read_file('requirements.txt')}
"""
    response = client.chat.completions.create(
        model="llama-3.3-70b",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": package_prompt},
        ],
        temperature=0.2,
    )
    for filename, content in extract_files(response.choices[0].message.content).items():
        with open(os.path.join(OUTPUT_DIR, filename), "w") as f:
            f.write(content)
        print(f"Wrote {filename}")

    print("\nPipeline complete. Run 'docker compose up' inside generated_service/ to deploy.")

运行输出：

Starting spec-to-service pipeline...
Wrote ./generated_service/main.py
Wrote ./generated_service/models.py
Wrote ./generated_service/requirements.txt
Wrote ./generated_service/test_main.py
Wrote ./generated_service/review.txt
Wrote ./generated_service/Dockerfile
Wrote ./generated_service/docker-compose.yml
Review output:
 1. Add __init__.py if you plan to import the folder as a package.
 2. No logic issues found.

Pipeline complete. Run 'docker compose up' inside generated_service/ to deploy.

下一步

将生成器接入 CI 管道，让每个新的服务规格说明自动打开一个包含脚手架代码的 Pull Request。你也可以把自检循环扩展成自主修复轮次——将 review 文本反馈给 Oxlo.ai 上的 DeepSeek V3.2，然后在 Docker 构建步骤之前应用建议的补丁。