大规模 AI 编排系统的可观测性工程实践

大规模 AI 编排系统的可观测性工程实践 | 前端进阶之旅

Full translation

完整中文译文

可观测性大规模应用：掌握 ADK 回调以优化成本、延迟和可审计性

AI 智能体编排框架获得了大量关注；然而，当部署变得延迟高、成本昂贵时，开发者往往忽略了一个关键能力：ADK 回调钩子。回调钩子的设计模式和最佳实践使开发者能够将逻辑从智能体重构到回调钩子中，从而增加可观测性、降低成本和延迟，并动态修改会话状态。

本文探讨如何在 ADK 智能体的各个阶段创建回调钩子，以演示以下设计模式：

日志记录和监控性能
管理动态状态
修改请求和响应
有条件地跳过步骤

编排器将项目描述路由到 sequentialEvaluationAgent。sequentialEvaluationAgent 包含 project、anti-patterns、decision、recommendation、audit、upload、merger 和 email 这些子智能体。

在 project、anti-patterns、decision、recommendation 和 merger 子智能体中，我实现了回调钩子以演示它们的能力和实用性。

Google ADK 代表 Agent Development Kit（智能体开发工具包）。它是一个开源智能体框架，使开发者能够以便利的方式构建和部署 AI 智能体。

project、anti-patterns、decision、recommendation 和 synthesis 智能体是 LLM 智能体。这些智能体需要 Gemini 来进行推理并生成文本响应。

Audit Trail、Cloud Storage 和 Email 智能体集成外部 API 或资源，触发确定性操作。

ADK 提供了六种回调钩子类型，分别在智能体执行前后、模型执行前后、工具执行前后被调用。

何时以及为何开始使用 ADK 回调钩子

当我使用 ADK web 进行本地测试和调试时，我的首要优先级是正确性。性能和令牌使用在智能体部署到 QA 环境前优先级较低。

当产品经理和 QA 团队注意到明显的速度缓慢和成本高昂时，他们会通知我。随后，我检查智能体以识别性能瓶颈以及可在回调钩子或应用代码中处理的确定性步骤。

回调的优势

在我的子智能体中使用回调钩子提供了多项关键优势：

前置条件

TypeScript 5.9.3 或更高版本
Node.js 24.13.0 或更高版本
Docker（用于本地运行 MailHog）
Google ADK（用于构建自定义智能体）
Vertex AI 中的 Gemini（在 LLM 智能体中调用模型，但自定义智能体不必需）

注意：由于区域可用性原因，我使用 Vertex AI 中的 Gemini 进行身份认证。Gemini 目前在香港不可用；因此我改用 Vertex AI 中的 Gemini。

安装 npm 依赖

npm i --save-exact @google/adk
npm i --save-dev --save-exact @google/adk-devtools
npm i --save-exact nodemailer
npm i --save-dev --save-exact @types/nodemailer rimraf
npm i --save-exact marked
npm i --save-exact zod

我安装了构建 ADK 智能体、将 Markdown 字符串转换为 HTML 以及在本地测试期间向 MailHog 发送电子邮件所需的依赖。

锁定依赖版本确保在企业级应用的开发和生产环境中版本相同。

环境变量

复制 .env.example 到 .env 并填入凭证：

GEMINI_MODEL_NAME="gemini-3.1-flash-lite-preview"
GOOGLE_CLOUD_PROJECT="<Google Cloud Project ID>"
GOOGLE_CLOUD_LOCATION="global" 
GOOGLE_GENAI_USE_VERTEXAI=TRUE

# SMTP Settings (MailHog)
SMTP_HOST="localhost"
SMTP_PORT=1025
SMTP_USER=""
SMTP_PASS=""
SMTP_FROM="no-reply@test.local"
ADMIN_EMAIL="admin@test.local"

SMTP_HOST、SMTP_PORT、SMTP_USER、SMTP_PASS 是设置 MailHog 进行本地邮件测试的必需项。

SMTP_FROM 是发件人邮箱地址，在本地测试中可以是任意字符串。

ADMIN_EMAIL 是接收 EmailAgent 发送邮件的管理员邮箱地址。在我的使用场景中这是一个环境变量，因为它是唯一的收件人。如果另一个场景需要向客户发送邮件，应移除该环境变量。

ADK 回调模式实战

以下是我如何在实践中实现这四种回调设计模式的说明。

日志记录和监控

使用的回调： beforeAgentCallback 和 afterAgentCallback

agentStartCallback 在 start_time 变量中存储当前时间（毫秒级）到会话状态。

