夜雨聆风用上下文管理、记忆系统和可观测性把 AI Agent 做到生产可用
本文聚焦一个非常具体的问题：如何把一个能演示的 AI Agent，改造成可以上线、可回放、可控成本、可排障的生产级系统。
很多团队做 Agent 的第一版时，通常只关注“能不能调用工具”“回答是否像样”。但一旦进入生产环境，真正决定成败的不是模型有多强，而是以下几个工程问题：上下文会不会无限增长、工具调用失败后怎么处理、一次请求花了多少 token、出了错能不能复现、Agent 会不会陷入循环调用、知识库检索结果是否被正确截断。
这篇文章不讲空泛概念，直接给出一套今天就能落地的工程方案：包括 Agent 配置、提示词设计、工具接入、错误处理、知识库构建、索引创建、部署命令、可运行代码、排障日志和回滚预案。
一、问题背景：生产环境里的 Agent 为什么经常“看起来聪明，用起来失控”
在测试环境里，一个 Agent 可能表现很好：用户问问题，模型拆解任务，调用搜索工具，再整理答案。可一旦进入真实业务，问题会迅速暴露。
第一个痛点是上下文失控。团队以为 128k 上下文就可以把完整文档、历史对话、工具返回结果全部塞进去。但实际表现往往是：开头和结尾信息被模型利用，中间长段内容被忽略，最后回答看似自信，实际引用了错误政策。上下文窗口不是无限记忆，token 也不是免费资源。
第二个痛点是工具调用风暴。Agent 调用知识库没有找到答案，又继续调用订单系统；订单系统超时后，又重复查询；格式校验失败后又重新尝试。没有步数限制和循环检测时，1 分钟花掉 10 美元并不夸张。
第三个痛点是不可观测。用户反馈“昨天那个回答不对”，研发却只能看到最终输出，看不到 Agent 做过哪些决策、调了哪些工具、每个工具返回了什么、在哪一步开始偏离。没有完整请求/响应日志，没有 replay 能力，就无法复现、无法排查、无法优化。
第四个痛点是自动化过度。不少团队追求 100% 自动处理，但在关键业务流程里，80% 自动化加 20% 人工审核往往更稳。特别是退款、合同、权限变更、对外承诺等场景，必须保留人工介入节点。
所以，生产级 Agent 的核心不是“让模型多想几步”，而是把边界、日志、错误处理、上下文管理、成本控制做成系统能力。
二、场景分析：三个业务场景下的真实问题
场景 1：电商客服主管在大促期间处理售后咨询
谁：电商平台客服主管和一线客服。
在什么情况下：618、双 11 等促销期间，用户大量咨询退货政策、订单状态、优惠券使用规则。
遇到什么问题：客服 Agent 需要同时查询知识库和订单系统。知识库文档很多，订单接口偶尔超时。如果没有工具重试、结果截断和上下文清理，Agent 会把完整政策文档和多次接口返回都塞进上下文，单轮对话很快超过 8k token，延迟从 2 秒变成 10 秒以上，还可能反复调用同一个工具。
场景 2：制造企业运维分析师排查设备异常
谁：制造企业的运维分析师。
在什么情况下：设备报警后，需要查询故障手册、历史工单、传感器最近数据。
遇到什么问题：Agent 可能需要调用多个工具：知识库检索、工单系统、监控系统。如果监控系统超时，Agent 没有降级方案，就会卡死；如果没有 trace，事后无法判断到底是知识库召回错误、工具超时，还是模型输出格式异常。
场景 3：中大型企业内部员工使用政策问答助手
谁：企业员工、HR 共享服务团队、IT 服务台。
在什么情况下：员工询问报销规则、假期政策、权限申请流程。
遇到什么问题：政策文档版本频繁变化，Agent 如果把旧对话长期放在上下文里，可能引用过期内容。没有统一 Agent 网关时，各部门重复造轮子，权限、审计、限流都不统一。最终表现是：有的 Agent 访问了不该访问的数据，有的 Agent 没有限流导致成本暴涨。
三、架构原则：先画边界，再谈智能
生产级 Agent 建议遵循四条原则。
1. 边界优先
先明确 Agent 能做什么、不能做什么。例如：可以查询订单状态，但不能直接退款；可以给出政策解释，但不能替用户提交合同审批；可以总结故障原因，但不能直接关闭告警。边界要写进提示词，也要写进工具权限。
2. 可观测性优先
每个请求都必须有 trace_id。每一步都要记录：输入摘要、选择的工具、工具参数、工具返回摘要、耗时、token 估算、是否命中缓存、是否触发降级。这里记录的是外显决策轨迹，不需要保存模型内部隐含推理文本。
3. 降级机制必备
工具调用失败时，重试 2 次；仍失败则走降级方案。例如订单系统超时后，返回“当前无法查询实时订单，请稍后重试或转人工”，而不是让 Agent 一直循环。模型输出格式异常时，先做 JSON 校验，失败后重新生成；连续失败则终止并转人工。
4. token 是预算，不是无限资源
单轮对话建议控制在 8k token 以内。思考阶段可以使用较小模型，最终面向用户的输出再使用较大模型。工具返回结果必须截断，只保留关键信息；历史上下文定期清理，只保留摘要和必要记忆。
四、方案对比：三种落地方式怎么选
如果团队只有 2 到 5 名工程师，建议从方案 B 开始：用成熟框架或轻量自研运行时，不要一开始就做大平台。先把单轮 token 控制在 8k 以内，把关键流程加人工审核，把日志和 replay 做完整。
如果企业内已经有多个 Agent 同时运行，则应尽快走向方案 C：统一 Agent 网关，统一权限、审计、限流和成本统计，避免每个部门各自实现一套不兼容的工具系统。
五、实操教程一：Agent 配置、工具接入、提示词和错误处理
第 1 步：定义 Agent 的运行配置
先不要写代码，先写配置。配置里必须包含：最大步数、重试次数、token 预算、工具结果最大长度、是否启用缓存、是否需要人工审核。
# 场景：生产环境 Agent 的基础配置，保存为 agent_config.yaml
agent_name: customer_support_agent
max_steps: 6
tool_retry: 2
token_budget_per_turn: 8000
tool_result_max_chars: 1200
enable_cache: true
cache_ttl_seconds: 3600
manual_review_required:
- refund
- contract
- permission_change
models:
planning: small_model
final_answer: large_model
这里的关键数值来自生产经验：最大步数先设为 6，不要放开；工具失败重试 2 次；单轮 token 控制在 8k 以内；工具结果截断到 1200 字符左右。这样可以避免一次请求拖垮成本。
第 2 步：接入工具时必须写清楚输入、输出和失败策略
工具不是简单注册一个函数名。每个工具要定义输入 schema、超时、重试、降级方案和权限。
# 场景：给客服 Agent 增加知识库检索和订单查询工具
tools:
- name: kb_search
description: 查询售后政策、优惠券规则、常见问题
timeout_ms: 1500
retry: 2
degrade: 返回未命中结果，并提示转人工或缩小问题范围
input_schema:
query: string
- name: order_status
description: 查询订单支付、发货、签收状态
timeout_ms: 2000
retry: 2
degrade: 返回实时订单不可用，并创建人工跟进记录
input_schema:
order_id: string
第 3 步：提示词必须包含边界、输出格式和人工审核规则
提示词不要只写“你是一个聪明助手”。生产环境要写清楚三类内容：能做什么、不能做什么、输出必须是什么格式。
# 场景：客服 Agent 的系统提示词模板
你是企业客服任务助手，只能基于已注册工具和检索到的知识回答。
边界：
1. 可以解释退货、换货、优惠券、发票等政策。
2. 可以查询订单状态，但不能直接执行退款。
3. 涉及退款、合同、权限变更，必须输出 manual_review=true。
4. 工具没有返回依据时，不要编造答案，应提示转人工或补充信息。
输出格式必须是 JSON：
{"answer":"面向用户的回答","manual_review":false,"used_tools":["工具名"]}
第 4 步：错误处理必须写成固定流程
建议按下面流程实现：
工具调用失败：等待短暂间隔后重试，最多重试 2 次。 仍然失败：走降级方案，返回可解释错误，不继续无限调用。 模型输出异常：做 JSON 格式校验，失败后重新请求一次。 连续异常：终止任务，写入 trace，转人工。 死循环检测：相同工具和相同参数连续出现 2 次以上，直接终止。 步数超限：超过 max_steps 后停止，并返回“需要人工介入”。
六、实操教程二：知识库创建完整流程
这里以客服售后政策知识库为例，使用 SQLite FTS5 做全文索引。它不是最复杂的方案，但部署简单、成本低，适合小企业和创业团队起步。中大型企业后续可以把这层替换为统一的内部知识服务。
第 1 步：准备原始数据
建议把原始数据分成三类：
正式政策：退货政策、发票政策、优惠券规则。 操作手册：客服处理 SOP、升级流程、人工审核规则。 历史问答：高频问题和标准答复。
目录建议如下：
# 场景：知识库原始数据目录
data/
raw/
return_policy.txt
invoice_policy.txt
coupon_rule.txt
support_sop.txt
第 2 步：清洗数据
清洗规则要固定下来，避免每次入库结果不同：
去掉 HTML 标签、导航栏、页脚、版权信息。 统一空白字符，把连续空行压缩为一个空格。 敏感字段脱敏，例如手机号、身份证号、银行卡号。 补充元数据：文档标题、版本号、更新时间、适用业务线。 过期文档不要直接删除，先标记为 inactive，避免回溯时找不到依据。
第 3 步：切分 chunk
不要把一整篇政策作为一个检索单元。建议每个 chunk 控制在 600 到 900 个中文字符，重叠 80 到 120 个字符。这样既能保留上下文，又不会把太多无关内容塞进 prompt。
第 4 步：建立索引
SQLite FTS5 可以直接建立全文索引。索引字段建议包括 title、content、source、version、updated_at。检索时只返回 top 3 到 top 5，并对内容做二次截断。
第 5 步：验证召回质量
不要建完库就上线。至少准备 30 条高频问题做回归测试，例如：
“商品签收 8 天还能退吗？” “优惠券过期能补发吗？” “发票抬头填错了怎么办？” “订单显示已签收但用户没收到怎么办？”
每条问题记录：是否召回正确文档、是否引用最新版本、回答是否需要人工审核。低于预期时，先调清洗和切分策略，再考虑换更复杂的检索方案。
七、核心代码：可运行的三段完整示例
代码示例 1：创建知识库、清洗数据、建立全文索引
使用场景：当你需要把售后政策、SOP、FAQ 做成可检索记忆库时运行。运行后会得到 kb.db，并打印一次检索结果。
# 场景：创建客服知识库。运行：python build_kb.py。结果：生成 kb.db，并输出检索命中。
import os
import re
import sqlite3
from pathlib import Path
RAW_DIR = Path("data/raw")
DB_PATH = "kb.db"
CHUNK_SIZE = 800
OVERLAP = 100
def ensure_sample_data():
RAW_DIR.mkdir(parents=True, exist_ok=True)
sample = RAW_DIR / "return_policy.txt"
if not sample.exists():
sample.write_text("退货政策 v2026：商品签收后 7 天内，如不影响二次销售，可申请退货。生鲜、定制商品、已拆封特殊商品不支持无理由退货。超过 7 天的订单必须转人工审核。退款动作不能由客服助手直接执行。", encoding="utf-8")
coupon = RAW_DIR / "coupon_rule.txt"
if not coupon.exists():
coupon.write_text("优惠券规则 v2026：优惠券过期后默认不补发。如因系统故障导致无法使用，可由人工客服核验后补偿。涉及补偿金额的流程必须人工审核。", encoding="utf-8")
def clean_text(text):
text = re.sub(r"<[^>]+>", " ", text)
text = re.sub(r"1[3-9]\\d{9}", "手机号已脱敏", text)
text = re.sub(r"\\s+", " ", text).strip()
return text
def split_chunks(text):
chunks = []
start = 0
while start < len(text):
end = start + CHUNK_SIZE
chunks.append(text[start:end])
start = end - OVERLAP
if start < 0:
start = end
return [c for c in chunks if c.strip()]
def build_index():
if os.path.exists(DB_PATH):
os.remove(DB_PATH)
conn = sqlite3.connect(DB_PATH)
conn.execute("CREATE VIRTUAL TABLE kb USING fts5(title, content, source, version, updated_at)")
for path in RAW_DIR.glob("*.txt"):
raw = path.read_text(encoding="utf-8")
text = clean_text(raw)
for i, chunk in enumerate(split_chunks(text)):
title = f"{path.stem}#{i}"
conn.execute("INSERT INTO kb(title, content, source, version, updated_at) VALUES (?, ?, ?, ?, ?)", (title, chunk, str(path), "v2026", "2026-06-04"))
conn.commit()
return conn
def search(conn, query):
rows = conn.execute("SELECT title, substr(content, 1, 160), source FROM kb WHERE kb MATCH ? LIMIT 3", (query,)).fetchall()
return rows
if __name__ == "__main__":
ensure_sample_data()
conn = build_index()
print("知识库已创建：", DB_PATH)
print("检索测试：", search(conn, "退货"))
conn.close()
代码示例 2：Agent 运行时，包含重试、降级、截断和 trace
使用场景：验证 Agent 的完整执行链路。运行后会输出最终答案，并在 traces 目录生成可回放 JSON。这个例子重点解决四件事：工具失败不能无限重试、工具返回不能无限塞进上下文、关键动作必须可追踪、超过预算时必须降级。
# 场景：生产级 Agent 最小运行时。运行：python agent_runtime.py
import json, time, uuid
from pathlib import Path
CONFIG = {
"max_steps": 6,
"tool_retry": 2,
"token_budget": 8000,
"tool_result_max_chars": 1200,
"trace_dir": "traces"
}
SYSTEM_PROMPT = """你是企业客服任务助手。只能调用已注册工具。涉及退款、合同、权限变更必须转人工。工具无依据时不要编造答案。最终输出 JSON。"""
def estimate_tokens(text):
return max(1, len(text) // 4)
def truncate(text, max_chars):
if len(text) <= max_chars:
return text
return text[:max_chars] + "..."
def call_tool(name, args):
if name == "search_policy":
return {"answer": "7天无理由退货，定制商品除外", "source": "policy_v3"}
if name == "create_refund":
return {"error": "REFUND_REQUIRES_HUMAN_APPROVAL"}
return {"error": "UNKNOWN_TOOL"}
def run_agent(task):
Path(CONFIG["trace_dir"]).mkdir(exist_ok=True)
trace = {"trace_id": str(uuid.uuid4()), "task": task, "steps": []}
token_used = estimate_tokens(SYSTEM_PROMPT + task)
for step in range(CONFIG["max_steps"]):
if token_used > CONFIG["token_budget"]:
trace["final"] = {"status": "degraded", "answer": "上下文超预算，已转人工"}
break
result = None
for retry in range(CONFIG["tool_retry"] + 1):
result = call_tool("search_policy", {"query": task})
if "error" not in result:
break
time.sleep(0.2)
safe_result = truncate(json.dumps(result, ensure_ascii=False), CONFIG["tool_result_max_chars"])
trace["steps"].append({"step": step + 1, "tool": "search_policy", "result": safe_result})
token_used += estimate_tokens(safe_result)
if "source" in result:
trace["final"] = {"status": "ok", "answer": result["answer"], "source": result["source"]}
break
if "final" not in trace:
trace["final"] = {"status": "failed", "answer": "工具无有效结果，已降级到人工队列"}
path = Path(CONFIG["trace_dir"]) / (trace["trace_id"] + ".json")
path.write_text(json.dumps(trace, ensure_ascii=False, indent=2), encoding="utf-8")
return trace["final"]
print(run_agent("用户购买耳机第6天申请退款，应该怎么处理？"))
这段代码的重点不是复杂，而是把生产环境最容易出事故的边界写死：最大步数、工具重试次数、token 预算、工具结果截断长度。上线前先把这些参数暴露出来，比后面靠人工救火更便宜。
八、踩坑记录：三类问题最容易把 Agent 做崩
坑 1：工具调用风暴
现象通常是接口 QPS 突然升高，日志里同一个查询被重复调用十几次。根因不是模型“笨”，而是系统没有告诉它什么时候停止、工具失败后该如何降级。解决方式是：每个任务设置最大步数；每个工具设置最大重试次数；同一参数的重复调用走缓存；连续两次相同错误直接转人工或返回明确失败原因。
坑 2：上下文无限膨胀
很多团队把 RAG 检索结果、工具返回、历史对话全部塞给模型，短期看答案更丰富，长期看成本和延迟都会失控。正确做法是把 token 当预算管理：检索结果按相关性排序，只取 Top 3；工具返回先结构化摘要，再进入上下文；历史对话只保留用户目标、已确认事实和未解决问题。
坑 3：没有 trace，事故后无法复盘
Agent 出错时，最怕只看到最终答案，看不到中间过程。最低限度的 trace 应包含：用户输入、提示词版本、工具名称、工具参数、工具返回摘要、最终答案、耗时、token 用量和错误码。
九、落地建议：一套可执行的上线清单
第一步，定义不可做清单。退款审批、合同修改、权限变更等高风险动作默认不让 Agent 直接执行，只能给建议或转人工。
第二步，把工具 schema 写细。每个工具必须说明输入字段、输出字段、错误码、重试策略、超时时间和是否需要人工确认。
第三步，建立黄金评测集。先覆盖最高频、最高风险的 20 个任务。每次改 prompt、工具或知识库，都跑一遍评测。
第四步，设置硬上限。例如最多 6 步、每个工具最多重试 2 次、单次工具结果最多 1200 字符。超过上限就降级，不让模型自己猜。
第五步，保存 trace 并做周复盘。每周抽样真实会话，按检索错误、工具错误、推理错误、边界错误、体验问题分类，持续回灌评测集。
十、总结：生产级 Agent 的关键不是更聪明，而是更可控
AI Agent 上线以后，真正决定体验的不是某一次回答有多惊艳，而是它在复杂场景下是否稳定、可解释、可回滚。工具调用要有限制，上下文要有预算，知识库要能评测，关键动作要可追踪，高风险任务要能降级。
如果只做 demo，可以把所有能力都塞给模型；如果要进生产，就要把 Agent 当成一个有权限、有成本、有失败模式的软件系统来设计。先把边界、监控、评测和回滚做好，再逐步扩大自动化范围，这才是更稳的路线。
