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 以下几点：

1. 依赖注入（DI）与模块化架构：这是实现“OpenClaw 式”可插拔性的基石。每个“爪牙”（例如，一个数据抓取器、一个邮件发送器、一个数据分析模块）都可以被封装成一个独立的Provider，并通过 NestJS 的 DI 容器进行管理。大脑（调度器）不需要知道爪牙的具体实现，只需要知道它的接口（Token），就可以动态地获取和使用它。这使得增加、替换或移除功能模块变得异常简单，完全符合开闭原则。
2. 强大的生命周期与可观测性：NestJS 内置了生命周期钩子（OnModuleInit,OnApplicationBootstrap等），非常适合在应用启动时初始化我们的 Agent 系统、加载配置、连接外部服务。同时，结合像@nestjs/terminus这样的健康检查库，可以轻松暴露我们 Agent 系统的运行状态，方便监控。
3. TypeScript 原生支持与工程化：对于构建一个可能日益复杂的 Agent 系统，类型安全是防止低级错误、提升开发体验的关键。NestJS 与 TypeScript 的深度集成，让我们的“爪牙”接口定义、数据流传递都非常清晰。其 CLI 工具和项目结构也强制了一种良好的工程实践，有利于团队协作和长期维护。
4. 丰富的生态系统：对于定时任务，NestJS 有官方的@nestjs/schedule模块，它基于node-cron和rxjs，提供了装饰器和 API 两种方式来定义任务，并且能与 NestJS 的 DI 完美集成，让我们的任务类也能方便地注入其他服务。

注意：不要一上来就用@nestjs/schedule的@Cron装饰器定义我们的智能任务。我们将把它作为底层的时间触发器，而上层的智能调度逻辑由我们自己的 Agent 服务来实现。

2.2 LangChain：为 Agent 注入“思考”能力

LangChain 是一个用于构建 LLM 应用的框架。在我们的场景中，LLM 并非必需，但 LangChain 提供的抽象极其有价值。

1. Agent 与 Tools 的标准化抽象：LangChain 的核心概念之一就是Agent。一个 LangChain Agent 由三部分组成：一个 LLM（思考核心）、一组Tools（可执行的动作）、以及一个决定如何调用 Tools 的AgentExecutor（决策循环）。即使我们不使用 LLM，也可以借鉴其Tool的抽象。我们可以将每一个“爪牙”定义为一个 LangChain Tool，它有明确的name,description和execute方法。这样，所有功能单元就有了统一的接口。
2. 工作流（Chain）的编排能力：LangChain 的Chain概念允许我们将多个步骤（LLM调用、Tool调用、数据转换）串联起来。这对于实现复杂的、多步骤的自动化任务非常有用。我们可以构建一个“价格监控链”：先调用“网页抓取Tool”，再调用“数据分析Tool”，最后调用“通知Tool”。
3. 与外部世界的连接器：LangChain 社区提供了海量的集成（Toolkits），从搜索引擎、数据库到各种 SaaS API。这意味着我们的 Agent 系统可以轻松获得“感知”和“操作”外部世界的能力。例如，直接使用SerpAPITool 进行搜索，或用SQLDatabaseToolkit来查询数据库。
4. 为未来接入 LLM 预留可能性：也许当前的任务还不需要自然语言理解或生成。但一旦需求升级，比如需要让系统根据自然语言描述自动生成任务流程，或者让 Agent 能理解模糊的告警信息并自主决策，那么基于 LangChain 构建的底层架构可以无缝地接入 LLM，只需替换或增强 Agent 的“思考”部分即可，而所有的 Tools（爪牙）都可以复用。

两者的结合点：NestJS 作为宿主容器，管理着所有服务（包括 LangChain 的 Agents 和 Tools）的生命周期、配置和依赖关系。而 LangChain 提供的抽象，则让 NestJS 容器内的这些服务能够以一种标准化的、高内聚低耦合的方式进行交互和编排。我们可以创建一个LangChainService在 NestJS 中，它负责初始化和管理所有的 Tools 和 Agents。

