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，丢进FastAPI，curl一下返回JSON。这就像用胶带把发动机绑在自行车架上——能动，但震动大、易散架、没法加油。我们这套技术栈强制划出四层，每层解决一类生产痛点：

L1 执行层（Execution Layer）：专注“指令如何精准落地”。这里不用抽象的Runnable，而是定义AgentExecutor实体类，它必须实现三个硬接口：validateInput()（校验用户原始输入是否符合业务规则，比如合同编号必须是12位数字）、executeWithTimeout()（所有Tool调用必须带毫秒级超时，且超时后自动触发降级策略，如返回缓存结果或兜底文案）、emitTrace()（向PostgreSQL写入结构化执行日志）。关键设计点在于：所有Tool调用都封装为ToolInvocation对象，包含toolName、inputJson、startTime、endTime、status（success/error/timeout）、errorStack（仅error时填充），这个对象是后续所有可观测性的唯一数据源。

L2 编排层（Orchestration Layer）：解决“多个Tool如何安全协同”。我们弃用LangChain的RouterChain，自研轻量级WorkflowEngine。它不解析自然语言，只执行预定义DAG（有向无环图）。每个节点是一个WorkflowStep，含stepId、toolName、inputMapping（JSONPath表达式，如$.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、state（CLOSED/OPEN/HALF_OPEN），OPEN状态持续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()直接推送SSE（Server-Sent Events），前端用EventSource监听，无需WebSocket握手开销。我们实测：1000并发下，Express+Socket.IO的内存占用是Next.js+SSE的2.7倍，且SSE天然支持自动重连——当用户地铁进隧道断网，出来后浏览器自动续传，而WebSocket需手动处理reconnect逻辑。

为什么坚持OpenAI Compatible？上个月OpenAI突然将gpt-4-turbo的max_tokens从4096下调至2048，我们3个核心Agent立即超限失败。但因为我们所有调用都走统一的OpenAIApiClient，只需改一行环境变量OPENAI_BASE_URL=http://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: Map<string, Tool>; constructor(db: Pool, toolRegistry: Map<string, Tool>) { this.db = db; this.toolRegistry = toolRegistry; } // 主执行入口，返回Promise<ExecutionResult> async execute( input: Record<string, unknown>, context: ExecutionContext ): Promise<ExecutionResult> { 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 Promise<ExecutionResult>((_, 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: Record<string, unknown>): Promise<void> { // 实际项目中，这里会加载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 ): Promise<ExecutionLog> { const { rows } = await this.db.query<ExecutionLog>( `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: Record<string, unknown> ): Promise<WorkflowResult> { // 步骤1：加载最新激活的workflow定义 const { rows: workflowRows } = await this.db.query<WorkflowDefinition>( `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.query<WorkflowStep>( `SELECT * FROM workflow_steps WHERE workflow_id = $1 ORDER BY step_order`, [workflow.id] ); // 步骤3：执行DAG（简化版，实际含循环、条件分支） let context: Record<string, 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): Record<string, unknown> { // 实现JSONPath解析，如 "$.input.documentUrl" -> context.input.documentUrl // 使用 jsonpath-plus 库 } private async executeTool( toolName: string, input: Record<string, 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 ): Promise<ChatCompletion> { const url = new URL('/chat/completions', this.config.baseUrl); const headers = { 'Content-Type': 'application/json', 'Authorization': `Bearer ${this.config.apiKey}` }; // 兼容vLLM：vLLM要求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): Promise<ChatCompletion> { // 处理SSE流，组装为完整ChatCompletion对象 // 省略具体实现，核心是解析data: {...}事件 } }

关键设计点：

• URL构造健壮：new URL('/chat/completions', this.config.baseUrl)自动处理baseUrl末尾是否有/，避免拼出http://localhost:8000//v1/chat/completions。
• 超时双重保障：AbortController控制fetch，setTimeout兜底清理，防止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 Handler+Streaming+Server 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] = useState<string[]>([]); 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 className="space-y-2"> {messages.map((msg, i) => ( <div key={i} className="text-sm">{msg}</div> ))} </div> ); }

这个模式的优势：

• 零WebSocket开销：SSE基于HTTP，Nginx/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