AI Agent 编排的声明式配置：像写 K8s YAML 一样定义 Agent

AI Agent 编排的声明式配置像写 K8s YAML 一样定义 Agent把 Agent 的行为硬编码在代码里改一个步骤就要重新编译部署——声明式配置让 Agent 逻辑和运行引擎彻底分离。一、场景痛点你的 Agent 系统有三条业务链路用户问答、文档生成、数据查询。每条链路的步骤数量、工具调用顺序、错误处理策略都不一样。你用 Python 写了三个 orchestrator 类每个类 200 行步骤之间的跳转逻辑嵌套在条件判断里。两周后业务要求问答链路在用户情绪低落时插入一个安抚步骤。你改了问答 orchestrator 的代码重新部署其他两条链路不受影响——但你需要重新测试问答链路的所有分支。一个月后文档生成链路也要加类似步骤你又改一次代码又部署一次。核心矛盾Agent 的编排逻辑与执行引擎耦合在一起每次逻辑变更都牵动代码和部署无法做到改配置不改代码。二、底层机制与原理剖析2.1 声明式 vs 命令式编排2.2 配置模型设计Agent 编排配置至少包含五个维度steps步骤列表每个步骤定义工具名、输入映射、超时时间conditions条件跳转步骤间的前置判断error_policy错误策略——重试次数、降级方案、终止条件dependencies步骤依赖关系——哪些步骤必须在哪些步骤之后resources资源约束——token 上限、并发限制、超时阈值类比 K8ssteps 对应 Pod 的容器列表conditions 对应 initContainer 的条件error_policy 对应 Pod 的 restartPolicydependencies 对效 Pod 的依赖顺序。2.3 配置验证与运行时执行声明式配置必须经过两阶段验证静态验证配置加载时YAML 结构是否合法、步骤名是否唯一、条件表达式是否可解析、循环依赖检测动态验证执行时工具是否存在、参数类型是否匹配、超时阈值是否可达三、生产级代码实现3.1 Agent 配置 YAML 定义# agent_configs/qa_pipeline.yaml —— 问答链路的声明式配置 # 格式仿照 K8s YAMLapiVersion、kind、metadata、spec apiVersion: agent.orb/v1 kind: AgentPipeline metadata: name: qa-pipeline description: 用户问答链路意图识别→知识检索→回答生成→情绪安抚 labels: team: support tier: l2 spec: # 全局资源约束 resources: maxTokensPerStep: 4000 totalTokenLimit: 16000 maxConcurrentSteps: 3 globalTimeoutMs: 60000 # 步骤定义顺序执行条件跳转控制分支 steps: - name: intent-parse tool: nlu-parser version: v2.3 input: # 输入映射从上下文中提取字段映射到工具参数 text: {{ user_input }} language: {{ metadata.language | default(zh) }} timeoutMs: 2000 retryPolicy: maxRetries: 1 backoffMs: 500 output: # 输出绑定工具返回值映射到上下文变量 intent: {{ result.intent }} confidence: {{ result.confidence }} emotion: {{ result.emotion | default(neutral) }} - name: knowledge-search tool: vector-search version: v1.5 # 条件前置只有置信度 0.6 才执行检索 condition: {{ steps.intent-parse.confidence 0.6 }} input: query: {{ steps.intent-parse.intent }} topK: 5 timeoutMs: 5000 retryPolicy: maxRetries: 2 backoffMs: 1000 output: documents: {{ result.hits }} - name: emotion-check # 安抚步骤用户情绪低落时插入 tool: emotion-response version: v1.0 condition: {{ steps.intent-parse.emotion negative }} input: emotion: {{ steps.intent-parse.emotion }} intent: {{ steps.intent-parse.intent }} timeoutMs: 3000 output: comfortMessage: {{ result.message }} - name: answer-generate tool: llm-generator version: v3.1 # 没有条件则总是执行依赖前面步骤的输出 input: intent: {{ steps.intent-parse.intent }} documents: {{ steps.knowledge-search.documents | default([]) }} comfortMessage: {{ steps.emotion-check.comfortMessage | default(null) }} timeoutMs: 10000 output: answer: {{ result.text }} # 错误策略全局兜底 errorPolicy: # 全链路超时超过 60 秒直接终止 onGlobalTimeout: abort # 单步超时跳过该步骤用降级方案继续 onStepTimeout: strategy: fallback fallbackTool: fallback-response fallbackInput: intent: {{ steps.intent-parse.intent | default(unknown) }} # 工具调用失败重试后仍失败则降级 onToolFailure: strategy: retry_then_fallback maxRetries: 2 fallbackTool: fallback-response3.2 配置验证器// config-validator.ts —— Agent 配置静态验证 import Ajv, { ValidateFunction } from ajv; import addFormats from ajv-formats; /** 配置验证结果 */ export interface ValidationResult { valid: boolean; errors: string[]; warnings: string[]; } /** Agent 配置的类型定义从 YAML schema 转换 */ interface AgentConfig { apiVersion: string; kind: string; metadata: { name: string; description?: string; labels?: Recordstring, string }; spec: { resources: { maxTokensPerStep: number; totalTokenLimit: number; maxConcurrentSteps: number; globalTimeoutMs: number; }; steps: StepConfig[]; errorPolicy: ErrorPolicyConfig; }; } interface StepConfig { name: string; tool: string; version: string; condition?: string; input: Recordstring, string; timeoutMs: number; retryPolicy?: { maxRetries: number; backoffMs: number }; output?: Recordstring, string; } interface ErrorPolicyConfig { onGlobalTimeout: string; onStepTimeout?: { strategy: string; fallbackTool?: string; fallbackInput?: Recordstring, string }; onToolFailure?: { strategy: string; maxRetries?: number; fallbackTool?: string }; } export class ConfigValidator { private ajv: Ajv; private validateFn: ValidateFunction; constructor() { this.ajv new Ajv({ allErrors: true, strict: true }); addFormats(this.ajv); // JSON Schema 定义 Agent 配置的结构约束 const schema { type: object, required: [apiVersion, kind, metadata, spec], properties: { apiVersion: { type: string, pattern: ^agent\\.orb/v[0-9]$ }, kind: { type: string, enum: [AgentPipeline] }, metadata: { type: object, required: [name], properties: { name: { type: string, minLength: 3, maxLength: 64 }, }, }, spec: { type: object, required: [steps], properties: { resources: { type: object, required: [globalTimeoutMs], properties: { globalTimeoutMs: { type: number, minimum: 1000 }, maxConcurrentSteps: { type: number, minimum: 1, maximum: 10 }, }, }, steps: { type: array, minItems: 1, items: { type: object, required: [name, tool, version, input, timeoutMs], properties: { name: { type: string, minLength: 2 }, tool: { type: string, minLength: 1 }, timeoutMs: { type: number, minimum: 100, maximum: 120000 }, }, }, }, }, }, }, }; this.validateFn this.ajv.compile(schema); } /** 验证配置文件的结构合法性 */ validateStructure(config: AgentConfig): ValidationResult { const errors: string[] []; const warnings: string[] []; // Ajv 结构验证 if (!this.validateFn(config)) { for (const err of this.validateFn.errors ?? []) { errors.push(Schema error at ${err.instancePath}: ${err.message}); } } // 自定义业务规则验证 const stepNames config.spec.steps.map((s) s.name); // 步骤名唯一性检测 const duplicates stepNames.filter((name, idx) stepNames.indexOf(name) ! idx); if (duplicates.length 0) { errors.push(Duplicate step names: ${duplicates.join(, )}); } // 循环依赖检测条件表达式引用了后面的步骤 for (const step of config.spec.steps) { if (step.condition) { // 简化检测条件中引用了不存在的步骤名 for (const otherStep of config.spec.steps) { if (step.condition.includes(otherStep.name) otherStep ! step) { // 检查引用步骤是否在当前步骤之前 const refIdx config.spec.steps.indexOf(otherStep); const curIdx config.spec.steps.indexOf(step); if (refIdx curIdx) { errors.push( Step ${step.name} condition references future step ${otherStep.name} — circular dependency ); } } } } } // 超时总和检测所有步骤超时之和不应超过全局超时 const totalStepTimeout config.spec.steps.reduce((sum, s) sum s.timeoutMs, 0); const globalTimeout config.spec.resources?.globalTimeoutMs ?? 60000; if (totalStepTimeout globalTimeout) { warnings.push( Sum of step timeouts (${totalStepTimeout}ms) exceeds global timeout (${globalTimeout}ms) ); } return { valid: errors.length 0, errors, warnings, }; } }3.3 运行时执行引擎// pipeline-engine.ts —— 声明式配置的运行时执行引擎 // 引擎只负责按配置执行不包含任何业务逻辑——逻辑全在配置里 import { ConfigValidator, ValidationResult } from ./config-validator; /** 工具注册表name → executor function */ type ToolExecutor (params: Recordstring, unknown) Promiseunknown; export class PipelineEngine { private tools: Mapstring, ToolExecutor new Map(); private validator: ConfigValidator; constructor() { this.validator new ConfigValidator(); } /** 注册工具执行器 */ registerTool(name: string, executor: ToolExecutor): void { this.tools.set(name, executor); } /** 加载并验证配置构建执行上下文 */ async execute(config: AgentConfig, userInput: Recordstring, unknown): Promise{ result: unknown; trace: Array{ step: string; status: string; durationMs: number }; } { // 静态验证配置结构合法性 const validation: ValidationResult this.validator.validateStructure(config); if (!validation.valid) { throw new Error(Invalid config: ${validation.errors.join(; )}); } // 动态验证工具是否已注册 for (const step of config.spec.steps) { if (!this.tools.has(step.tool)) { throw new Error(Tool not registered: ${step.tool} (step: ${step.name})); } } // 初始化执行上下文存储步骤输出和全局变量 const context: Recordstring, unknown { user_input: userInput.text, metadata: userInput.metadata ?? {}, steps: {} as Recordstring, Recordstring, unknown, }; const trace: Array{ step: string; status: string; durationMs: number } []; const globalStart Date.now(); // 按配置顺序执行步骤 for (const step of config.spec.steps) { // 条件前置不满足则跳过 if (step.condition) { const shouldExecute this.evaluateCondition(step.condition, context); if (!shouldExecute) { trace.push({ step: step.name, status: skipped, durationMs: 0 }); continue; } } // 输入映射从上下文中提取变量值 const inputParams this.resolveInput(step.input, context); // 执行工具调用带超时和重试 const result await this.executeWithRetry( step.tool, inputParams, step.timeoutMs, step.retryPolicy ); // 输出绑定工具返回值写入上下文 if (step.output result.status success) { context.steps[step.name] this.resolveOutput(step.output, result.data); } trace.push({ step: step.name, status: result.status, durationMs: result.durationMs, }); // 全局超时检测 if (Date.now() - globalStart config.spec.resources?.globalTimeoutMs ?? 60000) { trace.push({ step: __global_timeout__, status: timeout, durationMs: 0 }); break; } } // 返回最终结果和执行追踪 const lastStep config.spec.steps[config.spec.steps.length - 1]; return { result: context.steps[lastStep.name] ?? null, trace, }; } /** 条件表达式求值简化版模板引擎 */ private evaluateCondition(condition: string, context: Recordstring, unknown): boolean { try { // 将 {{ }} 模板替换为上下文中的实际值 let expr condition.replace(/\{\{([^}])\}\}/g, (_, path) { const value this.getPathValue(path.trim(), context); return JSON.stringify(value); }); // 安全求值仅允许比较表达式不允许任意 JS 执行 // 使用 Function 构造器限制作用域 const fn new Function(return expr); return fn() true; } catch { // 条件求值失败默认不跳过保守策略 return true; } } /** 从上下文中按路径取值 */ private getPathValue(path: string, context: Recordstring, unknown): unknown { const parts path.split(.); let current: unknown context; for (const part of parts) { if (current typeof current object) { current (current as Recordstring, unknown)[part]; } else { return undefined; } } return current; } /** 带超时和重试的工具执行 */ private async executeWithRetry( toolName: string, params: Recordstring, unknown, timeoutMs: number, retryPolicy?: { maxRetries: number; backoffMs: number } ): Promise{ status: string; data: unknown; durationMs: number } { const executor this.tools.get(toolName)!; const maxRetries retryPolicy?.maxRetries ?? 0; const backoffMs retryPolicy?.backoffMs ?? 500; for (let attempt 0; attempt maxRetries; attempt) { const start Date.now(); try { const data await Promise.race([ executor(params), new Promisenever((_, reject) setTimeout(() reject(new Error(Timeout)), timeoutMs) ), ]); return { status: success, data, durationMs: Date.now() - start }; } catch (err) { if (attempt maxRetries) { await new Promise((r) setTimeout(r, backoffMs * (attempt 1))); } else { return { status: failure, data: null, durationMs: Date.now() - start, }; } } } return { status: failure, data: null, durationMs: 0 }; } /** 输入映射解析{{ }} 模板替换 */ private resolveInput( input: Recordstring, string, context: Recordstring, unknown ): Recordstring, unknown { const resolved: Recordstring, unknown {}; for (const [key, template] of Object.entries(input)) { resolved[key] this.resolveTemplate(template, context); } return resolved; } /** 模板字符串解析 */ private resolveTemplate(template: string, context: Recordstring, unknown): unknown { if (!template.includes({{)) return template; return template.replace(/\{\{([^}])\}\}/g, (_, path) { const trimmed path.trim(); // 支持 default 过滤器{{ value | default(fallback) }} const parts trimmed.split(|); const valuePath parts[0].trim(); const value this.getPathValue(valuePath, context); if (value ! undefined value ! null) return JSON.stringify(value); // default 过滤器 if (parts.length 1) { const defaultExpr parts[1].trim(); const defaultMatch defaultExpr.match(/default\(([^]*)\)/); if (defaultMatch) return defaultMatch[1]; } return null; }); } }四、边界分析与架构权衡4.1 条件表达式安全性模板引擎用new Function()执行条件表达式理论上可以注入恶意代码。虽然模板变量来自上下文而非用户输入但用户输入最终会写入上下文。对策条件表达式只允许比较运算,,,!禁止函数调用和赋值。生产环境建议用专门的模板引擎如 jsonpath-plus避免new Function()。4.2 配置热更新的原子性改配置后立即生效但如果新配置有错误比如引用了不存在的工具正在执行的链路会崩溃。你需要保证配置更新时已启动的链路用旧配置执行完毕新链路用新配置。对策配置版本化管理——每个配置有版本号执行引擎启动链路时锁定当前版本新版本只对后续启动的链路生效。4.3 适用边界与禁用场景适用多条业务链路共享同一执行引擎、频繁调整步骤顺序和条件、需要可视化展示 Agent 拓扑禁用单条简单链路配置文件比代码还复杂、条件表达式需要复杂计算模板引擎表达力不足、对执行性能要求极高模板解析增加延迟4.4 与 DAG 编排引擎的对比Argo Workflows、Temporal 等 DAG 编排引擎也是声明式配置但它们侧重于任务调度并发、依赖、重试而 Agent 编排侧重于实时决策条件跳转、降级、上下文传递。两者可以结合Agent 编排配置定义决策逻辑DAG 引擎处理调度和重试。五、总结声明式配置把 Agent 编排逻辑从代码中分离出来用 YAML 定义步骤、条件、错误策略运行引擎只负责按配置执行。核心收益热更新、可视化、逻辑与引擎解耦。代价配置验证的复杂度、条件表达式的安全性、热更新的原子性保证。配置版本化是解决原子性的关键——旧链路用旧配置新链路用新配置不交叉。模板引擎的安全限制是重中之重——只允许比较运算禁止函数调用。