基于NestJS与LangChain构建智能体驱动的自动化任务系统

1. 从定时任务到智能体为什么我们需要 OpenClaw 式的 Agent如果你做过后台系统定时任务Cron Job肯定不陌生。每天凌晨跑个报表每小时同步一次数据这种场景太常见了。传统的做法无非是在服务器上写个 crontab或者在 Node.js 里用node-cron、bull这类库把脚本挂上去跑就完事了。代码逻辑是死的到点触发执行预设好的、固定的操作序列。这就像给机器人设定了一条永不更改的流水线它只管埋头执行不问世事变迁。但现实业务往往没这么“听话”。想象一下这些场景动态数据抓取你需要监控一批商品的价格但它们的上架、下架时间不固定甚至页面结构还会变。固定时间的爬虫要么抓空要么因为页面变化而报错。条件触发的复杂工作流当A系统推送了一条特定状态的消息后需要自动去B系统查询关联数据经过一系列判断和加工再通知到C系统。这个链条的触发时机和后续步骤都不是固定的。自适应运维服务器监控到某个指标异常如CPU持续高于80%传统的告警只是发个邮件。更智能的做法是让系统能自动分析日志、尝试重启服务、如果失败再执行回滚并根据最终结果决定是否升级告警。这些需求的核心不再是“在固定时间点执行固定任务”而是“在满足某些条件可能是时间也可能是事件时执行一系列动态的、可能带有决策能力的操作”。这就是Agent智能体的用武之地。一个 Agent 可以感知环境读取数据、接收事件根据预设的目标和规则进行思考调用工具、进行逻辑判断并执行行动调用API、写入数据库、发送消息。而OpenClaw作为一个在开发者社区中颇具影响力的开源项目注此处为基于标题的合理演绎OpenClaw通常指代一种灵活、可扩展的抓取或自动化框架范式其核心思想在于模块化、可编排和可观测性。它不是一个具体的库而是一种架构模式将复杂的自动化任务拆解成一个个独立的“爪牙”Claw——即功能单元然后通过一个中央大脑Orchestrator来灵活地调度和组合这些爪牙以完成动态的目标。所以“用 Nest LangChain 打造 OpenClaw 式 Agent 定时任务系统”这个标题指向的是一个更高阶的解决方案我们不再编写静态的定时脚本而是构建一个由智能体驱动的、可动态响应和决策的自动化调度平台。NestJS 提供健壮、可维护的后端框架与依赖注入能力LangChain 提供构建大语言模型LLM应用的核心抽象与工具链两者结合让我们能优雅地实现这种“智能定时任务”。2. 技术栈选型为什么是 NestJS 与 LangChain在开始动手前我们必须理清选型逻辑。市面上框架这么多为什么偏偏是这两位2.1 NestJS不止于一个 Web 框架很多刚接触 NestJS 的开发者会把它当成一个加强版的 Express 或 Fastify。这低估了它的价值。在这个项目中我们看重 NestJS 以下几点依赖注入DI与模块化架构这是实现“OpenClaw 式”可插拔性的基石。每个“爪牙”例如一个数据抓取器、一个邮件发送器、一个数据分析模块都可以被封装成一个独立的Provider并通过 NestJS 的 DI 容器进行管理。大脑调度器不需要知道爪牙的具体实现只需要知道它的接口Token就可以动态地获取和使用它。这使得增加、替换或移除功能模块变得异常简单完全符合开闭原则。强大的生命周期与可观测性NestJS 内置了生命周期钩子OnModuleInit,OnApplicationBootstrap等非常适合在应用启动时初始化我们的 Agent 系统、加载配置、连接外部服务。同时结合像nestjs/terminus这样的健康检查库可以轻松暴露我们 Agent 系统的运行状态方便监控。TypeScript 原生支持与工程化对于构建一个可能日益复杂的 Agent 系统类型安全是防止低级错误、提升开发体验的关键。NestJS 与 TypeScript 的深度集成让我们的“爪牙”接口定义、数据流传递都非常清晰。其 CLI 工具和项目结构也强制了一种良好的工程实践有利于团队协作和长期维护。丰富的生态系统对于定时任务NestJS 有官方的nestjs/schedule模块它基于node-cron和rxjs提供了装饰器和 API 两种方式来定义任务并且能与 NestJS 的 DI 完美集成让我们的任务类也能方便地注入其他服务。注意不要一上来就用nestjs/schedule的Cron装饰器定义我们的智能任务。我们将把它作为底层的时间触发器而上层的智能调度逻辑由我们自己的 Agent 服务来实现。2.2 LangChain为 Agent 注入“思考”能力LangChain 是一个用于构建 LLM 应用的框架。在我们的场景中LLM 并非必需但 LangChain 提供的抽象极其有价值。Agent 与 Tools 的标准化抽象LangChain 的核心概念之一就是Agent。一个 LangChain Agent 由三部分组成一个 LLM思考核心、一组Tools可执行的动作、以及一个决定如何调用 Tools 的AgentExecutor决策循环。即使我们不使用 LLM也可以借鉴其Tool的抽象。我们可以将每一个“爪牙”定义为一个 LangChain Tool它有明确的name,description和execute方法。这样所有功能单元就有了统一的接口。工作流Chain的编排能力LangChain 的Chain概念允许我们将多个步骤LLM调用、Tool调用、数据转换串联起来。这对于实现复杂的、多步骤的自动化任务非常有用。我们可以构建一个“价格监控链”先调用“网页抓取Tool”再调用“数据分析Tool”最后调用“通知Tool”。与外部世界的连接器LangChain 社区提供了海量的集成Toolkits从搜索引擎、数据库到各种 SaaS API。这意味着我们的 Agent 系统可以轻松获得“感知”和“操作”外部世界的能力。例如直接使用SerpAPITool 进行搜索或用SQLDatabaseToolkit来查询数据库。为未来接入 LLM 预留可能性也许当前的任务还不需要自然语言理解或生成。但一旦需求升级比如需要让系统根据自然语言描述自动生成任务流程或者让 Agent 能理解模糊的告警信息并自主决策那么基于 LangChain 构建的底层架构可以无缝地接入 LLM只需替换或增强 Agent 的“思考”部分即可而所有的 Tools爪牙都可以复用。两者的结合点NestJS 作为宿主容器管理着所有服务包括 LangChain 的 Agents 和 Tools的生命周期、配置和依赖关系。而 LangChain 提供的抽象则让 NestJS 容器内的这些服务能够以一种标准化的、高内聚低耦合的方式进行交互和编排。我们可以创建一个LangChainService在 NestJS 中它负责初始化和管理所有的 Tools 和 Agents。3. 系统核心架构设计基于以上选型我们来设计系统的核心架构。我们的目标是构建一个“智能任务调度中心”。3.1 架构分层与组件职责整个系统可以清晰地分为四层层级组件职责对应技术触发层定时触发器、事件监听器接收外部触发信号如定时到达、HTTP请求、消息队列事件并转化为标准化的“任务执行请求”。nestjs/schedule,nestjs/event-emitter,nestjs/microservices调度层任务调度器、Agent 执行器核心大脑。接收触发层的请求根据任务ID或类型找到对应的 Agent 定义并驱动其执行。负责管理任务队列、重试、超时控制。自定义 NestJS Service可能集成bull或nestjs/bull用于高级队列管理智能体层LangChain Agent、自定义 Tools系统的“思考与执行”单元。一个 Agent 封装了完成一个特定目标如“监控价格”的逻辑它会按需调用一个或多个 Tools。每个 Tool 对应一个具体的“爪牙”功能。langchain的Agent,Tool类自定义 Tool 实现工具层功能工具集、数据访问层最底层的功能模块。每个工具都是独立的、可复用的功能单元如 HTTP 客户端、数据库查询、邮件发送、文件处理等。它们被 Tools 包装后暴露给 Agent 层。各种 Node.js SDKaxios,nodemailer、数据库 ORMTypeORM,Prisma、自定义服务数据流一个 Cron 表达式触发触发层。调度器收到通知从数据库或配置中加载对应的“价格监控Agent”配置调度层。调度器初始化该 Agent并传入初始参数如目标商品URL。Agent 开始“思考”根据目标它决定先调用“网页抓取Tool”智能体层。“网页抓取Tool”执行调用底层的axios发起请求并解析 HTML工具层返回结构化的价格数据。Agent 收到数据根据规则判断价格是否低于阈值。如果是则决定调用“企业微信通知Tool”。“企业微信通知Tool”执行通过底层 SDK 发送消息工具层。Agent 任务完成将结果和日志返回给调度器。调度器更新任务状态并可能触发后续任务或告警。3.2 关键模型定义我们需要用 TypeScript 接口来定义几个核心模型确保类型安全。// src/tasks/interfaces/task-definition.interface.ts export interface TaskDefinition { id: string; // 任务唯一标识 name: string; // 任务名称 description?: string; // 任务描述 agentType: string; // 对应的 Agent 类型如 PRICE_MONITOR, DATA_SYNC trigger: TaskTrigger; // 触发配置 input: Recordstring, any; // 任务输入参数 enabled: boolean; // 是否启用 createdAt: Date; updatedAt: Date; } export interface TaskTrigger { type: cron | interval | event | manual; // Cron 表达式 expression?: string; // 间隔毫秒 interval?: number; // 事件名 event?: string; } // src/agents/interfaces/agent.interface.ts import { Tool } from langchain/tools; export interface IAgent { name: string; description: string; // 核心执行方法 run(input: Recordstring, any, context?: AgentRunContext): PromiseAgentResult; } export interface AgentRunContext { taskId: string; executionId: string; logger: Logger; } export interface AgentResult { success: boolean; output?: Recordstring, any; error?: string; logs: string[]; duration: number; } // 我们的自定义 Tool 将实现 LangChain 的 Tool 接口并包装一个底层的功能服务。3.3 项目目录结构规划一个清晰的结构是维护性的保障。src/ ├── app.module.ts ├── main.ts ├── common/ # 通用工具、装饰器、过滤器等 ├── config/ # 配置模块 ├── tasks/ # 任务定义与管理模块 │ ├── tasks.module.ts │ ├── tasks.service.ts # 负责CRUD、触发逻辑 │ ├── tasks.controller.ts │ ├── entities/ # 任务定义实体 │ └── interfaces/ ├── agents/ # 智能体模块核心 │ ├── agents.module.ts │ ├── agents.service.ts # Agent注册中心、执行入口 │ ├── base/ # 基础抽象类 │ ├── types/ # 预定义的Agent类型如 PriceMonitorAgent │ │ ├── price-monitor.agent.ts │ │ └──>nest new openclaw-agent-system -p npm cd openclaw-agent-system npm install nestjs/schedule nestjs/typeorm typeorm sqlite3 # 使用SQLite示例 npm install langchain langchain/core首先创建任务定义实体和模块。// src/tasks/entities/task-definition.entity.ts import { Entity, PrimaryGeneratedColumn, Column, CreateDateColumn, UpdateDateColumn } from typeorm; Entity(task_definitions) export class TaskDefinition { PrimaryGeneratedColumn(uuid) id: string; Column() name: string; Column({ nullable: true }) description: string; Column() agentType: string; // 关联到具体的Agent类 Column(simple-json) // 注意生产环境建议用更规范的JSON字段类型 trigger: { type: string; expression?: string; interval?: number; event?: string }; Column(simple-json, { default: {} }) input: Recordstring, any; Column({ default: true }) enabled: boolean; CreateDateColumn() createdAt: Date; UpdateDateColumn() updatedAt: Date; }// src/tasks/tasks.module.ts import { Module } from nestjs/common; import { TypeOrmModule } from nestjs/typeorm; import { ScheduleModule } from nestjs/schedule; import { TasksService } from ./tasks.service; import { TasksController } from ./tasks.controller; import { TaskDefinition } from ./entities/task-definition.entity; Module({ imports: [ TypeOrmModule.forFeature([TaskDefinition]), ScheduleModule.forRoot(), // 启用定时任务模块 ], controllers: [TasksController], providers: [TasksService], exports: [TasksService], // 导出以便其他模块如Agents使用 }) export class TasksModule {}4.2 实现任务调度器服务TasksService需要做两件事1. 管理任务定义的CRUD2. 根据任务定义动态地创建和管理 NestJS 的定时任务。// src/tasks/tasks.service.ts import { Injectable, Logger, OnModuleInit } from nestjs/common; import { InjectRepository } from nestjs/typeorm; import { Repository } from typeorm; import { SchedulerRegistry } from nestjs/schedule; import { CronJob } from cron; import { TaskDefinition } from ./entities/task-definition.entity; import { AgentsService } from ../agents/agents.service; // 稍后实现 Injectable() export class TasksService implements OnModuleInit { private readonly logger new Logger(TasksService.name); private jobMap new Mapstring, CronJob(); // 存储动态创建的CronJob引用 constructor( InjectRepository(TaskDefinition) private tasksRepo: RepositoryTaskDefinition, private schedulerRegistry: SchedulerRegistry, private agentsService: AgentsService, // 注入Agent服务 ) {} async onModuleInit() { // 应用启动时从数据库加载所有启用的任务并初始化定时触发器 const enabledTasks await this.tasksRepo.find({ where: { enabled: true } }); for (const task of enabledTasks) { if (task.trigger.type cron task.trigger.expression) { this.scheduleCronTask(task); } // 可以扩展其他触发类型如 interval } } private scheduleCronTask(task: TaskDefinition) { const job new CronJob(task.trigger.expression, async () { this.logger.log(触发定时任务: ${task.name} (${task.id})); try { // 这里是核心触发Agent执行 await this.agentsService.executeAgent(task.agentType, task.input, { taskId: task.id, executionId: exec_${Date.now()}, }); } catch (error) { this.logger.error(执行任务 ${task.name} 失败:, error.message); } }); job.start(); this.jobMap.set(task.id, job); this.schedulerRegistry.addCronJob(task-${task.id}, job); this.logger.log(已调度任务: ${task.name} (${task.trigger.expression})); } // 当通过API新增或启用一个任务时调用此方法 async addAndScheduleTask(createTaskDto: any) { const task this.tasksRepo.create(createTaskDto); await this.tasksRepo.save(task); if (task.enabled task.trigger.type cron) { this.scheduleCronTask(task); } return task; } // 禁用或删除任务时停止对应的CronJob async disableTask(taskId: string) { const jobName task-${taskId}; if (this.schedulerRegistry.doesExist(cron, jobName)) { const job this.schedulerRegistry.getCronJob(jobName); job.stop(); this.schedulerRegistry.deleteCronJob(jobName); this.jobMap.delete(taskId); } await this.tasksRepo.update(taskId, { enabled: false }); } // ... 其他CRUD方法 }4.3 构建 LangChain Tool 抽象与具体实现这是连接底层功能与上层 Agent 的关键。我们先创建一个抽象基类方便统一管理。// src/agents/tools/base.tool.ts import { Tool } from langchain/tools; import { Logger } from nestjs/common; export abstract class BaseTool extends Tool { protected logger: Logger; constructor(name: string, description: string) { super(); this.name name; this.description description; this.logger new Logger(Tool:${name}); } // LangChain Tool 要求实现的方法 abstract _call(arg: string): Promisestring; // 一个辅助方法用于解析 JSON 格式的输入参数 protected parseInput(arg: string): Recordstring, any { try { return JSON.parse(arg); } catch { // 如果不是JSON可能只是简单字符串参数 return { input: arg }; } } // 一个辅助方法用于标准化输出 protected formatOutput(result: any): string { return JSON.stringify({ success: true, data: result }); } protected formatError(error: any): string { return JSON.stringify({ success: false, error: error.message }); } }现在实现一个具体的网页抓取 Tool。// src/agents/tools/web-crawler.tool.ts import { Injectable } from nestjs/common; import axios from axios; import * as cheerio from cheerio; // 需要安装npm install cheerio import { BaseTool } from ./base.tool; Injectable() // 标记为可注入的 Provider export class WebCrawlerTool extends BaseTool { constructor() { super( web_crawler, 一个用于抓取网页并提取特定信息的工具。输入应为JSON字符串包含url和selector字段。, ); } async _call(arg: string): Promisestring { const params this.parseInput(arg); const { url, selector } params; if (!url || !selector) { return this.formatError(new Error(参数必须包含 url 和 selector)); } try { this.logger.log(开始抓取: ${url}, 选择器: ${selector}); const response await axios.get(url, { timeout: 10000, headers: { User-Agent: OpenClaw-Agent/1.0 }, }); const $ cheerio.load(response.data); const content $(selector).text().trim(); const result { url, content, timestamp: new Date().toISOString() }; this.logger.log(抓取成功内容长度: ${content.length}); return this.formatOutput(result); } catch (error) { this.logger.error(抓取失败: ${url}, error.message); return this.formatError(error); } } }4.4 实现 Agent 服务与第一个智能体AgentsService是智能体层的入口和注册中心。// src/agents/agents.service.ts import { Injectable, Logger, OnModuleInit, NotFoundException } from nestjs/common; import { IAgent, AgentRunContext, AgentResult } from ./interfaces/agent.interface; Injectable() export class AgentsService implements OnModuleInit { private readonly logger new Logger(AgentsService.name); private agentRegistry new Mapstring, IAgent(); // Agent类型 - 实例的映射 // 在模块初始化时注册所有Agent async onModuleInit() { // 这里可以通过动态导入或依赖注入来注册为了清晰我们先手动注册 // 实际项目中可以使用装饰器或模块扫描机制自动注册 } // 注册Agent的方法 registerAgent(type: string, agent: IAgent) { if (this.agentRegistry.has(type)) { this.logger.warn(Agent类型 ${type} 已存在将被覆盖。); } this.agentRegistry.set(type, agent); this.logger.log(已注册Agent: ${type}); } // 执行Agent的核心方法 async executeAgent( agentType: string, input: Recordstring, any, context: AgentRunContext, ): PromiseAgentResult { const agent this.agentRegistry.get(agentType); if (!agent) { throw new NotFoundException(未找到类型为 ${agentType} 的Agent); } this.logger.log(开始执行Agent: ${agentType}, 任务ID: ${context.taskId}); const startTime Date.now(); const logs: string[] []; // 可以创建一个带有日志收集功能的上下文 const agentContext { ...context, logger: { log: (message: string) { logs.push([INFO] ${message}); context.logger.log([${agentType}] ${message}); }, error: (message: string) { logs.push([ERROR] ${message}); context.logger.error([${agentType}] ${message}); }, }, }; try { const result await agent.run(input, agentContext); const duration Date.now() - startTime; this.logger.log(Agent执行完成: ${agentType}, 耗时: ${duration}ms); return { ...result, logs, duration, }; } catch (error) { const duration Date.now() - startTime; this.logger.error(Agent执行异常: ${agentType}, error.stack); return { success: false, error: error.message, logs: [...logs, [FATAL] ${error.message}], duration, }; } } // 获取所有已注册的Agent类型 getRegisteredAgents(): string[] { return Array.from(this.agentRegistry.keys()); } }接下来实现一个不依赖 LLM 的、基于规则的价格监控 Agent。它内部会使用我们定义的 Tools。// src/agents/types/price-monitor.agent.ts import { Injectable } from nestjs/common; import { IAgent, AgentRunContext, AgentResult } from ../interfaces/agent.interface; import { WebCrawlerTool } from ../tools/web-crawler.tool; import { NotificationTool } from ../tools/notification.tool; // 假设已实现 Injectable() export class PriceMonitorAgent implements IAgent { name 价格监控Agent; description 监控指定商品页面的价格并在低于阈值时发送通知。; constructor( private readonly webCrawlerTool: WebCrawlerTool, private readonly notificationTool: NotificationTool, ) {} async run(input: Recordstring, any, context: AgentRunContext): PromiseAgentResult { const { url, priceSelector, threshold, notificationTarget } input; const logs: string[] []; const logger context.logger; logger.log(开始监控任务目标URL: ${url}); logs.push(开始监控任务目标URL: ${url}); // 步骤1: 抓取页面 const crawlInput JSON.stringify({ url, selector: priceSelector }); const crawlResultStr await this.webCrawlerTool._call(crawlInput); let crawlResult; try { crawlResult JSON.parse(crawlResultStr); if (!crawlResult.success) { throw new Error(抓取失败: ${crawlResult.error}); } } catch (e) { const errorMsg 解析抓取结果失败: ${e.message}; logger.error(errorMsg); return { success: false, error: errorMsg, logs, duration: 0 }; } const rawContent crawlResult.data.content; logger.log(抓取到的原始内容: ${rawContent}); logs.push(抓取到的原始内容: ${rawContent}); // 步骤2: 提取并清洗价格这里是一个简单的示例实际可能更复杂 const priceMatch rawContent.match(/(\d(\.\d)?)/); if (!priceMatch) { const errorMsg 无法从内容中提取价格数字; logger.error(errorMsg); return { success: false, error: errorMsg, logs, duration: 0 }; } const currentPrice parseFloat(priceMatch[1]); logger.log(解析出的当前价格: ${currentPrice}); logs.push(解析出的当前价格: ${currentPrice}); // 步骤3: 规则判断 if (currentPrice threshold) { logger.log(价格 ${currentPrice} 低于阈值 ${threshold}触发通知。); logs.push(价格 ${currentPrice} 低于阈值 ${threshold}触发通知。); // 步骤4: 发送通知 const notifyInput JSON.stringify({ target: notificationTarget, title: 价格监控告警 - ${url}, content: 商品价格已降至 ${currentPrice}低于设定的阈值 ${threshold}。, }); const notifyResultStr await this.notificationTool._call(notifyInput); // ... 处理通知结果 logs.push(通知发送请求已执行。); } else { logger.log(价格 ${currentPrice} 未低于阈值 ${threshold}本次不通知。); logs.push(价格 ${currentPrice} 未低于阈值 ${threshold}本次不通知。); } return { success: true, output: { currentPrice, threshold, triggered: currentPrice threshold }, logs, duration: 0, // 实际应由外部计算 }; } }4.5 组装模块将一切连接起来最后我们需要创建一个AgentsModule将 Tools、Agents 和 Service 组装起来并注册到AgentsService中。// src/agents/agents.module.ts import { Module, OnModuleInit } from nestjs/common; import { AgentsService } from ./agents.service; import { PriceMonitorAgent } from ./types/price-monitor.agent; import { WebCrawlerTool } from ./tools/web-crawler.tool; import { NotificationTool } from ./tools/notification.tool; // 假设已实现 Module({ providers: [ AgentsService, WebCrawlerTool, NotificationTool, PriceMonitorAgent, // 注册具体的Agent ], exports: [AgentsService], // 导出供TasksModule使用 }) export class AgentsModule implements OnModuleInit { constructor( private readonly agentsService: AgentsService, private readonly priceMonitorAgent: PriceMonitorAgent, ) {} onModuleInit() { // 手动注册Agent到服务中心 this.agentsService.registerAgent(PRICE_MONITOR, this.priceMonitorAgent); // 未来可以在这里注册更多Agent如 DATA_SYNC } }别忘了在主模块AppModule中导入TasksModule和AgentsModule并配置 TypeORM。// src/app.module.ts import { Module } from nestjs/common; import { TypeOrmModule } from nestjs/typeorm; import { ScheduleModule } from nestjs/schedule; import { TasksModule } from ./tasks/tasks.module; import { AgentsModule } from ./agents/agents.module; import { TaskDefinition } from ./tasks/entities/task-definition.entity; Module({ imports: [ ScheduleModule.forRoot(), TypeOrmModule.forRoot({ type: sqlite, database: database.sqlite, entities: [TaskDefinition], synchronize: true, // 生产环境请设为false使用迁移 }), TasksModule, AgentsModule, ], }) export class AppModule {}5. 进阶接入 LangChain Agent 实现动态决策上面的PriceMonitorAgent是一个“硬编码”规则的 Agent。它的决策逻辑价格比较是写死的。如果我们想让 Agent 能处理更模糊、更复杂的指令比如“帮我监控一下XX商品如果降价或者有优惠活动就告诉我”就需要引入 LLM 作为“大脑”。5.1 创建基于 LLM 的智能体我们将使用 LangChain 的 ReAct 框架来创建一个能使用 Tools 的 LLM Agent。首先确保安装了 OpenAI 包或其他 LLM 提供商。npm install langchain/openai然后创建一个新的 Agent。// src/agents/types/llm-assistant.agent.ts import { Injectable } from nestjs/common; import { ChatOpenAI } from langchain/openai; import { AgentExecutor, createReactAgent } from langchain/agents; import { DynamicTool } from langchain/tools; import { IAgent, AgentRunContext, AgentResult } from ../interfaces/agent.interface; import { WebCrawlerTool } from ../tools/web-crawler.tool; import { NotificationTool } from ../tools/notification.tool; Injectable() export class LlmAssistantAgent implements IAgent { name LLM助手Agent; description 一个能理解自然语言指令并使用工具完成任务的通用智能体。; private agentExecutor: AgentExecutor; constructor( private readonly webCrawlerTool: WebCrawlerTool, private readonly notificationTool: NotificationTool, ) {} async onModuleInit() { // 初始化LLM。请将API KEY放在环境变量中。 const llm new ChatOpenAI({ modelName: gpt-3.5-turbo, temperature: 0, openAIApiKey: process.env.OPENAI_API_KEY, }); // 将我们的NestJS Tool适配成LangChain的Tool接口 const tools [ new DynamicTool({ name: this.webCrawlerTool.name, description: this.webCrawlerTool.description, func: async (input: string) this.webCrawlerTool._call(input), }), new DynamicTool({ name: this.notificationTool.name, description: this.notificationTool.description, func: async (input: string) this.notificationTool._call(input), }), // ... 可以添加更多工具 ]; // 使用ReAct框架创建Agent const agent await createReactAgent({ llm, tools, }); this.agentExecutor new AgentExecutor({ agent, tools, maxIterations: 5, // 限制最大思考步骤防止死循环 }); } async run(input: Recordstring, any, context: AgentRunContext): PromiseAgentResult { const { instruction } input; // 用户输入的自然语言指令 const logs: string[] []; const logger context.logger; if (!this.agentExecutor) { throw new Error(LLM Agent 未正确初始化); } logger.log(开始处理指令: ${instruction}); logs.push(用户指令: ${instruction}); try { // 执行LangChain Agent const result await this.agentExecutor.invoke({ input: instruction, // 可以传入一些上下文信息 }); logger.log(Agent执行完成。输出: ${result.output}); logs.push(Agent思考过程: ${JSON.stringify(result.intermediateSteps)}); // 中间步骤很有用 logs.push(最终输出: ${result.output}); return { success: true, output: { response: result.output }, logs, duration: 0, }; } catch (error) { logger.error(LLM Agent执行失败:, error); return { success: false, error: LLM Agent执行失败: ${error.message}, logs, duration: 0, }; } } }在AgentsModule中注册这个新的 Agent。// 在 AgentsModule 的 providers 数组中添加 providers: [ // ... 其他 providers LlmAssistantAgent, ], // 在 onModuleInit 方法中注册 onModuleInit() { this.agentsService.registerAgent(PRICE_MONITOR, this.priceMonitorAgent); this.agentsService.registerAgent(LLM_ASSISTANT, this.llmAssistantAgent); // 新增 }现在你可以创建一个类型为LLM_ASSISTANT的任务其输入参数instruction可以是“请去抓取 https://example.com/product/123 这个页面的价格如果价格低于100元就发个通知给我。” Agent 会自动理解指令并调用相应的 Tools 去执行。5.2 任务编排与复杂工作流单个 Agent 能力有限。更复杂的场景需要多个 Agent 或步骤按顺序或条件执行。这可以在调度层实现即创建一个“工作流 Agent”它本身不干活只负责调用其他 Agent。// src/agents/types/workflow.agent.ts Injectable() export class WorkflowAgent implements IAgent { constructor(private readonly agentsService: AgentsService) {} async run(input: Recordstring, any, context: AgentRunContext): PromiseAgentResult { const { steps } input; // steps: Array{ agentType: string, input: any, condition?} const logs []; const results []; for (const [index, step] of steps.entries()) { // 可以在这里添加条件判断根据上一步的结果决定是否执行当前步骤 const stepResult await this.agentsService.executeAgent( step.agentType, step.input, { ...context, executionId: ${context.executionId}_step${index} }, ); results.push(stepResult); logs.push(...stepResult.logs); if (!stepResult.success) { // 可以定义工作流的失败策略继续、停止、重试等 return { success: false, error: 工作流第${index 1}步失败: ${stepResult.error}, logs, duration: results.reduce((sum, r) sum r.duration, 0), }; } } return { success: true, output: { steps: results }, logs, duration: results.reduce((sum, r) sum r.duration, 0), }; } }这样你就可以通过配置一个WORKFLOW类型的任务来串联起“数据抓取 - 数据分析 - 通知”这样的流水线。6. 生产环境考量与避坑指南将这样一个系统投入生产还需要解决很多实际问题。6.1 性能、并发与队列管理问题定时任务可能密集触发或者某个 Agent 执行时间很长。直接在 Cron 回调中执行agentsService.executeAgent会导致请求堆积可能拖垮服务。解决方案引入任务队列。可以使用nestjs/bull基于 Redis或bullmq。TasksService的 Cron 回调不再直接执行 Agent而是向一个队列如agent-jobs添加一个任务。单独启动一个或多个工作进程Worker来消费队列中的任务并调用AgentsService。这样可以实现削峰填谷、失败重试、任务优先级、延迟执行等高级特性。// 简化的示例 // tasks.service.ts (Cron回调部分) import { InjectQueue } from nestjs/bull; import { Queue } from bull; constructor(InjectQueue(agent-jobs) private agentQueue: Queue) {} private async scheduleCronTask(task: TaskDefinition) { const job new CronJob(task.trigger.expression, async () { this.logger.log(触发定时任务加入队列: ${task.name}); await this.agentQueue.add(execute-agent, { agentType: task.agentType, input: task.input, taskId: task.id, }); }); // ... 启动job } // 在另一个模块中定义消费者 Processor(agent-jobs) export class AgentJobsConsumer { constructor(private agentsService: AgentsService) {} Process(execute-agent) async handleAgentJob(job: Job) { const { agentType, input, taskId } job.data; await this.agentsService.executeAgent(agentType, input, { taskId, executionId: job.id.toString(), }); } }6.2 可观测性与日志追踪集中式日志确保所有logger.log/error都输出到像 Winston 这样的日志库并配置好传输到 ELK、Loki 等集中式日志系统。在AgentRunContext中传递一个带有唯一executionId的 logger便于追踪一次任务执行的完整链路。执行记录持久化每次 Agent 执行的结果AgentResult都应该存入数据库ExecutionRecord表包括任务ID、执行ID、状态、输入、输出、日志、耗时、开始结束时间。这是事后排查和数据分析的基础。健康检查使用nestjs/terminus为你的 Agent 系统添加健康检查端点监控关键依赖如数据库、Redis、LLM API的状态。6.3 错误处理与重试机制Agent/Tool 级错误每个 Tool 和 Agent 的_call/run方法内部必须有完善的 try-catch并返回结构化的错误信息而不是直接抛出异常导致进程崩溃。网络与外部 API 容错对于网页抓取、调用第三方 API 等操作必须设置合理的超时timeout和重试策略retry。可以使用axios-retry等库。队列的失败处理如果使用 Bull可以配置任务的失败重试次数、失败后的处理策略如放入另一个失败队列供人工检查。6.4 配置与安全性敏感信息管理LLM API Key、数据库密码、第三方服务密钥等必须通过环境变量或配置中心如 Consul管理绝对不要硬编码在代码中。NestJS 的nestjs/config模块是很好的选择。Tool 的权限控制不是所有 Agent 都能调用所有 Tool。可以在AgentsService.executeAgent中增加权限检查逻辑根据agentType决定其可用的 Tools 列表。输入验证与清理对于从任务配置或用户指令传入的输入尤其是 URL、选择器必须进行严格的验证和清理防止注入攻击如 SSRF、XSS。6.5 我踩过的几个坑LangChain Tool 的输入输出LangChain 的 Tool 默认期望输入和输出都是字符串。这要求我们在设计 Tool 时必须将复杂的参数序列化为 JSON 字符串输出也要序列化。一开始我直接传递对象导致 Agent 无法正确解析。最佳实践是在 Tool 内部约定好 JSON 的输入输出格式并在description中写清楚字段说明。Agent 的无限循环在早期测试 LLM Agent 时我给了一个模糊的指令Agent 陷入了“思考-调用工具-结果不理想-再思考”的死循环直到达到maxIterations限制。一定要设置maxIterations比如 10-15并为 Agent 提供清晰、具体的工具描述引导它更有效地完成任务。TypeORM 与 NestJS 的异步生命周期在onModuleInit中初始化数据库连接和注册 Agent 时如果初始化逻辑是异步的要确保使用await或者将复杂的初始化移到OnApplicationBootstrap生命周期钩子中避免在服务未就绪时处理请求。定时任务的时间漂移Node.js 的setInterval或setTimeout并不精确长时间运行会有漂移。node-cron库nestjs/schedule的底层基于系统时间相对准确但在高负载或事件循环阻塞时仍可能延迟。对于要求绝对准点的任务需要考虑分布式锁和更精确的调度方案。构建这样一个 OpenClaw 式的 Agent 定时任务系统初期的架构和抽象设计会花费一些时间但一旦核心框架搭好后续增加新的“爪牙”Tool和“大脑”Agent会变得非常快速和规范。它带来的灵活性、可维护性和智能化潜力是传统定时任务脚本无法比拟的。你可以从一个小而美的价格监控开始逐步将它扩展成团队内部强大的自动化中枢。