创建日期：2026-09-20 ｜ 最近更新：2026-09-20 全部代码完整贴出（含注释，不做省略）。源码留档在 source/agent-engineering/。前置：Agent 工程化总览｜前两篇讲了「两条阶梯」，这篇讲剩下的控制杆与可观测性。

五个控制杆与可观测性：给 Agent 加约束、算清账

前两篇（结构化输出、弹性容错）解决的是「不出错」。这篇解决两件更日常的事：

给 Agent 加约束——五个「控制杆」（levers），每个都是「别让模型自由发挥」的具体落点；

算清成本——Agent 应用的钱是按 token 烧的，没有计数器，成本就是一笔糊涂账。

一、五个控制杆总览

控制杆	管什么	一句话
输入校验	进来的查询	空/超长/危险关键词直接拒
记忆与上下文	喂给模型的历史	滑动窗口 + 超窗部分压成摘要
工具输入校验	调工具的参数	用 schema 卡住工具入参，不信模型
引导（guidance）	提示词	Few-shot 示例把输出格式固化
兜底（fallback）	失败之后	缓存 → 主模型 → 备用 → 静态，四级

先看它们共用的类型定义（这是「单一数据源」做法的好处：一份 schema 同时给类型和运行时校验）：

import { z } from "zod";

// Lever 1: 输入校验 Schema
export const QueryValidationResult = z.discriminatedUnion("valid", [
  z.object({ valid: z.literal(true), query: z.string() }),
  z.object({ valid: z.literal(false), error: z.string() }),
]);

export type ValidationResult = z.infer<typeof QueryValidationResult>;

// Lever 3: NL2SQL Tool 输入 Schema
export const Nl2SqlInput = z.object({
  user_query: z.string().min(1).max(500),
  selected_tables: z.array(z.string()).min(1).max(10),
  filters: z.record(z.string()).optional(),
});

export type Nl2SqlInputType = z.infer<typeof Nl2SqlInput>;

// Lever 5: Fallback 结果 Schema
export const FallbackResultSchema = z.object({
  source: z.enum(["cache", "primary", "fallback", "static"]),
  content: z.string(),
  latencyMs: z.number(),
});

export type FallbackResult = z.infer<typeof FallbackResultSchema>;

// Observability: Token 统计 Schema
export const TokenUsageSchema = z.object({
  inputTokens: z.number(),
  outputTokens: z.number(),
  totalTokens: z.number(),
  estimatedCostUsd: z.number(),
});

export type TokenUsage = z.infer<typeof TokenUsageSchema>;

注意 QueryValidationResult 用的是 z.discriminatedUnion（判别联合）——这样 valid: true 的分支一定带 query，valid: false 的分支一定带 error。用类型系统表达「成功和失败携带不同字段」，比用可选字段（query?: string）安全得多。

二、Lever 1：输入校验

import { z } from "zod";
import type { ValidationResult } from "../entities.js";

const dangerousPatterns = [
  /\b(drop|delete|truncate|alter)\b/i,
  /--/,
  /\/\*/,
  /;\s*(drop|delete|insert|update)/i,
];

export function validateQuery(raw: string): ValidationResult {
  const trimmed = raw.trim();

  if (!trimmed) {
    return { valid: false, error: "查询不能为空" };
  }

  if (trimmed.length > 500) {
    return { valid: false, error: "查询长度不能超过 500 字符" };
  }

  for (const pattern of dangerousPatterns) {
    if (pattern.test(trimmed)) {
      return { valid: false, error: "查询包含不允许的关键词" };
    }
  }

  return { valid: true, query: trimmed };
}

三层校验，各有各的理由：

校验	挡住什么
非空	无意义调用（省一次 token）
长度上限 500	超长输入烧上下文窗口
危险关键词	NL2SQL 场景下，`drop` / `delete` / `--`（注释截断）/ `/* */` / `;` 拼接

这条控制杆的价值在于「入口拦截」：一个空查询如果放进去，会走完整个链路（还可能重试 + 降级），每一步都在烧钱。在入口用 20 行代码拒掉，是性价比最高的优化。

