1. 从单体 Skill 到可测试模块我踩过的三个坑一个 Skill 写着写着就变成上帝类是很多做智能体工具链的同学都会遇到的阶段。我最初做项目管理类 Skill 时所有逻辑塞在一个index.ts里命令解析、数据库读写、报表生成、消息通知全在一个文件改一处要重新读三百行。这个场景的核心检索词就是 Skill 架构设计、依赖注入、TypeORM、BullMQ——它们分别解决怎么拆怎么换怎么存怎么异步四个问题。具体痛点有三个。第一是测试困难new ProjectService()里直接new ProjectRepository()想跑单元测试就必须连真实数据库CI 里跑一次要几十秒。第二是同步阻塞生成一份月度报表要查几千条任务记录再拼 PDF用户发一条命令后 Skill 卡住十几秒超时直接失败。第三是失败无重试报表生成中途 Redis 抖动一下任务就丢了用户只能重新发命令。这篇要交付的东西很明确一套可复制的config.toml骨架、TaoToken 统一 Key 的配置片段、TypeORM 实体与仓储的依赖注入写法、BullMQ 队列的生产者与消费者代码以及队列消费、依赖替换、失败重试三个可验证动作。适合已经写过一两个 Skill、想把代码从能跑推进到能维护的开发者。下面所有代码都是 TypeScriptNode 18 以上可直接跑。2. TaoToken 前置统一 Key 与 config.toml 骨架在拆架构之前先把模型调用的出口统一掉。Skill 里如果散落着各种模型 API 地址和 Key重构时你会分不清哪些是业务逻辑、哪些是接入细节。我的做法是把模型调用收敛成一个LlmClient通过依赖注入传给需要的服务而它读取的配置来自 TaoToken。TaoToken 在这里的角色是统一模型接入层一个 Key、一个兼容 OpenAI 风格的接口地址就能在 Skill 里调用不同模型不用为每个供应商维护一套鉴权代码。官网入口是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 基址是 https://taotoken.net/api 这个地址不加 UTM 参数。先建 Key进入控制台 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 在 API Keys 页面 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 创建一个 Key复制出来只显示一次先存进环境变量。如果你后面要做长期编码或 Agent 类任务可以看 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 想先在网页里验证模型是否通用模型对话 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 试一句即可。下面是config.toml骨架把模型、数据库、Redis 三段配置分开环境变量用${VAR}占位避免把 Key 写进仓库# config.toml —— Skill 运行时配置骨架 [app] name project-management-skill env development log_level info [llm] # TaoToken 统一接入一个 Key 走所有模型 base_url https://taotoken.net/api api_key ${TAOTOKEN_API_KEY} default_model gpt-4o-mini timeout_ms 30000 max_retries 2 [database] type postgres host ${DB_HOST} port 5432 username ${DB_USER} password ${DB_PASSWORD} database skill_pm synchronize false # 生产必须 false用 migration logging [error, warn] [redis] host ${REDIS_HOST} port 6379 password ${REDIS_PASSWORD} db 0 [queue] name report-generation concurrency 4 attempts 3 backoff_type exponential backoff_delay_ms 1000 remove_on_complete true remove_on_fail 100对应的.env只放敏感值.gitignore里排除掉TAOTOKEN_API_KEYsk-你的Key DB_HOST127.0.0.1 DB_USERskill DB_PASSWORDchange_me REDIS_HOST127.0.0.1 REDIS_PASSWORD读取配置用一个薄封装别在业务代码里到处process.env// src/config/env.ts import fs from node:fs; import TOML from iarna/toml; const raw fs.readFileSync(./config.toml, utf-8); const parsed TOML.parse(raw) as Recordstring, any; function expand(value: string): string { return value.replace(/\$\{(\w)\}/g, (_, key) process.env[key] ?? ); } export const config { llm: { baseUrl: parsed.llm.base_url as string, apiKey: expand(parsed.llm.api_key as string), defaultModel: parsed.llm.default_model as string, timeoutMs: parsed.llm.timeout_ms as number, }, database: { host: expand(parsed.database.host as string), port: parsed.database.port as number, username: expand(parsed.database.username as string), password: expand(parsed.database.password as string), database: parsed.database.database as string, }, redis: { host: expand(parsed.redis.host as string), port: parsed.redis.port as number, password: expand(parsed.redis.password as string), }, queue: parsed.queue as Recordstring, any, };注意synchronize false是硬性要求。TypeORM 的自动同步在开发期方便但一旦表里有数据它会悄悄改结构线上出过事故的都懂。3. 可复制配置依赖注入容器 TypeORM 实体 BullMQ 队列3.1 用接口隔离依赖让 Service 不认具体实现依赖注入的核心不是引入某个容器库而是让 Service 依赖接口而不是具体类。先定义仓储接口// src/interfaces/repository.ts import { Project } from ../entities/project; export interface IProjectRepository { create(data: PartialProject): PromiseProject; findByUserId(userId: string): PromiseProject[]; findById(id: string): PromiseProject | null; } export interface INotificationService { send(userId: string, title: string, body: string): Promisevoid; }Service 只认接口构造函数注入// src/services/project.service.ts import { IProjectRepository, INotificationService } from ../interfaces/repository; import { Project } from ../entities/project; export class ProjectService { constructor( private readonly repo: IProjectRepository, private readonly notifier: INotificationService, ) {} async createProject(input: { name: string; description?: string; userId: string }): PromiseProject { const project await this.repo.create({ name: input.name, description: input.description ?? , userId: input.userId, status: active, }); await this.notifier.send(input.userId, 项目创建成功, 项目 ${project.name} 已创建); return project; } }这样写单元测试时传一个内存假仓储就行不用连数据库// tests/unit/project.service.test.ts import { ProjectService } from ../../src/services/project.service; const fakeRepo { create: async (d: any) ({ id: p1, ...d }), findByUserId: async () [], findById: async () null, }; const fakeNotifier { send: async () {} }; test(创建项目后发送通知, async () { const svc new ProjectService(fakeRepo as any, fakeNotifier); const p await svc.createProject({ name: Demo, userId: u1 }); expect(p.id).toBe(p1); });3.2 TypeORM 实体与仓储实现实体定义保持扁平索引单独在 migration 里加// src/entities/project.ts import { Entity, PrimaryGeneratedColumn, Column, CreateDateColumn, UpdateDateColumn, Index } from typeorm; Entity(project) export class Project { PrimaryGeneratedColumn(uuid) id!: string; Column({ length: 255 }) name!: string; Column({ type: text, nullable: true }) description!: string | null; Index() Column({ name: user_id, length: 255 }) userId!: string; Column({ length: 50, default: active }) status!: active | archived | deleted; CreateDateColumn({ name: created_at }) createdAt!: Date; UpdateDateColumn({ name: updated_at }) updatedAt!: Date; }仓储实现接口内部用 TypeORM 的Repository// src/repositories/project.repository.ts import { AppDataSource } from ../config/data-source; import { Project } from ../entities/project; import { IProjectRepository } from ../interfaces/repository; export class ProjectRepository implements IProjectRepository { private repo AppDataSource.getRepository(Project); async create(data: PartialProject): PromiseProject { const entity this.repo.create(data); return this.repo.save(entity); } async findByUserId(userId: string): PromiseProject[] { return this.repo.find({ where: { userId, status: active }, order: { createdAt: DESC }, }); } async findById(id: string): PromiseProject | null { return this.repo.findOne({ where: { id } }); } }数据源单独一个文件方便测试时替换// src/config/data-source.ts import reflect-metadata; import { DataSource } from typeorm; import { config } from ./env; import { Project } from ../entities/project; export const AppDataSource new DataSource({ type: postgres, host: config.database.host, port: config.database.port, username: config.database.username, password: config.database.password, database: config.database.database, entities: [Project], migrations: [src/migrations/*.ts], synchronize: false, });3.3 BullMQ 队列生产者与消费者分离报表生成这类耗时任务必须异步化。队列定义单独一个模块生产者和消费者都从这里拿连接// src/queues/report.queue.ts import { Queue, Worker, Job, QueueEvents } from bullmq; import { config } from ../config/env; const connection { host: config.redis.host, port: config.redis.port, password: config.redis.password || undefined, }; export const reportQueue new Queue(config.queue.name, { connection }); export interface ReportJobData { reportId: string; projectId: string; format: pdf | xlsx; userId: string; } export async function addReportJob(data: ReportJobData, priority 10): Promisestring { const job await reportQueue.add(generate, data, { priority, attempts: config.queue.attempts, backoff: { type: config.queue.backoff_type, delay: config.queue.backoff_delay_ms, }, removeOnComplete: config.queue.remove_on_complete, removeOnFail: config.queue.remove_on_fail, }); return job.id as string; }消费者单独进程启动处理器里通过依赖注入拿 Service// src/workers/report.worker.ts import { Worker, Job } from bullmq; import { config } from ../config/env; import { ReportService } from ../services/report.service; import { ProjectRepository } from ../repositories/project.repository; import { LlmClient } from ../clients/llm.client; const connection { host: config.redis.host, port: config.redis.port, password: config.redis.password || undefined, }; // 组合根在这里完成依赖装配 const reportService new ReportService( new ProjectRepository(), new LlmClient(), ); export const reportWorker new Worker( config.queue.name, async (job: Job) { const { reportId, projectId, format } job.data; await job.updateProgress(10); const project await reportService.loadProject(projectId); await job.updateProgress(40); const summary await reportService.summarize(project); await job.updateProgress(70); const filePath await reportService.export(reportId, summary, format); await job.updateProgress(100); return { filePath }; }, { connection, concurrency: config.queue.concurrency, }, ); reportWorker.on(failed, (job, err) { console.error([job ${job?.id}] 失败: ${err.message}); });LlmClient就是前面说的统一出口读 TaoToken 配置// src/clients/llm.client.ts import { config } from ../config/env; export class LlmClient { async chat(prompt: string, model config.llm.defaultModel): Promisestring { const res await fetch(${config.llm.baseUrl}/v1/chat/completions, { method: POST, headers: { Content-Type: application/json, Authorization: Bearer ${config.llm.apiKey}, }, body: JSON.stringify({ model, messages: [{ role: user, content: prompt }], }), signal: AbortSignal.timeout(config.llm.timeoutMs), }); if (!res.ok) throw new Error(LLM 调用失败: ${res.status}); const data await res.json(); return data.choices[0].message.content as string; } }4. 验证请求队列消费、依赖替换、失败重试三个动作配置写完不算完要能验证。下面三个动作我每次重构后都会跑一遍。动作一验证队列消费。先启动 Redis 和 worker再投一个任务# 终端 A启动 worker npx ts-node src/workers/report.worker.ts # 终端 B投递任务并查询状态 npx ts-node -e const { addReportJob, reportQueue } require(./src/queues/report.queue); (async () { const id await addReportJob({ reportId: r1, projectId: p1, format: pdf, userId: u1 }); console.log(jobId , id); const job await reportQueue.getJob(id); console.log(state , await job.getState()); process.exit(0); })(); 预期输出jobId 1、state waiting或activeworker 终端会打印进度到 100 并返回filePath。如果 state 一直是waiting说明 worker 没连上同一个 Redis。动作二验证依赖替换。把ProjectRepository换成内存实现跑单元测试确认 Service 不依赖真实数据库// tests/unit/di-swap.test.ts import { ProjectService } from ../../src/services/project.service; class InMemoryRepo { private store: any[] []; async create(d: any) { const p { id: p${this.store.length 1}, ...d }; this.store.push(p); return p; } async findByUserId(uid: string) { return this.store.filter(p p.userId uid); } async findById(id: string) { return this.store.find(p p.id id) ?? null; } } test(依赖替换后无需数据库, async () { const svc new ProjectService(new InMemoryRepo() as any, { send: async () {} }); const p await svc.createProject({ name: X, userId: u1 }); expect(p.name).toBe(X); });跑npx jest tests/unit/di-swap.test.ts全绿就说明解耦到位。动作三验证失败重试。在处理器里临时抛错观察 BullMQ 是否按指数退避重试三次// 临时改 report.worker.ts 的处理器 if (job.attemptsMade 2) { throw new Error(模拟临时故障); }投递任务后看 worker 日志应该出现三次失败记录间隔约 1s、2s、4s第三次成功后任务状态变completed。验证完记得把这段删掉。5. 本篇常见错排查报错一EntityMetadataNotFoundError: No metadata for Project was found。原因是AppDataSource的entities数组没包含实体或者用了synchronize: false但没跑 migration。检查data-source.ts里entities: [Project]是否写全然后执行npx typeorm migration:run -d src/config/data-source.ts。报错二Connection is closed或 worker 启动即退出。BullMQ 的connection对象如果传了空字符串密码会报错。改成password: config.redis.password || undefined别传。另外 worker 是长驻进程别用ts-node -e跑要用独立文件。报错三任务重复执行。通常是removeOnComplete: false加上任务 ID 没去重。给addReportJob加jobId参数用业务 ID 做幂等键await reportQueue.add(generate, data, { jobId: report:${data.reportId} });报错四TypeORM 查询报column user_id does not exist。实体里用了Column({ name: user_id })但 migration 里建的是userId。两边命名必须一致建议统一用 snake_case 建表、实体里显式name映射。报错五TaoToken 调用返回 401。检查config.toml里api_key的${TAOTOKEN_API_KEY}是否被正确展开以及.env是否被加载。用node -e console.log(process.env.TAOTOKEN_API_KEY)确认环境变量存在。如果 Key 没问题去接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 核对请求头格式。6. 下一步把重构固化成习惯拆到这里一个 Skill 的骨架就清楚了接口层定义契约Service 层只认接口Repository 和 LlmClient 作为可替换实现耗时任务全部走 BullMQ。我自己的经验是每次新增一个功能前先问一句这个依赖能不能在测试里换掉如果答案是否定的就先抽接口再写逻辑比事后重构省力得多。如果你还在用散落的 Key 调模型建议先把LlmClient和 TaoToken 配置落地再动队列。Key 在控制台 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 创建接入细节看文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。长期跑编码或 Agent 任务的话Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 比按量更划算。下一步可以试着把报表任务拆成取数—汇总—导出三个子任务串成 Flow失败时只重跑失败的那一段。