专栏第三篇:把前两篇的方法论和架构图,落成一个“能启动、能调用、能排错”的工程骨架。


一、问题描述:为什么第三篇必须先做“脚手架”

在前两篇里我们已经确定了方向:

  • 第 1 篇:0→1 阶段先做单 Agent,而不是一开始多 Agent;
  • 第 2 篇:单 Agent 最小可用架构要具备可运行、可观测、可恢复、可演进。

但在真实开发中,很多人会卡在这一步:

  • 懂架构,但不知道第一行代码从哪写;
  • 文件越建越乱,后面重构成本极高;
  • 本地能跑一次,第二天就跑不起来。

所以本篇目标很明确:
30 分钟内搭出一个可运行的 Agent 服务骨架,并完成一次端到端验证。


二、环境说明(可复现前提)

  • 操作系统:Windows / macOS / Linux(本文命令以通用 shell 为主)
  • 语言:Python 3.10+
  • Web 框架:FastAPI(轻量、上手快)
  • 运行方式:本地开发模式

注:这里聚焦“工程骨架”而非模型能力,模型调用先用 Mock 或占位实现。


三、目标产物(30 分钟后你应该得到什么)

完成后你应具备:

  1. 一个清晰的目录结构(API / Agent / Tools / State / Obs 分层);
  2. 一个可启动的服务(/healthz/v1/agent/run);
  3. 一条最小闭环链路(输入 -> 规划 -> 执行 -> 输出);
  4. 基础日志能力(可定位请求);
  5. 可扩展的下一步接口(后续接入真实 LLM 与工具)。

四、目录结构(直接可用)

agent-mvp/
├─ app/
│  ├─ api/
│  │  ├─ routes.py
│  │  ├─ schemas.py
│  │  └─ middleware.py
│  ├─ agent/
│  │  ├─ core.py
│  │  ├─ planner.py
│  │  └─ executor.py
│  ├─ tools/
│  │  ├─ base.py
│  │  └─ registry.py
│  ├─ state/
│  │  └─ store.py
│  ├─ obs/
│  │  └─ logger.py
│  ├─ config/
│  │  └─ settings.py
│  └─ main.py
├─ tests/
│  └─ test_healthz.py
├─ requirements.txt
└─ README.md

五、实现步骤(按顺序执行)

Step 1:初始化项目

mkdir agent-mvp && cd agent-mvp
python -m venv .venv

激活环境:

  • Windows(PowerShell)
.\.venv\Scripts\Activate.ps1
  • macOS/Linux
source .venv/bin/activate

安装依赖:

pip install fastapi uvicorn pydantic
pip freeze > requirements.txt

Step 2:创建 API 协议

app/api/schemas.py

from pydantic import BaseModel
from typing import Optional, Dict, Any

class AgentRunRequest(BaseModel):
    session_id: str
    query: str
    context: Optional[Dict[str, Any]] = {}

class AgentRunResponse(BaseModel):
    ok: bool
    request_id: str
    answer: str
    meta: Dict[str, Any] = {}

Step 3:实现最小 Agent Core

app/agent/planner.py

def make_plan(query: str) -> dict:
    # 这里先做最小实现,后续可替换为 LLM 规划
    return {
        "steps": [
            {"id": "s1", "action": "analyze_query", "input": query},
            {"id": "s2", "action": "compose_answer"}
        ]
    }

app/agent/executor.py

def run_plan(plan: dict) -> dict:
    # 最小可用:模拟执行结果
    return {
        "summary": "已完成最小闭环执行(规划->执行->汇总)",
        "steps_count": len(plan.get("steps", []))
    }

app/agent/core.py

import uuid
from app.agent.planner import make_plan
from app.agent.executor import run_plan


def run_agent(session_id: str, query: str, context: dict | None = None) -> dict:
    request_id = f"req_{uuid.uuid4().hex[:10]}"

    plan = make_plan(query)
    result = run_plan(plan)

    answer = (
        f"会话 {session_id} 已处理。"
        f"{result['summary']}"
        f"步骤数:{result['steps_count']}。"
    )

    return {
        "ok": True,
        "request_id": request_id,
        "answer": answer,
        "meta": {
            "steps": result["steps_count"],
            "mode": "mvp-mock"
        }
    }

Step 4:接入路由

app/api/routes.py

from fastapi import APIRouter
from app.api.schemas import AgentRunRequest, AgentRunResponse
from app.agent.core import run_agent

router = APIRouter()

@router.get("/healthz")
def healthz():
    return {"ok": True, "service": "agent-mvp"}

@router.post("/v1/agent/run", response_model=AgentRunResponse)
def agent_run(req: AgentRunRequest):
    result = run_agent(
        session_id=req.session_id,
        query=req.query,
        context=req.context or {}
    )
    return result

app/main.py

from fastapi import FastAPI
from app.api.routes import router

app = FastAPI(title="Agent MVP", version="0.1.0")
app.include_router(router)

Step 5:启动服务并验证

启动:

uvicorn app.main:app --reload --port 8000

验证健康检查:

curl http://127.0.0.1:8000/healthz

预期:

{"ok":true,"service":"agent-mvp"}

验证 Agent 接口:

curl -X POST http://127.0.0.1:8000/v1/agent/run \
  -H "Content-Type: application/json" \
  -d '{"session_id":"sess_001","query":"帮我总结今天任务","context":{}}'

预期返回(示例):

{
  "ok": true,
  "request_id": "req_a1b2c3d4e5",
  "answer": "会话 sess_001 已处理。已完成最小闭环执行(规划->执行->汇总)步骤数:2。",
  "meta": {
    "steps": 2,
    "mode": "mvp-mock"
  }
}

六、常见报错与排查

  1. 报错:ModuleNotFoundError: No module named 'app'

    • 原因:启动目录不在项目根目录
    • 处理:先 cd agent-mvp 再执行 uvicorn
  2. 报错:Address already in use

    • 原因:8000 端口被占用
    • 处理:改端口 --port 8001
  3. 报错:PowerShell 无法激活 venv

    • 原因:执行策略限制
    • 处理:管理员 PowerShell 临时执行:
    Set-ExecutionPolicy -Scope Process Bypass
    
  4. 接口 422 参数错误

    • 原因:请求体字段与 schema 不匹配
    • 处理:核对 session_id/query/context 字段名与类型

七、本篇小结

本篇核心不是“模型多聪明”,而是先把工程地基打好:

  • 有结构,后续扩展不乱;
  • 能启动,开发节奏不断;
  • 可验证,问题可定位。

你现在已经有了一个可运行的单 Agent 骨架。下一步才是“把它做稳”。


八、下一篇预告

《状态机设计实战:从 INIT 到 FAILED 的可恢复执行流》

会重点讲:

  • 状态流转规则怎么定;
  • 失败重试如何不重复执行;
  • 如何设计幂等键和回滚点。
Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