项目骨架、领域模型与可替换存储

先把 API 写坏

最容易开始的写法是把状态判断、编号生成、文件读写全塞进路由：

@router.post('/tickets')
def create_ticket(payload: TicketPayload):
    data = json.loads(TICKETS.read_text())
    number = f"TICKET-{len(data) + 1:06d}"
    if payload.priority not in {'low', 'normal', 'high'}:
        raise HTTPException(400, 'bad priority')
    data.append({**payload.model_dump(), 'number': number, 'status': 'open'})
    TICKETS.write_text(json.dumps(data))
    return data[-1]

这段代码能跑，却有四个明确问题：测试必须启动 HTTP；CLI 无法复用；状态迁移没有单一规则；两个进程同时写文件会丢数据。我们先不做“完美存储”，只把业务边界固定下来。

环境：Python 3.12 与 uv

本专栏统一用 Python 3.12（StrEnum 和 X | None 写法都依赖 3.10+，3.12 是写这篇时的稳定选择）和 uv 管理依赖。uv 的价值不在于快，而在于 uv.lock 会锁死整棵依赖树——读者按 tag 检出任意一篇，装出来的环境和文章里跑的一致。

# pyproject.toml
[project]
name = "ai-agent-guide"
requires-python = ">=3.12"
dependencies = [
    "fastapi>=0.115",
    "uvicorn[standard]>=0.32",
    "pydantic-settings>=2.6",
]

[dependency-groups]
dev = ["pytest>=8.3", "pytest-asyncio>=0.24", "httpx>=0.27"]

[tool.pytest.ini_options]
pythonpath = ["src"]

uv sync
uv run pytest -q

uv sync 会创建 .venv 并按 lock 文件安装。后面所有命令都以 uv run 开头，是为了避免落到系统 Python 上——这也是本篇最常见的报错来源。

目录和依赖方向

src/ai_agent_guide/
├── domain/
│   ├── ticket.py       # 纯 Python，不依赖 FastAPI/SQLAlchemy
│   ├── evidence.py
│   └── errors.py
├── application/
│   └── tickets.py      # 用例与 Repository 接口
├── adapters/
│   └── json_ticket_repo.py
├── api/
│   └── tickets.py
├── config.py
└── logging.py
tests/

依赖只能向内：api -> application -> domain，adapters 实现 application 暴露的接口。不要让 domain import pydantic 的 Web 专用类型，领域对象应能在 Python REPL 里独立创建。

Preview

三层嵌套与单向依赖：domain 在最里面，谁都不认识

这张图现在看起来像是多余的仪式感，因为项目里只有四个文件。它真正的意义要到第 5 篇才兑现：那时整个存储层被换成 PostgreSQL，而 domain/ 和 application/ 一行不用改。

领域模型先于框架

# domain/ticket.py
from dataclasses import dataclass
from enum import StrEnum

class TicketStatus(StrEnum):
    NEW = 'new'                 # 刚从对话里建出来
    TRIAGING = 'triaging'       # 客服在查
    NEED_INFO = 'need_info'     # 缺信息，等用户补
    RESOLVED = 'resolved'       # 根因解决
    WORKAROUND = 'workaround'   # 有绕过办法，根因还在
    CLOSED = 'closed'

ALLOWED = {
    TicketStatus.NEW:        {TicketStatus.TRIAGING, TicketStatus.CLOSED},
    TicketStatus.TRIAGING:   {TicketStatus.NEED_INFO, TicketStatus.RESOLVED,
                              TicketStatus.WORKAROUND, TicketStatus.CLOSED},
    TicketStatus.NEED_INFO:  {TicketStatus.TRIAGING, TicketStatus.CLOSED},
    TicketStatus.WORKAROUND: {TicketStatus.TRIAGING, TicketStatus.RESOLVED,
                              TicketStatus.CLOSED},
    TicketStatus.RESOLVED:   {TicketStatus.CLOSED, TicketStatus.TRIAGING},
    TicketStatus.CLOSED:     set(),
}

@dataclass(frozen=True)
class Ticket:
    id: str
    number: str
    title: str
    description: str
    status: TicketStatus = TicketStatus.NEW
    reporter_id: str | None = None
    app_version: str | None = None    # 3.1.4
    os_name: str | None = None        # Windows 10
    error_code: str | None = None

    def transition(self, target: TicketStatus) -> 'Ticket':
        if target not in ALLOWED[self.status]:
            raise ValueError(f'invalid transition: {self.status} -> {target}')
        return Ticket(**{**self.__dict__, 'status': target})

先演示失败：Ticket(..., status=NEW).transition(RESOLVED) 必须抛异常。没人查过的问题不能直接标成已解决——这个约束不是提示词，不能让模型决定。

