1. 项目概述为什么需要一套“完整生产落地”的 Agent 技术栈“一套完整生产落地的 Agent 技术栈”——这八个字不是概念包装而是我在过去18个月里踩过37次线上故障、重构5轮核心调度模块、被3个客户凌晨两点电话叫醒后亲手焊出来的答案。它不讲LLM有多强也不吹Agent多智能只解决一个最朴素的问题当你的业务要靠Agent每天自动处理2000份合同审核、实时响应400路IoT设备告警、在电商大促期间每秒调度86个异步任务时你手里的代码能不能扛住能不能查清问题能不能快速迭代新技能能不能让运维同事不用翻三天日志才定位到是某个工具调用超时了0.3秒我见过太多团队卡在“Demo很炫上线就崩”的死循环里。用LangChain搭个聊天机器人本地跑通一上K8s环境变量没配对OpenAI API Key被注入成字符串而非Secret整个Agent集群静默失败换用自建模型服务发现TypeScript的fetch默认不带keepalive连接池耗尽后请求开始排队延迟从200ms飙到8秒PostgreSQL里存了120万条执行轨迹但没建GIN索引光是查“昨天失败的Tool调用”就要等47秒——这时候没人关心你用了什么推理框架大家只问“现在能修好吗”这套技术栈的核心关键词就是标题里那五个词Agent、TypeScript、Next.js、OpenAI Compatible、PostgreSQL。它们不是随意堆砌的流行语而是经过真实流量验证的组合逻辑TypeScript 是唯一能在千行级Agent编排逻辑中守住类型边界的语言它让toolInputSchema和toolOutputParser之间不会出现运行时类型错位Next.js 不只是前端框架它的App Router天然支持Server Actions Streaming Route Handler三件套让Agent的“思考-调用-反馈”链路能在单次HTTP请求内完成流式渲染避免前端反复轮询OpenAI Compatible 是生产兜底的生命线——当某天你依赖的云服务API限频突降50%你得能一键切到本地部署的Ollama或vLLM服务而不用重写所有createChatCompletion调用PostgreSQL 则是Agent世界的“黑匣子记录仪”它不只存结果更存上下文快照、token消耗明细、工具调用链路、甚至用户中断时的中间状态这是任何内存缓存或NoSQL都无法替代的审计刚需。适合谁参考如果你正面临这些场景中的任意一个已上线Agent但平均MTTR平均修复时间超过45分钟团队里有3人以上在维护同一套Agent逻辑却因类型不一致频繁合并冲突每次加一个新Tool都要改5个地方且无法保证输入输出结构兼容运维反馈“数据库慢”但你连该查哪个表、建什么索引都不知道客户问“这个决策是怎么做出的”你只能回答“模型自己想的”。那么这不是一篇教程而是一份从产线抄回来的作业本。接下来我会把每个模块拆开告诉你为什么选它、怎么焊牢、哪里最容易脱焊以及我贴在工位显示器边上的那张“凌晨三点救火清单”上写的前三条是什么。2. 整体架构设计拒绝“胶水式集成”构建可演进的Agent生命周期闭环2.1 架构分层逻辑从“能跑”到“可控”的四层跃迁很多团队的Agent系统停留在L1“能跑”层用LangChain写个Chain丢进FastAPIcurl一下返回JSON。这就像用胶带把发动机绑在自行车架上——能动但震动大、易散架、没法加油。我们这套技术栈强制划出四层每层解决一类生产痛点L1 执行层Execution Layer专注“指令如何精准落地”。这里不用抽象的Runnable而是定义AgentExecutor实体类它必须实现三个硬接口validateInput()校验用户原始输入是否符合业务规则比如合同编号必须是12位数字、executeWithTimeout()所有Tool调用必须带毫秒级超时且超时后自动触发降级策略如返回缓存结果或兜底文案、emitTrace()向PostgreSQL写入结构化执行日志。关键设计点在于所有Tool调用都封装为ToolInvocation对象包含toolName、inputJson、startTime、endTime、statussuccess/error/timeout、errorStack仅error时填充这个对象是后续所有可观测性的唯一数据源。L2 编排层Orchestration Layer解决“多个Tool如何安全协同”。我们弃用LangChain的RouterChain自研轻量级WorkflowEngine。它不解析自然语言只执行预定义DAG有向无环图。每个节点是一个WorkflowStep含stepId、toolName、inputMappingJSONPath表达式如$.user.contractId、outputBinding将上一步结果绑定到下一步的哪个字段、retryPolicy最大重试次数指数退避基数。为什么不用LLM做动态路由因为生产环境中92%的业务流程是确定性的——合同审核永远是“OCR→条款提取→风险识别→盖章”强行让LLM每次“思考”走哪条路既增加延迟又引入不可控分支。DAG配置存在PostgreSQL的workflow_definitions表里支持热更新改完配置无需重启服务。L3 协同层Collaboration Layer应对“多个Agent如何不打架”。当销售Agent和售后Agent同时处理同一客户时必须避免状态覆盖。我们采用“乐观锁版本号”机制每个Agent实例启动时从PostgreSQL获取agent_instance记录含instance_id、version、last_heartbeat。每次状态更新如update customer_status in_review都带WHERE version ${currentVersion}更新成功则version失败则抛出ConcurrentModificationException由上层决定是重试还是合并。这个version字段不是UUID而是整型便于数据库做高效比较——实测在10万并发下比用SELECT FOR UPDATE性能高3.2倍。L4 治理层Governance Layer保障“系统长期健康运行”。包含三块配额中心用PostgreSQL的pg_advisory_xact_lock()实现分布式锁控制每个租户每分钟调用次数防止单个客户脚本误触导致全站雪崩熔断器基于Hystrix思想但存储在PostgreSQL的circuit_breaker_states表里记录tool_name、failure_count、last_failure_time、stateCLOSED/OPEN/HALF_OPENOPEN状态持续5分钟自动转HALF_OPEN审计网关所有外部API调用OpenAI、支付网关、短信平台必须经此层自动记录request_id、masked_request_body敏感字段如手机号、身份证号打码、response_status、latency_ms这是GDPR合规的底线。提示四层之间严禁跨层调用。L2不能直接读L4的熔断状态必须通过L3的GovernanceService接口。我们用TypeScript的declare module强制约束模块依赖编译时即报错——这比文档约定管用100倍。2.2 关键技术选型背后的“血泪教训”为什么是TypeScript而不是Python去年Q3我们用Python重写了核心调度器上线第三天凌晨因datetime.utcnow()未加时区信息导致所有定时任务漂移8小时。TypeScript的zonedTimeToUtc()配合types/node的严格类型检查让这类错误在npm run build阶段就被拦截。更重要的是Agent的输入输出Schema极其复杂一个保险Agent可能要处理{policy: {id: string, coverage: {type: health|life, amount: number}}, documents: Array{url: string, type: id_card|bank_statement}}TypeScript的联合类型、映射类型、条件类型能精准描述这种嵌套约束而Python的TypedDict在深度嵌套时类型推导会失效。为什么是Next.js而不是Express关键在Streaming支持。Agent的思考过程需要实时反馈给前端“正在分析合同第3页...已识别出违约金条款...正在调用法务知识库...”。Next.js的Route Handler允许res.stream()直接推送SSEServer-Sent Events前端用EventSource监听无需WebSocket握手开销。我们实测1000并发下ExpressSocket.IO的内存占用是Next.jsSSE的2.7倍且SSE天然支持自动重连——当用户地铁进隧道断网出来后浏览器自动续传而WebSocket需手动处理reconnect逻辑。为什么坚持OpenAI Compatible上个月OpenAI突然将gpt-4-turbo的max_tokens从4096下调至2048我们3个核心Agent立即超限失败。但因为我们所有调用都走统一的OpenAIApiClient只需改一行环境变量OPENAI_BASE_URLhttp://localhost:8000/v1切到本地vLLM服务2分钟恢复。这个OpenAIApiClient是TypeScript类构造函数接收baseUrl、apiKey、timeoutMs内部用fetch封装所有方法签名createChatCompletion、createEmbedding完全复刻OpenAI官方SDK连429 Too Many Requests的重试逻辑都一模一样——这意味着你的Agent代码里永远不会出现if (isLocal) { vllmCall() } else { openaiCall() }这种污染逻辑。为什么是PostgreSQL而不是MongoDB当客户要求“查出上周所有被人工干预的Agent决策”MongoDB的聚合管道要写23行且无法利用索引加速。而PostgreSQL一句SELECT * FROM agent_executions WHERE status human_intervention AND created_at 2024-05-01配合created_at上的B-tree索引0.012秒返回结果。更关键的是PostgreSQL的JSONB类型支持Gin索引我们可以对input_json字段建索引CREATE INDEX idx_input_contract_id ON agent_executions USING GIN ((input_json-contractId))这样按合同ID查执行记录速度提升400倍。MongoDB的JSON查询在千万级数据下必然变慢而PostgreSQL的JSONB是二进制存储查询性能接近原生字段。3. 核心模块实现从零搭建可验证的Agent执行引擎3.1 AgentExecutor让每一次执行都“可追溯、可重放、可审计”AgentExecutor是整个技术栈的基石它必须做到三件事输入强校验、执行强隔离、日志强结构化。我们不把它做成一个函数而是一个Class强制约束行为边界。// src/core/agent-executor.ts export class AgentExecutor { private readonly db: Pool; // PostgreSQL连接池 private readonly toolRegistry: Mapstring, Tool; constructor(db: Pool, toolRegistry: Mapstring, Tool) { this.db db; this.toolRegistry toolRegistry; } // 主执行入口返回PromiseExecutionResult async execute( input: Recordstring, unknown, context: ExecutionContext ): PromiseExecutionResult { const executionId crypto.randomUUID(); const startTime Date.now(); // 步骤1输入校验 - 防止脏数据进入执行链 try { await this.validateInput(input); } catch (e) { const errorLog await this.logExecution( executionId, INPUT_VALIDATION_FAILED, startTime, Date.now(), JSON.stringify(e) ); throw new InputValidationError(Validation failed: ${e.message}, errorLog.id); } // 步骤2创建执行上下文快照 - 为重放提供依据 const contextSnapshot { ...context, executionId, startTime, inputHash: createHash(sha256).update(JSON.stringify(input)).digest(hex) }; // 步骤3执行主逻辑此处简化为单Tool调用实际为WorkflowEngine驱动 let result: ExecutionResult; try { const tool this.toolRegistry.get(context.toolName); if (!tool) throw new ToolNotFoundError(context.toolName); // 超时控制Promise.race AbortController const controller new AbortController(); const timeoutId setTimeout(() controller.abort(), context.timeoutMs || 30000); result await Promise.race([ tool.invoke(input, contextSnapshot), new PromiseExecutionResult((_, reject) { controller.signal.addEventListener(abort, () reject(new ToolTimeoutError(${context.toolName} timed out after ${context.timeoutMs}ms)) ); }) ]); clearTimeout(timeoutId); } catch (e) { result { status: error, output: null, error: e instanceof Error ? e.message : String(e), errorStack: e instanceof Error ? e.stack : undefined }; } finally { // 步骤4强制日志落库 - 无论成功失败都记录 await this.logExecution( executionId, result.status, startTime, Date.now(), JSON.stringify(result.output), result.error, result.errorStack ); } return result; } private async validateInput(input: Recordstring, unknown): Promisevoid { // 实际项目中这里会加载JSON Schema并校验 // 例如合同审核Agent要求input必须有contractId且为12位数字 if (!input.contractId || typeof input.contractId ! string || !/^\d{12}$/.test(input.contractId)) { throw new ValidationError(contractId must be a 12-digit string); } } private async logExecution( executionId: string, status: success | error | timeout | INPUT_VALIDATION_FAILED, startTime: number, endTime: number, outputJson?: string, error?: string, errorStack?: string ): PromiseExecutionLog { const { rows } await this.db.queryExecutionLog( INSERT INTO agent_executions ( id, status, start_time, end_time, input_hash, output_json, error, error_stack, latency_ms ) VALUES ($1, $2, to_timestamp($3/1000), to_timestamp($4/1000), $5, $6, $7, $8, $9) RETURNING *, [ executionId, status, startTime, endTime, , // input_hash暂空实际中由validateInput生成 outputJson || null, error || null, errorStack || null, endTime - startTime ] ); return rows[0]; } }这段代码的关键细节远超表面输入校验独立成步不是在Tool里做而是在Executor入口处。因为Tool可能被其他系统复用校验规则属于业务契约必须前置。超时控制双保险Promise.race确保不阻塞主线程AbortController提供标准取消信号clearTimeout防止内存泄漏——我们曾因漏掉clearTimeout导致10万并发时Node.js事件循环堆积数百万未清除定时器。日志强制落库finally块里写日志确保即使process.exit(1)也能留下线索。PostgreSQL的INSERT ... RETURNING *原子性保证日志ID可追溯。错误分类明确InputValidationError、ToolTimeoutError、ToolNotFoundError都是继承自Error的自定义类前端可根据error.name做差异化提示而不是统一封装成“系统错误”。注意agent_executions表的DDL必须包含关键索引否则日志查询会拖垮数据库-- 按状态和时间范围查如查所有失败记录 CREATE INDEX idx_executions_status_time ON agent_executions (status, created_at); -- 按executionId查单次执行详情调试必备 CREATE INDEX idx_executions_id ON agent_executions (id); -- 按输入哈希查重复执行防幂等 CREATE INDEX idx_executions_input_hash ON agent_executions (input_hash) WHERE input_hash IS NOT NULL;3.2 WorkflowEngine用声明式DAG替代LLM动态路由我们放弃让LLM决定下一步做什么转而用PostgreSQL存储DAG定义实现稳定、可测试、可回滚的编排逻辑。-- 表workflow_definitions CREATE TABLE workflow_definitions ( id SERIAL PRIMARY KEY, name VARCHAR(100) NOT NULL UNIQUE, -- 如 contract_review_v2 description TEXT, version INTEGER NOT NULL DEFAULT 1, is_active BOOLEAN NOT NULL DEFAULT true, created_at TIMESTAMPTZ DEFAULT NOW(), updated_at TIMESTAMPTZ DEFAULT NOW() ); -- 表workflow_steps CREATE TABLE workflow_steps ( id SERIAL PRIMARY KEY, workflow_id INTEGER NOT NULL REFERENCES workflow_definitions(id) ON DELETE CASCADE, step_order INTEGER NOT NULL, -- 执行顺序从0开始 tool_name VARCHAR(100) NOT NULL, -- 必须在tool_registry中存在 input_mapping JSONB NOT NULL, -- JSONPath映射如 {documentUrl: $.input.documentUrl} output_binding VARCHAR(200), -- 绑定到下一步的哪个字段如 ocrResult retry_policy JSONB, -- {maxRetries: 2, baseDelayMs: 1000} timeout_ms INTEGER DEFAULT 30000, created_at TIMESTAMPTZ DEFAULT NOW() ); -- 索引加速按workflow_id查步骤 CREATE INDEX idx_workflow_steps_workflow_id ON workflow_steps (workflow_id);WorkflowEngine的执行逻辑是纯函数式// src/core/workflow-engine.ts export class WorkflowEngine { constructor(private readonly db: Pool) {} async execute( workflowName: string, initialInput: Recordstring, unknown ): PromiseWorkflowResult { // 步骤1加载最新激活的workflow定义 const { rows: workflowRows } await this.db.queryWorkflowDefinition( SELECT * FROM workflow_definitions WHERE name $1 AND is_active true ORDER BY version DESC LIMIT 1, [workflowName] ); if (workflowRows.length 0) throw new WorkflowNotFoundError(workflowName); const workflow workflowRows[0]; // 步骤2加载所有steps按step_order排序 const { rows: stepRows } await this.db.queryWorkflowStep( SELECT * FROM workflow_steps WHERE workflow_id $1 ORDER BY step_order, [workflow.id] ); // 步骤3执行DAG简化版实际含循环、条件分支 let context: Recordstring, unknown { ...initialInput }; const executionTrace: ExecutionTrace[] []; for (const step of stepRows) { // 解析input_mapping从context中提取参数 const toolInput this.resolveInputMapping(step.input_mapping, context); // 执行Tool const toolResult await this.executeTool(step.tool_name, toolInput, step.timeout_ms); // 记录trace executionTrace.push({ stepId: step.id, toolName: step.tool_name, input: toolInput, output: toolResult.output, status: toolResult.status, latencyMs: toolResult.latencyMs }); // 将输出绑定到context供下一步使用 if (step.output_binding toolResult.output) { set(context, step.output_binding, toolResult.output); // 使用lodash.set } // 若失败且有重试策略执行重试 if (toolResult.status error step.retry_policy) { const { maxRetries, baseDelayMs } step.retry_policy as RetryPolicy; for (let i 0; i maxRetries; i) { await new Promise(r setTimeout(r, baseDelayMs * Math.pow(2, i))); const retryResult await this.executeTool(step.tool_name, toolInput, step.timeout_ms); if (retryResult.status success) break; } } } return { status: success, output: context, trace: executionTrace }; } private resolveInputMapping(mapping: any, context: any): Recordstring, unknown { // 实现JSONPath解析如 $.input.documentUrl - context.input.documentUrl // 使用 jsonpath-plus 库 } private async executeTool( toolName: string, input: Recordstring, unknown, timeoutMs: number ): Promise{ status: success | error; output: any; latencyMs: number } { // 调用AgentExecutor复用其超时、日志能力 } }这个设计的价值在于可测试性workflow_definitions表可以导出为SQL文件CI流水线中用psql -f test-workflow.sql初始化测试数据然后跑单元测试验证DAG逻辑。可回滚性当新版本workflow上线出问题只需UPDATE workflow_definitions SET is_active false WHERE name contract_review_v2 AND version 2再激活v1版本5秒内恢复。可观测性executionTrace数组完整记录每一步的输入输出前端可渲染为甘特图运维可导出为JSON供ELK分析。实操心得我们曾把input_mapping写成{url: $.document.url}但实际输入结构是{document: {url: xxx}}导致Tool收到undefined。后来强制要求所有mapping必须通过jsonpath-plus的parse()方法预编译并在workflow保存时校验——parse($.document.url)成功才允许入库否则报错“无效JSONPath”。3.3 OpenAIApiClient抹平不同LLM服务的差异让切换成本趋近于零OpenAIApiClient是技术栈的“协议适配器”它让业务代码完全感知不到底层是OpenAI、Anthropic还是本地vLLM。// src/lib/openai-api-client.ts export interface OpenAIApiConfig { baseUrl: string; // e.g., https://api.openai.com/v1 or http://localhost:8000/v1 apiKey: string; timeoutMs?: number; defaultModel?: string; } export class OpenAIApiClient { private readonly config: OpenAIApiConfig; private readonly fetch: typeof globalThis.fetch; constructor(config: OpenAIApiConfig, fetchImpl: typeof globalThis.fetch fetch) { this.config config; this.fetch fetchImpl; } // 完全复刻OpenAI官方SDK的createChatCompletion签名 async createChatCompletion( params: ChatCompletionCreateParams ): PromiseChatCompletion { const url new URL(/chat/completions, this.config.baseUrl); const headers { Content-Type: application/json, Authorization: Bearer ${this.config.apiKey} }; // 兼容vLLMvLLM要求model字段为字符串OpenAI接受字符串或null const body { model: params.model || this.config.defaultModel, messages: params.messages, temperature: params.temperature ?? 0.7, max_tokens: params.max_tokens ?? 1024, stream: params.stream ?? false }; const controller new AbortController(); const timeoutId setTimeout(() controller.abort(), this.config.timeoutMs || 60000); try { const res await this.fetch(url.toString(), { method: POST, headers, body: JSON.stringify(body), signal: controller.signal }); clearTimeout(timeoutId); if (!res.ok) { const errorData await res.json(); throw new OpenAIError( HTTP ${res.status}: ${errorData.error?.message || res.statusText}, res.status, errorData ); } if (params.stream) { return this.handleStreamResponse(res); } else { const data await res.json(); return data as ChatCompletion; } } catch (e) { clearTimeout(timeoutId); if (e instanceof OpenAIError) throw e; throw new OpenAIError(Network error: ${e instanceof Error ? e.message : String(e)}); } } private async handleStreamResponse(res: Response): PromiseChatCompletion { // 处理SSE流组装为完整ChatCompletion对象 // 省略具体实现核心是解析data: {...}事件 } }关键设计点URL构造健壮new URL(/chat/completions, this.config.baseUrl)自动处理baseUrl末尾是否有/避免拼出http://localhost:8000//v1/chat/completions。超时双重保障AbortController控制fetchsetTimeout兜底清理防止controller.abort()被忽略。错误标准化所有错误都包装为OpenAIError含statusCode、errorData业务层可统一处理if (err.statusCode 429) { showRateLimitTip() }。流式响应兼容handleStreamResponse方法将SSE流解析为标准ChatCompletion对象前端无需区分stream/non-stream调用方式。注意TypeScript 7.0已弃用moduleResolution: node10必须升级为node。我们在tsconfig.json中明确指定{ compilerOptions: { moduleResolution: node, resolveJsonModule: true, esModuleInterop: true, skipLibCheck: true } }否则import * as jsonpath from jsonpath-plus会报错因为jsonpath-plus的类型声明依赖新解析规则。4. 生产就绪实践PostgreSQL优化、Next.js流式渲染与TypeScript工程化4.1 PostgreSQL深度调优让千万级Agent日志查询不卡顿Agent系统最重的数据库压力来自日志表agent_executions。当它增长到500万行时SELECT * FROM agent_executions WHERE status error ORDER BY created_at DESC LIMIT 10可能需要8秒。我们通过三层优化将其压到0.015秒第一层分区表Partitioning按月分区避免单表过大-- 创建父表 CREATE TABLE agent_executions ( id UUID PRIMARY KEY, status VARCHAR(20) NOT NULL, start_time TIMESTAMPTZ NOT NULL, end_time TIMESTAMPTZ NOT NULL, input_hash VARCHAR(64), output_json JSONB, error TEXT, error_stack TEXT, latency_ms INTEGER NOT NULL, created_at TIMESTAMPTZ DEFAULT NOW() ) PARTITION BY RANGE (created_at); -- 创建2024年5月分区 CREATE TABLE agent_executions_202405 PARTITION OF agent_executions FOR VALUES FROM (2024-05-01) TO (2024-06-01); -- 创建索引在每个分区上 CREATE INDEX idx_agent_executions_202405_status_time ON agent_executions_202405 (status, created_at);PostgreSQL自动路由查询到对应分区WHERE created_at BETWEEN 2024-05-01 AND 2024-05-31只扫描一个分区。第二层物化视图Materialized View为高频聚合查询建物化视图-- 创建物化视图每日各状态统计 CREATE MATERIALIZED VIEW daily_execution_stats AS SELECT DATE(created_at) as day, status, COUNT(*) as count, AVG(latency_ms) as avg_latency, MAX(latency_ms) as max_latency FROM agent_executions GROUP BY DATE(created_at), status; -- 创建索引 CREATE INDEX idx_daily_stats_day_status ON daily_execution_stats (day, status); -- 刷新命令在CRON中每天0点执行 REFRESH MATERIALIZED VIEW daily_execution_stats;SELECT * FROM daily_execution_stats WHERE day 2024-05-20毫秒级返回。第三层JSONB高级索引针对output_json中的关键字段建GIN索引-- 索引快速查找所有输出了risk_level:high的记录 CREATE INDEX idx_executions_risk_high ON agent_executions USING GIN ((output_json-risk_analysis-risk_level)) WHERE (output_json-risk_analysis-risk_level) high; -- 索引按合同ID查所有相关执行假设output_json包含contractId CREATE INDEX idx_executions_contract_id ON agent_executions USING GIN ((output_json-contractId));GIN索引使JSON字段查询速度媲美普通字段。常见问题pg_stat_statements显示agent_executions的shared_blks_read极高。排查发现是SELECT *查询未加WHERE条件。解决方案在应用层强制所有查询必须带WHERE或在PostgreSQL中创建pg_rules限制全表扫描。我们选择前者在AgentExecutor.logExecution中记录query_context字段监控哪些业务方在扫全表。4.2 Next.js App Router流式渲染让Agent“思考过程”实时可见Next.js的App Router让Agent响应不再是“白屏等待”而是渐进式呈现。关键在Route HandlerStreamingServer Components三者联动。// app/api/agent/route.ts import { OpenAIApiClient } from /lib/openai-api-client; import { WorkflowEngine } from /core/workflow-engine; export async function POST(request: Request) { const { workflowName, input } await request.json(); // 创建OpenAIApiClient实例从env读取 const openaiClient new OpenAIApiClient({ baseUrl: process.env.OPENAI_BASE_URL!, apiKey: process.env.OPENAI_API_KEY! }); // 创建WorkflowEngine注入db连接 const workflowEngine new WorkflowEngine(getDbPool()); // 设置Response为流式 const encoder new TextEncoder(); const stream new ReadableStream({ async start(controller) { try { // 步骤1发送初始状态 controller.enqueue(encoder.encode(event: status\ndata: {message:Agent started,step:0}\n\n)); // 步骤2执行Workflow逐段推送 const result await workflowEngine.execute(workflowName, input); // 推送最终结果 controller.enqueue(encoder.encode(event: result\ndata: ${JSON.stringify(result)}\n\n)); } catch (e) { controller.enqueue(encoder.encode(event: error\ndata: ${JSON.stringify({ message: (e as Error).message })}\n\n)); } finally { controller.close(); } } }); return new Response(stream, { headers: { Content-Type: text/event-stream, Cache-Control: no-cache, Connection: keep-alive } }); }前端用EventSource监听// components/AgentResponse.tsx use client; export default function AgentResponse({ workflowName }: { workflowName: string }) { const [messages, setMessages] useStatestring[]([]); useEffect(() { const eventSource new EventSource(/api/agent?workflow${workflowName}); eventSource.onmessage (e) { setMessages(prev [...prev, e.data]); }; eventSource.addEventListener(status, (e) { const data JSON.parse(e.data); setMessages(prev [...prev, → ${data.message}]); }); eventSource.addEventListener(result, (e) { const result JSON.parse(e.data); setMessages(prev [...prev, ✅ Done: ${JSON.stringify(result.output)}]); }); eventSource.addEventListener(error, (e) { const error JSON.parse(e.data); setMessages(prev [...prev, ❌ Error: ${error.message}]); }); return () eventSource.close(); }, [workflowName]); return ( div classNamespace-y-2 {messages.map((msg, i) ( div key{i} classNametext-sm{msg}/div ))} /div ); }这个模式的优势零WebSocket开销SSE基于HTTPNginx/Apache原生支持无需额外配置反向代理。自动重连浏览器内置重连机制断网后自动恢复EventSource的retry属性可自定义重连间隔。服务端可控ReadableStream.start中可精确控制推送节奏比如每执行完一个Workflow Step就推送一次让前端渲染进度条。实操心得我们曾遇到SSE在iOS Safari上偶发中断。原因是TextEncoder.encode()对中文字符编码异常。解决方案改用new Blob([data], {type: text/plain}).text()虽稍慢但100%兼容。4.3 TypeScript工程化用类型即文档消灭90%的协作摩擦TypeScript不是加个.ts后缀就完事我们用三招让它真正成为团队协作的“活文档”第一招Schema即类型所有Agent输入输出Schema不写YAML直接写TypeScript接口并用zod生成运行时校验// src/schemas/contract-review.schema.ts import { z } from zod; export const ContractReviewInputSchema z.object({ contractId: z.string().regex(/^\d{12}$/, Contract ID must be 12 digits), documentUrl: z.string().url(Document URL must be valid), reviewType: z.enum([full, risk_only]).default(full) }); export type ContractReviewInput z.in