（生产级当然不止文本匹配——真要防注入，还得靠数据库只读账号 + 语句白名单，正则只是第一道筛子。）

三、Lever 2：记忆与上下文窗口

import type { BaseMessage } from "@langchain/core/messages";
import { HumanMessage, AIMessage } from "@langchain/core/messages";

export interface ConversationTurn {
  role: "user" | "assistant";
  content: string;
}

export interface MemoryStore {
  shortTerm: ConversationTurn[]; // 最近 N 轮精确对话
  summary: string;              // LLM 生成的历史摘要
}

export function createMemoryStore(windowSize = 5): MemoryStore {
  return {
    shortTerm: [],
    summary: "",
  };
}

export function addToMemory(
  store: MemoryStore,
  turn: ConversationTurn,
  maxWindowSize = 5,
): MemoryStore {
  const newShortTerm = [...store.shortTerm, turn];

  // 当短期记忆超过窗口大小时，将最旧的对话"压缩"到摘要
  if (newShortTerm.length > maxWindowSize) {
    const overflow = newShortTerm.shift()!;
    const newSummary = store.summary
      ? `${store.summary}\n${overflow.role}: ${overflow.content}`
      : `${overflow.role}: ${overflow.content}`;
    return { shortTerm: newShortTerm, summary: newSummary };
  }

  return { ...store, shortTerm: newShortTerm };
}

export function buildContext(store: MemoryStore): string[] {
  const context: string[] = [];

  if (store.summary) {
    context.push(`[历史摘要] ${store.summary}`);
  }

  for (const turn of store.shortTerm) {
    context.push(`${turn.role === "user" ? "用户" : "助手"}: ${turn.content}`);
  }

  return context;
}

滑动窗口 + 溢出压缩是上下文管理的标准骨架：

[历史摘要] 用户: 我叫张三 / 助手: 你好张三          ← 溢出部分进摘要（便宜、粗粒度）
用户: 帮我查订单                                    ← 最近 N 轮保持原样（贵、保留细节）
助手: 好的，请问订单号？

两个值得注意的点：

addToMemory 返回新对象而不是原地修改（newShortTerm + 展开），是不可变更新的写法——配合 reducer 式状态管理（如 LangGraph 的 Annotation）才安全；
代码里 await import("@langchain/core/messages") 引入了 HumanMessage / AIMessage 但实际没用到——真正做的时候，要把 buildContext 的输出转成消息数组，而不是拼字符串（见下面的注意点）。

⚠️ 一个值得改进的点：buildContext 返回的是字符串数组，真实调用模型时应该返回结构化消息数组（HumanMessage / AIMessage 实例）。把这层「字符串拼接」当抽象挺好的（方便打印/断言），但别真的把拼好的字符串丢给模型——那会丢掉角色边界，模型更容易串戏。

四、Lever 3：工具输入校验

import pRetry from "p-retry";
import { z } from "zod";
import type { Nl2SqlInputType } from "../entities.js";
import { Nl2SqlInput } from "../entities.js";

export function validateToolInput(input: unknown): Nl2SqlInputType {
  return Nl2SqlInput.parse(input);
}

export async function callWithRetry<T>(
  fn: () => Promise<T>,
  options?: { retries?: number; minTimeout?: number; maxTimeout?: number },
): Promise<T> {
  return pRetry(fn, {
    retries: options?.retries ?? 3,
    minTimeout: options?.minTimeout ?? 2000,
    maxTimeout: options?.maxTimeout ?? 10000,
    factor: 2,
    onFailedAttempt: (error) => {
      // Log for observability; production would use structured logger
      void error;
    },
  });
}

export function isRetryableError(error: Error): boolean {
  const msg = error.message.toLowerCase();
  // 4xx client errors are not retryable
  if (msg.includes("400") || msg.includes("401") || msg.includes("403") || msg.includes("404")) {
    return false;
  }
  return true;
}

核心就一行：Nl2SqlInput.parse(input)。