WORKAROUND 和 RESOLVED 分开是有原因的。群里大量结论其实是「先这么绕一下」而不是「修好了」，把两者混成一个状态，后面就没法回答「这个问题到底解决了没有」。第 10 篇给群聊线程判定闭环状态时会复用这套语义。

后面三个字段——版本、系统、错误码——是第 12 篇多轮收集的目标。Agent 答不出时不能直接建一张只有一句抱怨的工单，它得把这三样问出来。现在先把字段留在这里，允许为空。

证据：任何结论都要能回查

# domain/evidence.py
from dataclasses import dataclass
from typing import Literal

@dataclass(frozen=True)
class Evidence:
    chunk_id: str
    source_type: Literal['document', 'chat']
    locator: str          # 文档：'第四章 故障排查 > 4.3 启动异常'
                          # 群聊：'msg_7e04…'
    occurred_at: str      # 文档更新时间，或线程最后一条消息时间

这个类现在还没有使用者——第 9、10 篇才会产出它，第 12 篇才会展示它。提前放进 domain 是因为它代表一条贯穿全篇的约束：系统给出的每一句结论，都要能指回一段可以被人核对的原文。

occurred_at 单独存一份而不是去查来源，是为了让引用在来源被删除后仍然可读。source_type 决定 locator 怎么解释，也决定回答时怎么措辞——文档和群聊的权威性不一样，第 10 篇会正面处理这件事。

Preview

实线是合法迁移，虚线是会抛异常的那条

异常基类：让错误能被分类处理

上面的 transition 抛的是 ValueError，这在只有一个调用方时够用，但很快就不够——API 层需要知道「这是用户输入错了（400）还是系统坏了（500）」，而第 3 篇模型调用工具失败时，还要把错误结构化地喂回给模型。用内置异常无法区分这些。

所以 domain 层先定义一个基类，所有业务异常从它派生：

# domain/errors.py
class DomainError(Exception):
    """所有业务异常的根。捕获它就能捕获全部预期内的失败。"""
    code = 'domain_error'

class InvalidTransition(DomainError):
    code = 'invalid_transition'

class ValidationFailed(DomainError):
    code = 'validation_failed'

transition 改为抛 InvalidTransition。这样 API 层只需要一个异常处理器：

@app.exception_handler(DomainError)
def handle_domain_error(request, exc: DomainError):
    return JSONResponse(status_code=400, content={'error': exc.code, 'message': str(exc)})

关键是那个 code 字段。它是稳定的机器可读标识，不随文案变化——第 3 篇把工具执行失败返回给模型时，返回的正是这个 code 而不是 Python 堆栈。堆栈对模型没有意义，还会泄露内部路径。

Repository 接口和最笨的实现

# application/tickets.py
from typing import Protocol
from ai_agent_guide.domain.ticket import Ticket

class TicketRepository(Protocol):
    def next_number(self) -> str: ...
    def save(self, ticket: Ticket) -> Ticket: ...
    def get(self, ticket_id: str) -> Ticket | None: ...
    def list(self, reporter_id: str | None = None) -> list[Ticket]: ...

class TicketService:
    def __init__(self, repo: TicketRepository):
        self.repo = repo

    def create(self, title: str, description: str, reporter_id: str) -> Ticket:
        if not title.strip():
            raise ValueError('title is required')
        return self.repo.save(Ticket(
            id=uuid4().hex,
            number=self.repo.next_number(),
            title=title.strip(),
            description=description.strip(),
            reporter_id=reporter_id,
        ))

JSON 实现只负责序列化，不负责状态规则：

class JsonTicketRepository:
    def __init__(self, path: Path):
        self.path = path
        self.path.touch(exist_ok=True)

    def next_number(self) -> str:
        rows = json.loads(self.path.read_text() or '[]')
        return f'TICKET-{len(rows) + 1:06d}'

    def save(self, ticket: Ticket) -> Ticket:
        rows = json.loads(self.path.read_text() or '[]')
        rows.append(asdict(ticket) | {'status': ticket.status.value})
        self.path.write_text(json.dumps(rows, ensure_ascii=False, indent=2))
        return ticket

这个实现故意不宣称“生产可用”。第 5 篇会用同一个接口换成 PostgreSQL；本篇的验收点是业务代码不跟着存储一起改。

配置：现在就给模型留好位置

本篇一行模型代码都没有，但配置的形状现在就要定下来，否则第 2 篇会在十几个文件里散落 os.getenv。

# config.py
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file='.env', extra='ignore')

    data_dir: str = 'data'
    log_level: str = 'INFO'

    # 第 2 篇接入模型时用到，现在只是占位
    llm_base_url: str = 'http://localhost:11434/v1'
    llm_api_key: str = ''
    llm_model: str = 'mock'
    llm_timeout_seconds: float = 30.0