import { SingleAgentCallback } from '@google/adk';

export const START_TIME_KEY = 'start_time';

export const agentStartCallback: SingleAgentCallback = (context) => {
  if (!context || !context.state) {
    return undefined;
  }

context.state.set(START_TIME_KEY, Date.now());
  return undefined;
};

agentEndCallback 从会话状态中获取开始时间并计算持续时间。随后，它使用 console.log 输出性能指标。回调返回 undefined，以便任何子智能体始终流向顺序工作流中的下一个。

export const agentEndCallback: SingleAgentCallback = (context) => {
  if (!context || !context.state) {
    return undefined;
  }

const now = Date.now();
  const startTime = context.state.get<number>(START_TIME_KEY) || now;
  console.log(
    `Performance Metrics for Agent "${context.agentName}": Total Elapsed Time: ${(now - startTime) / 1000} seconds.`,
  );
  return undefined;
};

接下来，我在子智能体中调用这两个回调钩子以了解它们的执行耗时。下面的例子展示了我如何记录 project 子智能体的性能指标。

export function createProjectAgent(model: string) {
  const projectAgent = new LlmAgent({
    name: 'ProjectAgent',
    model,
    beforeAgentCallback: agentStartCallback,
    instruction: (context) => {
      ... LLM instruction ....
    },
    afterAgentCallback: agentEndCallback,
    ... 
  });

return projectAgent;
}

重置会话状态

使用的回调： beforeAgentCallback

编排器在智能体生命周期开始前重置会话状态中的变量。

export const AUDIT_TRAIL_KEY = 'auditTrail';
export const RECOMMENDATION_KEY = 'recommendation';
export const CLOUD_STORAGE_KEY = 'cloudStorage';
export const DECISION_KEY = 'decision';
export const PROJECT_KEY = 'project';
export const ANTI_PATTERNS_KEY = 'antiPatterns';
export const MERGED_RESULTS_KEY = 'mergedResults';
export const PROJECT_DESCRIPTION_KEY = 'project_description';
export const VALIDATION_ATTEMPTS_KEY = 'validation_attempts';

const resetNewEvaluationCallback: SingleAgentCallback = (context) => {
  if (!context || !context.state) {
    return undefined;
  }

const state = context.state;

// Clear all previous evaluation data
  state.set(PROJECT_KEY, null);
  state.set(ANTI_PATTERNS_KEY, null);
  state.set(DECISION_KEY, null);
  state.set(RECOMMENDATION_KEY, null);
  state.set(AUDIT_TRAIL_KEY, null);
  state.set(CLOUD_STORAGE_KEY, null);
  state.set(MERGED_RESULTS_KEY, null);
  state.set(VALIDATION_ATTEMPTS_KEY, 0);

console.log(
    `beforeAgentCallback: Agent ${context.agentName} has reset the session state for a new evaluation cycle.`,
  );

return undefined;
};

编排器在 beforeAgentCallback 中将变量设置为 null。在 prepareEvaluationTool 中，编排器仅用新的项目描述替换描述。

const prepareEvaluationTool = new FunctionTool({
  name: 'prepare_evaluation',
  description: 'Stores the new project description to prepare for a fresh evaluation.',
  parameters: z.object({
    description: z.string().describe('The validated project description from the user.'),
  }),
  execute: ({ description }, context) => {
    if (!context || !context.state) {
      return { status: 'ERROR', message: 'No session state found.' };
    }

// Set the new description for the ProjectAgent to find
    context.state.set(PROJECT_DESCRIPTION_KEY, description);

return { status: 'SUCCESS', message: 'Description updated.' };
  },
});

工具逻辑更加高效，令牌使用量也得以降低。

export const rootAgent = new LlmAgent({
  name: 'ProjectEvaluationAgent',
  model: 'gemini-3.1-flash-lite-preview',
  beforeAgentCallback: resetNewEvaluationCallback,
  instruction: `
    ... LLM instruction ....
  `,
  tools: [prepareEvaluationTool],
  subAgents: [sequentialEvaluationAgent],
});

编排器使用 resetNewEvaluationCallback 重置会话变量，使用 prepareEvaluationTool 替换项目描述。最后，sequentialEvaluationAgent 启动智能体工作流以推导决策并生成建议。

动态状态管理

前置条件： 子智能体使用工具调用执行操作

使用的回调： afterToolCallback