为什么这行重要：Agent 的工具调用是模型生成参数、你执行的。模型完全可能生成 { user_query: "..." }（漏了 selected_tables）或者 selected_tables: "users"（字符串而非数组）。如果你直接把这些参数喂给 SQL 构造器，就等于把模型的幻觉当成了可信输入。

这条控制杆背后是一个更普遍的原则：模型给的一切都是「不可信输入」——和用户提交的表单是同一种东西。凡是「模型 → 你的代码」的边界，都该有一次 schema 校验。（这和 InkOS 的「模型手里只有工具，没有权限」是同一个思路。）

注意这里用的是 .parse() 而不是 .safeParse()——工具入参校验失败时应该抛错（让上层重试或记录），而不是返回一个 { ok: false } 让调用方自己判断。同步的、必过不可的校验用 parse；跨网络边界的解析用 safeParse。

五、Lever 4：引导（Few-shot 固化输出格式）

import { FewShotPromptTemplate, PromptTemplate } from "@langchain/core/prompts";

const nl2sqlExamples = [
  {
    input: "查一下北京用户的数量",
    output: "SELECT COUNT(*) AS user_count FROM users WHERE city = 'Beijing'",
  },
  {
    input: "每个城市的订单总额",
    output: `SELECT u.city, SUM(o.amount) AS total
FROM orders o JOIN users u ON o.user_id = u.id
GROUP BY u.city`,
  },
  {
    input: "上个月状态为 paid 的订单数",
    output: `SELECT COUNT(*) AS order_count FROM orders
WHERE status = 'paid' AND created_at >= date('now', '-1 month')`,
  },
];

const examplePrompt = new PromptTemplate({
  template: "问：{input}\nSQL：{output}",
  inputVariables: ["input", "output"],
});

export function createSqlFewShotPrompt() {
  return new FewShotPromptTemplate({
    examples: nl2sqlExamples,
    examplePrompt,
    prefix: "你是一个 SQL 专家。根据以下示例，将用户的自然语言查询转换为 SQL。",
    suffix: "问：{query}\nSQL：",
    inputVariables: ["query"],
  });
}

export async function formatWithGuidance(query: string): Promise<string> {
  const prompt = createSqlFewShotPrompt();
  return prompt.format({ query });
}

Few-shot 是「用示例代替规则」。NL2SQL 场景下，与其写一堆「请用标准 SQL、表名要用 users、时间用 date('now')」的规则，不如给三个示例——模型对「格式一致」的学习，示例比指令有效得多。

这份实现里的设计要点：

三段式结构：prefix（角色设定）+ examples（示例）+ suffix（当前问题，格式与示例一致）。关键是 suffix 写成 问：{query}\nSQL：——收紧到「SQL：」结尾，模型自然会接着往下补 SQL，而不是先寒暄一段；
示例覆盖了三种典型形态：单表聚合、JOIN + GROUP BY、时间条件。示例的多样性比数量重要；
FewShotPromptTemplate 只负责格式化，不负责调用——引导是「构造输入」，不是「执行」，这层保持纯粹才好测。

六、Lever 5：四级兜底

import type { FallbackResult } from "../entities.js";

// Simple in-memory LRU cache
const cache = new Map<string, { result: string; expiry: number }>();

export function clearCache(): void {
  cache.clear();
}

function cacheKey(query: string): string {
  return `llm:${query}`;
}

export async function callWithFallback(
  query: string,
  primaryLLM: () => Promise<string>,
  fallbackLLM: () => Promise<string>,
  options?: { cacheTtlMs?: number },
): Promise<FallbackResult> {
  const key = cacheKey(query);
  const cacheTtlMs = options?.cacheTtlMs ?? 300_000;
  const start = Date.now();

  // 第 1 级：缓存
  const cached = cache.get(key);
  if (cached && cached.expiry > Date.now()) {
    return { source: "cache", content: cached.result, latencyMs: Date.now() - start };
  }

  // 第 2 级：主 LLM
  try {
    const result = await primaryLLM();
    cache.set(key, { result, expiry: Date.now() + cacheTtlMs });
    return { source: "primary", content: result, latencyMs: Date.now() - start };
  } catch {
    // 继续到备用
  }

  // 第 3 级：备用 LLM
  try {
    const result = await fallbackLLM();
    return { source: "fallback", content: result, latencyMs: Date.now() - start };
  } catch {
    // 继续到静态兜底
  }

  // 第 4 级：静态兜底
  return {
    source: "static",
    content: "抱歉，系统暂时繁忙，请稍后重试。",
    latencyMs: Date.now() - start,
  };
}