3. 系统核心架构设计

基于以上选型，我们来设计系统的核心架构。我们的目标是构建一个“智能任务调度中心”。

3.1 架构分层与组件职责

整个系统可以清晰地分为四层：

数据流：

1. 一个 Cron 表达式触发（触发层）。
2. 调度器收到通知，从数据库或配置中加载对应的“价格监控Agent”配置（调度层）。
3. 调度器初始化该 Agent，并传入初始参数（如目标商品URL）。
4. Agent 开始“思考”：根据目标，它决定先调用“网页抓取Tool”（智能体层）。
5. “网页抓取Tool”执行，调用底层的axios发起请求并解析 HTML（工具层），返回结构化的价格数据。
6. Agent 收到数据，根据规则判断价格是否低于阈值。如果是，则决定调用“企业微信通知Tool”。
7. “企业微信通知Tool”执行，通过底层 SDK 发送消息（工具层）。
8. Agent 任务完成，将结果和日志返回给调度器。
9. 调度器更新任务状态，并可能触发后续任务或告警。

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: Record<string, 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: Record<string, any>, context?: AgentRunContext): Promise<AgentResult>; } export interface AgentRunContext { taskId: string; executionId: string; logger: Logger; } export interface AgentResult { success: boolean; output?: Record<string, 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: Record<string, 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. 管理任务定义的CRUD；2. 根据任务定义，动态地创建和管理 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 Map<string, CronJob>(); // 存储动态创建的CronJob引用 constructor( @InjectRepository(TaskDefinition) private tasksRepo: Repository<TaskDefinition>, 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): Promise<string>; // 一个辅助方法，用于解析 JSON 格式的输入参数 protected parseInput(arg: string): Record<string, 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): Promise<string> { 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 Map<string, 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: Record<string, any>, context: AgentRunContext, ): Promise<AgentResult> { 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: Record<string, any>, context: AgentRunContext): Promise<AgentResult> { 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: Record<string, any>, context: AgentRunContext): Promise<AgentResult> { 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: Record<string, any>, context: AgentRunContext): Promise<AgentResult> { 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 我踩过的几个坑

1. LangChain Tool 的输入输出：LangChain 的 Tool 默认期望输入和输出都是字符串。这要求我们在设计 Tool 时，必须将复杂的参数序列化为 JSON 字符串，输出也要序列化。一开始我直接传递对象，导致 Agent 无法正确解析。最佳实践是：在 Tool 内部约定好 JSON 的输入输出格式，并在description中写清楚字段说明。
2. Agent 的无限循环：在早期测试 LLM Agent 时，我给了一个模糊的指令，Agent 陷入了“思考-调用工具-结果不理想-再思考”的死循环，直到达到maxIterations限制。一定要设置maxIterations（比如 10-15），并为 Agent 提供清晰、具体的工具描述，引导它更有效地完成任务。
3. TypeORM 与 NestJS 的异步生命周期：在onModuleInit中初始化数据库连接和注册 Agent 时，如果初始化逻辑是异步的，要确保使用await，或者将复杂的初始化移到OnApplicationBootstrap生命周期钩子中，避免在服务未就绪时处理请求。
4. 定时任务的时间漂移：Node.js 的setInterval或setTimeout并不精确，长时间运行会有漂移。node-cron库（@nestjs/schedule的底层）基于系统时间，相对准确，但在高负载或事件循环阻塞时仍可能延迟。对于要求绝对准点的任务，需要考虑分布式锁和更精确的调度方案。

构建这样一个 OpenClaw 式的 Agent 定时任务系统，初期的架构和抽象设计会花费一些时间，但一旦核心框架搭好，后续增加新的“爪牙”（Tool）和“大脑”（Agent）会变得非常快速和规范。它带来的灵活性、可维护性和智能化潜力，是传统定时任务脚本无法比拟的。你可以从一个小而美的价格监控开始，逐步将它扩展成团队内部强大的自动化中枢。