该用例旨在递增会话状态中 validation_attempts 的值。在咨询 AI 后，我认为项目子智能体和决策子智能体的 afterToolCallback 阶段是执行递增操作的最佳位置。因此，我定义了一个元函数，用于创建递增验证尝试次数的 afterToolCallback。

export const VALIDATION_ATTEMPTS_KEY = 'validation_attempts';

export const MAX_ITERATIONS = 3;

import { AfterToolCallback } from '@google/adk';
import { VALIDATION_ATTEMPTS_KEY } from '../output-keys.const.js';
import { MAX_ITERATIONS } from '../validation.const.js';

export function createAfterToolCallback(fatalErrorMessage: string, maxAttempts = MAX_ITERATIONS): AfterToolCallback {
  return ({ tool, context, response }) => {
    if (!tool || !context || !context.state) {
      return undefined;
    }

const toolName = tool.name;
    const agentName = context.agentName;
    const state = context.state;

if (!response || typeof response !== 'object' || !('status' in response)) {
      return undefined;
    }

// [1] Dynamic state management
    const attempts = (state.get
<number>(VALIDATION_ATTEMPTS_KEY) || 0) + 1;
    state.set(VALIDATION_ATTEMPTS_KEY, attempts);

// [2] Response modification
    const status = response.status || 'ERROR';
    if (status === 'ERROR' && attempts >= maxAttempts) {
      context.actions.escalate = true;

return {
        status: 'FATAL_ERROR',
        message: fatalErrorMessage,
      };
    }
  };
}

在项目子智能体和决策子智能体中，我都调用 createAfterToolCallback，分别创建各自用于递增验证尝试次数的 afterToolCallback。

const projectAfterToolCallback = createAfterToolCallback(
  `STOP processing immediately. Max validation attempts reached. Return the most accurate data found so far or empty strings if none.`,
);

const decisionAfterToolCallback = createAfterToolCallback(
  `STOP processing immediately and output the final JSON schema with verdict: "None".`,
);

接下来，我在项目子智能体的 afterToolCallback 阶段调用 projectAfterToolCallback。

export function createProjectAgent(model: string) {
  const projectAgent = new LlmAgent({
    name: 'ProjectAgent',
    model,
    beforeAgentCallback: agentStartCallback,
    instruction: (context) => {
      ... LLM instruction ....
    },
    afterToolCallback: projectAfterToolCallback,
    afterAgentCallback: agentEndCallback,
    tools: [validateProjectTool],
    ... 
  });

return projectAgent;
}

然后，我在决策子智能体的 afterToolCallback 阶段调用 decisionAfterToolCallback。

export function createDecisionTreeAgent(model: string) {
  const decisionTreeAgent = new LlmAgent({
    name: 'DecisionTreeAgent',
    model,
    beforeAgentCallback: [resetAttemptsCallback, agentStartCallback],
    instruction: (context) => {
      ... instruction  of the LLM flow ...
    },
    afterToolCallback: decisionAfterToolCallback,
    afterAgentCallback: agentEndCallback,
    tools: [validateDecisionTool],
    ...
  });

return decisionTreeAgent;
}

请求/响应修改

前提条件：子智能体使用工具调用执行操作
使用的回调：AfterToolCallback

当验证尝试次数超过最大迭代次数时，同一个 AfterToolCallback 还会修改响应。

当状态为 ERROR，且 validation_attempts 的值至少等于 maximum_iterations 时，会将 context.actions.escalate 标志设为 true，以跳出循环。此外，状态会被修改为 FATAL_ERROR，并返回自定义的致命错误消息。

条件式跳过步骤

这是一种用于避免不必要的 LLM 执行的重要设计模式。

使用的回调：beforeModelCallback。

在我的子智能体中，我运行这个回调来验证会话数据。当满足特定条件时，回调会立即返回内容，从而跳过后续的 LLM 流程。

如果项目智能体能够将项目描述拆解为任务、问题、约束和目标，它就会立即返回拆解结果。否则，该智能体会提示 Gemini 使用推理完成拆解。

import { SingleBeforeModelCallback } from '@google/adk';

const beforeModelCallback: SingleBeforeModelCallback = ({ context }) => {
  const { project } = getEvaluationContext(context);
  const { isCompleted } = isProjectDetailsFilled(project);

if (isCompleted) {
    return {
      content: {
        role: 'model',
        parts: [
          {
            text: JSON.stringify(project),
          },
        ],
      },
    };
  }

return undefined;
};

随后，将 beforeModelCallback 挂接到项目子智能体的 beforeModelCallback 阶段。