四级兜底：缓存 → 主模型 → 备用模型 → 静态文案。

这个函数的返回值设计比实现更值得学：source 字段是可观测性的抓手——上层可以统计「缓存命中率」「降级占比」。没有这个字段，你就永远不知道兜底链到底在不在工作。

两个诚实的注记：

注释写着 Simple in-memory LRU cache，但用的是普通 Map，没有实现 LRU 淘汰（只按过期时间判断）。单进程、小规模够用，生产要么换 lru-cache，要么交给 Redis；
cacheKey 直接拼 llm:${query}——同义不同字的查询不会命中。想做「语义缓存」得靠向量相似度（参见上一篇的 ModelRouter 与多路检索）。

七、可观测性（一）：Token 计数与成本

import type { TokenUsage } from "../entities.js";

// GPT-4o pricing (per 1M tokens)
const PRICING: Record<string, { input: number; output: number }> = {
  "gpt-4o": { input: 2.5, output: 10.0 },
  "gpt-4o-mini": { input: 0.15, output: 0.6 },
};

export interface TokenCounter {
  add(model: string, inputTokens: number, outputTokens: number): void;
  getTotal(): TokenUsage;
  getByModel(model: string): TokenUsage | null;
  reset(): void;
}

export function createTokenCounter(): TokenCounter {
  const usage = new Map<string, { input: number; output: number }>();

  return {
    add(model: string, inputTokens: number, outputTokens: number) {
      const current = usage.get(model) ?? { input: 0, output: 0 };
      usage.set(model, {
        input: current.input + inputTokens,
        output: current.output + outputTokens,
      });
    },

    getTotal(): TokenUsage {
      let totalInput = 0;
      let totalOutput = 0;
      let totalCost = 0;

      for (const [model, data] of usage) {
        totalInput += data.input;
        totalOutput += data.output;
        const pricing = PRICING[model] ?? PRICING["gpt-4o"];
        totalCost += (data.input * pricing.input + data.output * pricing.output) / 1_000_000;
      }

      return {
        inputTokens: totalInput,
        outputTokens: totalOutput,
        totalTokens: totalInput + totalOutput,
        estimatedCostUsd: Math.round(totalCost * 1_000_000) / 1_000_000,
      };
    },

    getByModel(model: string): TokenUsage | null {
      const data = usage.get(model);
      if (!data) return null;
      const pricing = PRICING[model] ?? PRICING["gpt-4o"];
      const cost = (data.input * pricing.input + data.output * pricing.output) / 1_000_000;
      return {
        inputTokens: data.input,
        outputTokens: data.output,
        totalTokens: data.input + data.output,
        estimatedCostUsd: Math.round(cost * 1_000_000) / 1_000_000,
      };
    },

    reset() {
      usage.clear();
    },
  };
}

这个计数器小到 60 行，但它是「成本可控」的前提。 没有它，你无法回答这些问题：

这个功能的单次成本是多少？
降级到便宜模型到底省了多少？
缓存命中率上去了，账单下来了多少？

两个实现细节：

按模型分别累计（Map<model, usage>），因为不同模型单价差 20 倍（上一篇的成本表：Premium 0.03 / Standard 0.0015）——混在一起算总账就失去意义了；
单价表按每 100 万 token 写，计算时除 1_000_000；输出价格通常是输入的 4 倍（gpt-4o: 2.5 vs 10.0），所以**「让模型少废话」比「少喂上下文」更省钱**——这个结论只有看到单价表才会意识到。