@lru_cache
def get_settings() -> Settings:
    return Settings()

四个 LLM_* 变量对应环境变量 LLM_BASE_URL、LLM_API_KEY、LLM_MODEL、LLM_TIMEOUT_SECONDS（pydantic-settings 自动做大小写和前缀映射）。默认值让整个项目在没有 .env 文件时也能启动，这是第 2～4 篇「不配 Key 也能跑」的前提。

lru_cache 让 Settings 只解析一次。不加这个，每次请求都会重读 .env。

结构化日志：给机器看，不是给人看

print 和默认的 logging 输出的是散文，没法过滤和聚合。从第一天就输出 JSON 行：

# logging.py
import json, logging, sys

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            'level': record.levelname,
            'logger': record.name,
            'message': record.getMessage(),
        }
        # logger.info('...', extra={'ticket_number': 'TICKET-000042'})
        payload.update(getattr(record, 'context', {}))
        if record.exc_info:
            payload['error'] = self.formatException(record.exc_info)
        return json.dumps(payload, ensure_ascii=False)

def setup_logging(level: str = 'INFO') -> None:
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(JsonFormatter())
    logging.basicConfig(level=level, handlers=[handler], force=True)

现在看不出价值，因为日志量很小。到第 15 篇要回答「这次慢在哪一步、花了多少 Token」时，靠的就是这些能被 jq 过滤的字段。届时只需往 context 里加 trace_id，不用重写日志调用。

API 和 CLI 共用同一个用例

# api/tickets.py
@router.post('/tickets', response_model=TicketOut)
def create(payload: TicketIn, service: TicketService = Depends(get_ticket_service)):
    return service.create(payload.title, payload.description, reporter_id='demo-user')

Depends(get_ticket_service) 是 FastAPI 的依赖注入。它的作用不是省几行代码，而是让路由不知道自己在用哪个存储：

def get_ticket_service() -> TicketService:
    settings = get_settings()
    repo = JsonTicketRepository(Path(settings.data_dir) / 'tickets.json')
    return TicketService(repo)

组装发生在这一个函数里。测试时用 app.dependency_overrides[get_ticket_service] = lambda: TicketService(FakeRepository()) 就能整体换掉，不需要真实文件、也不需要 mock 补丁。第 5 篇换 PostgreSQL 时，改的同样只有这个函数。

# cli.py
def main():
    service = TicketService(JsonTicketRepository(Path('data/tickets.json')))
    ticket = service.create(sys.argv[1], sys.argv[2], reporter_id='cli-user')
    print(ticket.number)

如果 CLI 和 API 对同一个输入得到不同状态，说明你把业务写进了适配层。

测试与故障排查

uv run pytest tests/domain/test_ticket.py tests/application/test_tickets.py -q
uv run python -m ai_agent_guide.cli "点图标没反应" "任务管理器里有进程，窗口不出来"
curl -X POST http://localhost:8000/tickets \
  -H 'content-type: application/json' \
  -d '{"title":"点图标没反应","description":"任务管理器里有进程，窗口不出来"}'

常见故障：

ModuleNotFoundError：确认使用 uv run，不要直接调用系统 Python。
JSON 文件为空导致 JSONDecodeError：读取时对空字符串回退为 []，保存时使用临时文件和 os.replace。
领域对象被 FastAPI 序列化失败：在 API 边界转换为响应 DTO，不要让路由返回数据库行对象。

本篇验收标准

状态机非法迁移有单元测试，覆盖 new -> resolved，且断言抛出的是 InvalidTransition 而非 ValueError。
API 和 CLI 都调用 TicketService.create，没有重复编号逻辑。
用内存 FakeRepository 替换 JSON 后，application 测试无需改动——通过 dependency_overrides 注入，不改任何业务代码。
/health 和 /tickets 可运行；全程不需要模型 Key，也不需要 .env 文件。
在 domain/ 目录下搜索 import fastapi|sqlalchemy|pydantic，结果为空。
日志输出是每行一个 JSON 对象，可以直接用 jq 过滤。

下一篇将接入大模型。Provider 会先用 Mock 固定输出，这样流式、取消和 SSE 协议都能在没有外部服务时测试。

#项目骨架、领域模型与可替换存储

#先把 API 写坏

#环境：Python 3.12 与 uv

#目录和依赖方向

#领域模型先于框架

#证据：任何结论都要能回查

#异常基类：让错误能被分类处理

#Repository 接口和最笨的实现

#配置：现在就给模型留好位置

#结构化日志：给机器看，不是给人看

#API 和 CLI 共用同一个用例

#测试与故障排查

#本篇验收标准