export function createProjectAgent(model: string) {
  const projectAgent = new LlmAgent({
    name: 'ProjectAgent',
    model,
    description:
      'Analyzes the user-provided project description to extract and structure its core components, including the primary task, underlying problem, ultimate goal, and architectural constraints.',
    beforeAgentCallback: agentStartCallback,
    beforeModelCallback,
    instruction: (context) => {
      const { projectDescription } = getEvaluationContext(context);
      if (!projectDescription) {
        return '';
      }

return generateProjectBreakdownPrompt(projectDescription);
    },
    afterToolCallback: projectAfterToolCallback,
    afterAgentCallback: agentEndCallback,
    tools: [validateProjectTool],
    outputSchema: projectSchema,
    outputKey: PROJECT_KEY,
    disallowTransferToParent: true,
    disallowTransferToPeers: true,
  });

return projectAgent;
}

决策智能体会在 beforeModelCallback 中验证 verdict 属性。如果 verdict 不是 None，回调会立即返回有效的决策。如果 verdict 为 None，回调则会检查项目拆解和反模式。当项目拆解和反模式均已提供时，回调返回 undefined，以触发 LLM 流程。否则，决策智能体便没有可用于推导结论的有效输入。在这种边界情况下，回调返回 None。

import { SingleBeforeModelCallback } from '@google/adk';

const beforeModelCallback: SingleBeforeModelCallback = ({ context }) => {
  const { decision } = getEvaluationContext(context);

if (decision && decision.verdict !== 'None') {
    return {
      content: {
        role: 'model',
        parts: [
          {
            text: JSON.stringify(decision),
          },
        ],
      },
    };
  }

const { project, antiPatterns } = getEvaluationContext(context);
  const { isCompleted } = isProjectDetailsFilled(project);

if (isCompleted && antiPatterns) {
    return undefined;
  }

return {
    content: {
      role: 'model',
      parts: [
        {
          text: JSON.stringify({
            verdict: 'None',
            nodes: [],
          }),
        },
      ],
    },
  };
};

随后，将 beforeModelCallback 挂接到项目子智能体的 beforeModelCallback 阶段。

export function createDecisionTreeAgent(model: string) {
  const decisionTreeAgent = new LlmAgent({
    name: 'DecisionTreeAgent',
    model,
    beforeAgentCallback: [resetAttemptsCallback, agentStartCallback],
    beforeModelCallback,
    instruction: (context) => {
      ... instruction  of the LLM flow ...
    },
    afterToolCallback: decisionAfterToolCallback,
    afterAgentCallback: agentEndCallback,
    tools: [validateDecisionTool],
    outputSchema: decisionSchema,
    outputKey: DECISION_KEY,
    disallowTransferToParent: true,
    disallowTransferToPeers: true,
  });

return decisionTreeAgent;
}

建议智能体使用 beforeModelCallback 检查项目拆解、反模式和结论。有两种场景需要 LLM 生成建议。第一种场景是项目拆解有效、反模式有效，并且结论不为 None。第二种场景是项目拆解不完整，并且结论为 None。LLM 会被要求说明项目拆解中缺失的字段，以及该缺失字段对于得出有效决策的重要性。在其他情况下，回调会立即返回静态建议，并跳过后续的 LLM 流程。

function constructRecommendation(recommendation: string) {
  return {
    content: {
      role: 'model',
      parts: [
        {
          text: JSON.stringify({
            text: recommendation,
          }),
        },
      ],
    },
  };
}

const beforeModelCallback: SingleBeforeModelCallback = ({ context }) => {
  const { project, antiPatterns, decision } = getEvaluationContext(context);

const { isCompleted } = isProjectDetailsFilled(project);
  const isDecisionNone = decision && decision.verdict === 'None';

if ((isCompleted && antiPatterns && decision && decision.verdict !== 'None') || (!isCompleted && isDecisionNone)) { return undefined; } else if (isCompleted && isDecisionNone) { return constructRecommendation( '## Recommendation: Manual Review Required\n\nStatus: Abnormal Case Detected\n\nThe provided project is complete and valid, but the decision tree could not reach a conclusive verdict (Result: None).\n\nPossible Reasons:\n- The requirements fall outside of known architectural patterns.\n- There are conflicting constraints and goals that cannot be resolved automatically.\n\nNext Steps:\n- Review and refine the constraints or goals.\n- Escalate for manual architectural review.', ); }

return constructRecommendation( '## Recommendation: Data Required\n\nStatus: Abnormal Case Detected\n\nNo decision is reached.', ); };