⚠️ 硬编码单价表会过期。生产应把它做成配置或从 provider 拉取，并且在账单对不上时能快速定位是单价变了还是用量涨了。

八、可观测性（二）：Tracing

import { getTracingCallbacks } from "@offerai/shared";

export interface TracingConfig {
  enabled: boolean;
  provider: "langsmith" | "langfuse" | "none";
}

export function getTracingConfig(): TracingConfig {
  if (process.env.LANGCHAIN_TRACING_V2 === "true") {
    return { enabled: true, provider: "langsmith" };
  }
  if (process.env.LANGFUSE_PUBLIC_KEY && process.env.LANGFUSE_SECRET_KEY) {
    return { enabled: true, provider: "langfuse" };
  }
  return { enabled: false, provider: "none" };
}

export function createTracingCallbacks() {
  try {
    return getTracingCallbacks();
  } catch {
    return [];
  }
}

这 24 行做了两件对的事：

用环境变量开关观测，不侵入业务代码——两块观测平台可以并存（LangSmith 看链路、Langfuse 留自己机房）；
createTracingCallbacks 用 try-catch 兜住：观测失败绝不能影响主流程。这是一个很容易被忽略的原则——埋点代码抛错把业务搞挂，是典型的「辅助系统反噬主系统」。

关于 Langfuse 的一个必须知道的坑：它已经进入 legacy → OpenTelemetry 迁移期，而且不 flush 就退出会静默丢光 trace（短生命周期进程尤其容易中招）。这条我单独写了一篇实测：LangChain 可观测性：Langfuse 是什么、你那片段的问题在哪。

九、这五个控制杆的共性

回头看，五个控制杆其实在重复同一个动作：在「模型」与「你的系统」之间插一道确定的边界。

模型 → [输入校验]   → 拒掉不该进来的
模型 ← [记忆窗口]   ← 只喂该记住的
模型 → [工具入参校验] → 不信模型给的参数
模型 ← [Few-shot 引导] ← 用示例约束输出形态
模型 ✗ [四级兜底]    → 失败也要有确定结果

共同点：每一处都不假设模型会「乖」。 这正是 Agent 工程化和「写个 prompt 调 API」的分水岭——不是不信任模型的能力，而是不把系统的正确性押在模型的行为上。

关联

前置：Agent 工程化总览
两条阶梯：结构化输出五层防护链、弹性容错五层链
观测的深入实测：Langfuse 与 LangSmith
schema 校验基础：zod 系列（判别联合、parse vs safeParse）
上下文管理的系统化做法：InkOS 架构拆解（FTS5 检索 + 分层记忆）

自测

输入校验里的三层（非空/长度/危险词）分别在防什么？为什么说它是「性价比最高的优化」？
为什么 buildContext 返回字符串数组是「可接受的抽象」，但不应该把拼好的字符串直接丢给模型？
工具入参校验为什么用 .parse() 而不是 .safeParse()？和跨网络边界的解析有什么不同？
Few-shot 模板为什么把 suffix 写成 问：{query}\nSQL：？这样写有什么作用？
四级兜底里的 source 字段为什么重要？
token 单价表按「每 100 万 token」写，输出价格通常是输入的 4 倍——由此能得出什么省钱结论？
createTracingCallbacks 为什么要 try-catch 兜住？

参考

完整源码（随仓库留档）：source/agent-engineering/（levers/ 五个控制杆、observability/ 两个模块、entities.ts 全部 schema）
运行环境：Node v24.14.1；依赖 zod、p-retry、@langchain/core
说明：本系列的运行输出见总览与各篇；本机未运行单元测试（运行器未安装）

一、五个控制杆总览​

二、Lever 1：输入校验​

三、Lever 2：记忆与上下文窗口​

四、Lever 3：工具输入校验​

五、Lever 4：引导（Few-shot 固化输出格式）​

六、Lever 5：四级兜底​

七、可观测性（一）：Token 计数与成本​

八、可观测性（二）：Tracing​

九、这五个控制杆的共性​

关联​

自测​

参考​