当前位置：首页 > news >正文

一个项目带你入门AI应用开发08

news 2026/8/3 12:17:03

第 8 课：工程化——让系统可靠、可观测、可测试

8.1 你的目标

给 Agent 系统加上生产级能力：

错误处理：不同异常返回不同的 HTTP 状态码
日志：每个 Agent 的执行耗时、异常记录
测试：mock LLM，不依赖真实 API Key 也能跑

8.2 错误处理

反例：所有错误返回 500

try:result = agent_graph.invoke(state)
except Exception as e:raise HTTPException(500, f"处理失败: {e}")

这样做的问题：

实际发生的错误	返回的 HTTP 状态码	用户看到的
LLM API 超时	500	"处理失败"
LLM 返回格式不对	500	"处理失败"
用户输入太短	500	"处理失败"
数据库连不上	500	"处理失败"

运维完全不知道哪里出了问题。

正解：分层异常

class LLMTimeoutError(Exception):"""LLM 请求超时"""passclass LLMFormatError(Exception):"""LLM 返回格式不符合预期"""pass

然后在 API 层区分处理：

@app.post("/api/chat")
async def chat(req: ChatRequest):try:result = agent_graph.invoke(state)return resultexcept LLMTimeoutError:# 503 = 服务暂时不可用，客户端可以重试raise HTTPException(503, "AI 服务暂时不可用，请稍后重试")except LLMFormatError:# 502 = 上游服务（LLM）返回异常raise HTTPException(502, "AI 返回格式异常")except ValidationError:# 400 = 客户端请求有问题raise HTTPException(400, "请求参数不合法")except Exception:# 500 = 服务器内部错误logger.exception("未预期的错误")raise HTTPException(500, "系统内部错误")

为什么异常分层很重要？

给不同的人看不同的信息：

用户看到友好的"服务暂时不可用"
前端看到明确的 HTTP 状态码（503 触发自动重试，400 不做重试）
运维从日志看到详细的 stack trace
开发者从"502"知道是 LLM 格式问题，去调整 prompt

重试机制

LLM API 经常因为网络波动或限流而失败。一次失败就抛异常太脆弱：

def call_llm(messages, max_retries=3):for attempt in range(max_retries):try:resp = session.post(url, json=body, timeout=60)return resp.json()["choices"][0]["message"]except requests.exceptions.Timeout:if attempt == max_retries - 1:raise LLMTimeoutError("请求超时")time.sleep(2 ** attempt)  # 1s → 2s → 4s

为什么是指数退避？ 第一次失败可能是网络抖动，第二次可能是瞬时负载高，第三次如果还失败说明真的出了问题。每次重试等待时间加倍，避免对已过载的服务造成更大压力。

8.3 可观测性

节点耗时追踪

import timedef timed_node(name):def decorator(node_func):def wrapper(state):start = time.perf_counter()try:result = node_func(state)elapsed = time.perf_counter() - startlogger.info(f"[{name}] 完成，耗时 {elapsed:.2f}s")return resultexcept Exception:elapsed = time.perf_counter() - startlogger.exception(f"[{name}] 在 {elapsed:.2f}s 后失败")raisereturn wrapperreturn decorator

这样每次请求都会记录：

2026-07-31 10:23:45 [INFO] [router] 完成，耗时 0.32s
2026-07-31 10:23:46 [INFO] [tool] 完成，耗时 1.87s
2026-07-31 10:23:46 [INFO] [summary] 完成，耗时 0.41s

从日志中你能看到什么？

Tool Agent 最慢（1.87s）——因为它调了两次 LLM
如果某天 Tool Agent 突然变成 5s，说明 LLM API 变慢了
如果 Router 经常失败，说明 prompt 可能有问题

8.4 测试

为什么测试 Agent 很困难？

Agent 依赖 LLM，而 LLM 调用需要 API Key、需要网络、需要花钱。

解法：Mock LLM

# conftest.py
def mock_call_llm(messages, **kwargs):"""不调真实 LLM，根据 prompt 关键词返回固定 JSON"""combined = " ".join(m.get("content", "") or "" for m in messages)if "意图分类" in combined:# 根据最后一条 user 消息判断返回什么意图last_user = [m["content"] for m in messages if m["role"] == "user"]text = last_user[-1] if last_user else ""if "投诉" in text:return {"content": '{"intent": "complaint", "reason": "mock"}'}elif "订单" in text or "物流" in text:return {"content": '{"intent": "order_query", "reason": "mock"}'}else:return {"content": '{"intent": "general", "reason": "mock"}'}return {"content": "mock 回复"}

这样测试有什么用？

不需要 API Key、不需要网络、测试秒级完成。测试的是 Agent 的逻辑（"意图为 complaint 时是否生成了工单"），而不是 LLM 的分类能力。

def test_complaint_creates_ticket():result = agent_graph.invoke({"user_message": "我要投诉", ...})assert result["escalation_ticket"] is not Noneassert "ticket_id" in result["escalation_ticket"]

那 LLM 的分类能力怎么测？

这是两个不同的问题：

Agent 的逻辑是否正确 → 用 mock 测试（本课的内容）
LLM 的分类能力是否满足需求 → 用评测集测试（不在本课范围内）

8.5 从第 1 课到第 8 课

第 1 课: 30 行 — 一个能聊天的终端程序
第 2 课: 80 行 — 加上意图分类和路由
第 3 课: 150 行 — 加上 Chroma 向量检索 RAG
第 4 课: 250 行 — 加上 Function Calling
第 5 课: 400 行 — 拆成 LangGraph 多 Agent
第 6 课: 500 行 — 加上会话管理
第 7 课: 600 行 — 加上可插拔数据源
第 8 课: 800 行 — 加上错误处理、日志、测试

每一步增加的代码都对应一个真实遇到的问题。 不是预先设计了一个大架构，而是问题驱动架构演进。

本课知识点

概念	你做了什么	为什么
分层异常	LLMTimeoutError / LLMFormatError	不同问题返回不同 HTTP 状态码
指数退避	失败后等 1s/2s/4s	避免对已过载的服务造成更大压力
Mock 测试	固定 JSON 替换真实 LLM	不消耗 API Key，离线可跑
节点耗时	timed_node 装饰器	性能瓶颈一目了然