与前面的项目智能体和决策智能体类似，`beforeModelCallback` 函数被挂接到建议智能体的 `beforeModelCallback` 阶段。

```typescript
export function createRecommendationAgent(model: string) {
  const recommendationAgent = new LlmAgent({
    name: 'RecommendationAgent',
    model,
    beforeModelCallback,
    beforeAgentCallback: agentStartCallback,
    instruction: (context) => {
      const { project, antiPatterns, decision } = getEvaluationContext(context);
      const { isCompleted, missingFields } = isProjectDetailsFilled(project);

if (project) {
        if (!isCompleted && decision && decision.verdict === 'None') {
          console.log('RecommendationAgent -> generateFailedDecisionPrompt');
          return generateFailedDecisionPrompt(project, missingFields);
        } else if (isCompleted && antiPatterns && decision && decision.verdict !== 'None') {
          console.log('RecommendationAgent -> generateRecommendationPrompt');
          return generateRecommendationPrompt(project, antiPatterns, decision);
        }
      }
      return 'Skipping LLM due to missing data.';
    },
    afterAgentCallback: agentEndCallback,
    outputSchema: recommendationSchema,
    outputKey: RECOMMENDATION_KEY,
    disallowTransferToParent: true,
    disallowTransferToPeers: true,
  });

return recommendationAgent;
}

以上就是该智能体采用的回调设计模式。下一节将介绍如何启动智能体，以便在终端中观察日志消息。

我从 Docker Hub 拉取了最新版本的 MailHog Docker 镜像，并在本地启动它，用于接收测试邮件并在 Web UI 中显示邮件。docker-compose.yml 文件包含相关配置。

services:
  mailhog:
    image: mailhog/mailhog
    container_name: mailhog
    ports:
      - '1025:1025' # SMTP port
      - '8025:8025' # HTTP (Web UI) port
    restart: always
    networks:
      - decision-tree-agent-network

networks:
  decision-tree-agent-network:

SMTP 服务器监听 1025 端口，Web UI 可通过 http://localhost:8025 访问。

docker compose up -d

在 Docker 中启动 MailHog。

在 package.json 中添加脚本，用于构建并启动 ADK Web 界面。

  "scripts": {
    "prebuild": "rimraf dist",
    "build": "npx tsc --project tsconfig.json",
    "web": "npm run build && npx @google/adk-devtools web --host 127.0.0.1 dist/agent.js"
  },

打开终端，输入 npm run web 启动 API 服务器。

打开一个新的浏览器标签页，输入 http://localhost:8000。

将以下文本粘贴到消息框中：

One of my favorite tech influencers just tweeted about a 'breakthrough in solid-state batteries.' Find which public company they might be referring to, check that company’s recent patent filings to see if it’s true, and then check their stock price to see if the market has already 'priced it in'.

确保根智能体开始执行，并在邮件智能体终止时停止。编排器和项目智能体会触发回调钩子，以重置会话状态、记录性能指标、验证会话数据、增加验证尝试次数并修改响应。

同样，反模式智能体和决策智能体都使用 beforeAgentCallback 和 afterAgentCallback 记录性能指标。它们还使用 beforeModelCallback，在调用 LLM 生成响应之前验证会话数据。此外，决策智能体会在 afterToolCallback 中增加验证尝试次数；当验证尝试次数超过或等于最大迭代次数时，将状态修改为 FATAL_ERROR。

同样，建议智能体和合并智能体都使用 beforeAgentCallback 和 afterAgentCallback 记录性能指标。建议智能体还使用 beforeModelCallback，在调用 LLM 生成建议之前验证会话数据。

ADK 支持适用于不同使用场景的生命周期回调钩子。本文介绍了如何在 beforeAgentCallback 和 afterAgentCallback 中记录性能指标。我重构了编排器 beforeAgentCallback 中重置会话状态的逻辑，使工具更加精简且更具成本效益。当验证—重试循环的循环次数超过最大迭代次数时，afterToolCallback 会将问题升级给更高级别的智能体，并将响应状态修改为 FATAL_ERROR。当需要有条件地跳过 LLM 调用时，beforeModelCallback 会立即返回自定义内容。这样，智能体就不会给执行流程增加不必要的时间，也不会消耗 token。

核心要点：遵循智能体各阶段回调的设计模式和最佳实践，以记录性能指标、降低成本和延迟，并修改响应。

TypeScript 版 Google Development Kit

回调的设计模式与最佳实践

使用 ADK、MCP 和 Gemini 构建多智能体系统

决策树智能体代码仓库