# @cogitator-ai/worker 中文文档

规范地址：https://npmcn.com/package/%40cogitator-ai/worker/

版本：0\.13\.9
npm 元数据更新时间：2026-10-29T00:50:36.839Z
中文译文更新时间：2026-11-10T10:17:18.432Z
软件包许可：MIT

本文由上游 README 翻译，可能存在偏差，请结合原文核对 API、命令与配置。版权与许可归原作者所有。

原始 npm 页面：https://www.npmjs.com/package/%40cogitator-ai/worker
原文对照：https://npmcn.com/package/%40cogitator-ai/worker/?lang=original
源码仓库：https://github.com/cogitator-ai/Cogitator-AI

## 关键信息（来自 npm 元数据）

- 包体积（解压后）：258 kB，59 个文件
- 模块格式：ES Module
- TypeScript 类型：包内自带
- Node.js 要求：\>=22\.12\.0
- 运行依赖：7 个
- peer 依赖：ioredis ^5\.0\.0 \|\| ^6\.0\.0
- 首次发布：2025-12-31T09:04:52.183Z
- 最新版本发布：2026-10-29T00:50:36.641Z
- 近 30 天下载量：9392（2026-10-11 至 2026-11-09，npm 官方下载统计）
- 已发布版本数：112（近 12 个月 112 个）

## 中文简介

为 Cogitator 代理执行分配的任务队列

## 安装

```sh
npm install @cogitator-ai/worker
```

也可以使用：`pnpm add @cogitator-ai/worker`、`yarn add @cogitator-ai/worker`、`bun add @cogitator-ai/worker`

## README

# @cogitator-ai/ 工人(% 1)

为 Cogitator 代理执行分配任务队列 。 为可靠可扩展的背景处理而建于BullMQ上.

## 安装

```bash
pnpm add @cogitator-ai/worker @cogitator-ai/core ioredis
```

布尔MQ, `@cogitator-ai/swarms` 财务报告和审定财务报表 `@cogitator-ai/workflows` 成为依附者, `ioredis` 是同级依赖。 网站文件 : [工人队列](https://cogitator.app/docs/deployment/worker-queues), (中文(简体) ), [分布的斯瓦尔姆](https://cogitator.app/docs/swarms/distributed)。 。 。 .

## 特征

-   **基于公牛MQ** - 与Redis进行可靠的工作处理
-   **工作类型** - 代理、工作流程图、轮廓和分布式轮廓
-   **工作时间** - 用你自己的工作来做 `Cogitator` (提供密钥、内存)和工具执行
-   **分布的斯瓦尔姆** - 说吧 `DistributedSwarmWorker` 执行代理 `@cogitator-ai/swarms`
-   **自动重试** - 对失败的工作进行指证后退,从最后的检查站开始重复工作流程,即耐用代理工作(`durable: true`从最后一步起
-   **工作编号** - 在 id 下添加一个工作,队列持有使该工作恢复,所以生产者可以重试添加
-   **使用量** - 工作流量和成群结果带有符号和成本,衡量标准按工作类型计算
-   **优先排队** - 首先处理重要工作
-   **延迟任务** - 以后执行的时间表
-   **普罗米修斯量子** - 内建的住房和财产管理局支助
-   **Redis 集群** - 生产可扩展性
-   **优雅的关机** - 在停止前等待积极的工作

* * *

## 快速启动

### 生产者:添加工作

```typescript
import { JobQueue } from '@cogitator-ai/worker';

const queue = new JobQueue({
  redis: { host: 'localhost', port: 6379 },
});

const agentConfig = {
  name: 'Assistant',
  instructions: 'You are a helpful assistant.',
  model: 'openai/gpt-6.1-sol',
  provider: 'openai' as const,
  tools: [],
};

const job = await queue.addAgentJob(agentConfig, 'Hello, world!', {
  threadId: 'user-123',
  priority: 1,
});

console.log(`Job added: ${job.id}`);
```

### 消费者:流程就业

```typescript
import { Cogitator } from '@cogitator-ai/core';
import { WorkerPool } from '@cogitator-ai/worker';

const pool = new WorkerPool({
  redis: { host: 'localhost', port: 6379 },
  concurrency: 5,
  workerCount: 2,
  cogitator: new Cogitator({
    llm: { providers: { openai: { apiKey: process.env.OPENAI_API_KEY! } } },
  }),
  tools: [searchTool], // implementations for tools referenced by serialized agents
});

await pool.start();
```

串行代理参考工具名. 工人解决他们的问题 `tools`;其代理人需要工人不提供的工具的工作失败 `Tools not registered on this worker: <names>`。 。 。 .

* * *

## 任务队列

这个 `JobQueue` 班级管理创造就业和跟踪状况.

### 创建队列

```typescript
import { JobQueue } from '@cogitator-ai/worker';

const queue = new JobQueue({
  name: 'my-queue',
  redis: {
    host: 'localhost',
    port: 6379,
    password: 'secret',
  },
  defaultJobOptions: {
    attempts: 3,
    backoff: { type: 'exponential', delay: 1000 },
    removeOnComplete: 100,
    removeOnFail: 500,
  },
});
```

### 队列配置

```typescript
interface QueueConfig {
  name?: string; // Default: 'cogitator-jobs'
  redis: {
    url?: string; // 'redis://user:password@host:port/db', 'rediss://' for TLS
    host?: string; // Default: 'localhost'
    port?: number; // Default: 6379
    username?: string;
    password?: string;
    db?: number;
    tls?: boolean; // implied by a rediss:// url
    cluster?: {
      nodes: { host: string; port: number }[];
    };
  };
  defaultJobOptions?: {
    attempts?: number; // Default: 3
    backoff?: {
      type: 'exponential' | 'fixed';
      delay: number; // Delay in ms
    };
    removeOnComplete?: boolean | number; // Default: 100
    removeOnFail?: boolean | number; // Default: 500
  };
}
```

`redis` 这是 `RedisConnectionConfig` 联 合 国 `@cogitator-ai/types`,同样的配置 `DistributedSwarmWorker` 散发的蜂群. `resolveRedisOptions(redis)` 给一个服务器的设置.

### 添加任务

**特工乔布斯:**

最简单的方法就是把已经拥有的特工序列化. `serializeAgent` 用代理电线格式写 `@cogitator-ai/core` (单位:千美元)`toAgentWire`),由代理,工作流程和群活以及分布式群活轮共享. 它保持了整个配置(模型和提供方,倒置模型,指令,采样,停止序列,推理努力,响应格式,迭代限制,超时),工具计划以及它可以交给的每一个代理,并将一个Zod响应计划转换为JSON Schema,因此工人验证结构化的答案的方式与进程运行时一样. 使用工人不知道的钥匙的配置被拒绝,而不是在没有这种设置的情况下运行:

```typescript
import { Agent } from '@cogitator-ai/core';
import { serializeAgent } from '@cogitator-ai/worker';
import { z } from 'zod';

const Verdict = z.object({ verdict: z.enum(['run', 'hold']), reason: z.string() });
const chief = new Agent({
  name: 'chief',
  model: 'openrouter/openai/gpt-6-luna',
  instructions: 'Decide whether the pitch runs.',
  reasoning: { effort: 'medium' },
  responseFormat: { type: 'json_schema', schema: Verdict },
});

const job = await queue.addAgentJob(serializeAgent(chief), 'A pitch about ...');
// later: (await queue.getJob(job.id))?.returnvalue.structured -> { verdict, reason }
```

或者用手书写序列式:

```typescript
const agentConfig: SerializedAgent = {
  name: 'Researcher',
  instructions: 'Research and summarize topics.',
  model: 'openai/gpt-6.1-sol', // or 'gpt-6.1-sol' - the provider is prepended when missing
  provider: 'openai',
  temperature: 0.7,
  maxTokens: 2048,
  maxIterations: 5,
  reasoning: { effort: 'low' }, // optional
  responseFormat: { type: 'json_schema', schema: { type: 'object', properties: {} } }, // optional, JSON Schema
  tools: [
    {
      name: 'search',
      description: 'Search the web',
      parameters: { type: 'object', properties: { query: { type: 'string' } } },
    },
  ],
};

const job = await queue.addAgentJob(agentConfig, 'Research quantum computing', {
  threadId: 'thread-123', // default: a new id per job
  userId: 'user-456', // the run's userId: owns the thread, reaches tools as context.userId
  priority: 1, // Lower = higher priority
  delay: 5000, // Delay 5 seconds
  metadata: { source: 'api' },
});
```

工人路线 `model` 和同一代理在进程中完全一样:一个前缀,命名一个内置提供商,在工人Cogitator的后端 `llm.backends` 或注册插件选择提供者,所以 `'openrouter/deepseek/deepseek-v4-pro'` 运行在一个 `openrouter` 无论是后端 `provider` 说. `provider` (可选的,任何工人路线的提供者,包括自定义后端和插件)只准备给一个没有前缀名的模型,例如 `'meta-llama/llama-4-scout'` 与 `provider: 'openrouter'`没有 `provider` 这种模式运行在工人身上 `llm.defaultProvider`,则a `provider` 劳动者无法通过途径无法完成任务. `serializeAgent` 发送带有明文的代理 `provider` 作为 `<provider>/<model>`,所以它采取同样的路线 在工人与在过程.

**工作流量:**

劳动力流动岗位 `WorkflowExecutor` 联 合 国 `@cogitator-ai/workflows`,超过共享的状态对象 节点以回合的形式运行,每个节点一次地运行,在节点全部落定之前,每个节点都有一个有多个根的图从所有节点开始.

```typescript
const workflowConfig: SerializedWorkflow = {
  id: 'triage',
  name: 'Ticket triage',
  nodes: [
    {
      id: 'classify',
      type: 'agent',
      config: {
        agentConfig: classifierAgent, // SerializedAgent; a json_schema agent writes its structured answer
        prompt: 'Classify this ticket as BUG or QUESTION: {{ticket}}',
        outputKey: 'category',
      },
    },
    { id: 'normalize', type: 'transform', config: { transform: 'trim', inputKey: 'category' } },
    {
      id: 'is-bug',
      type: 'condition',
      config: { key: 'normalize', operator: 'contains', value: 'BUG' },
    },
    {
      id: 'summary',
      type: 'transform',
      config: { transform: 'template', template: 'Bug report: {{ticket}}' },
    },
  ],
  edges: [
    { from: 'classify', to: 'normalize' },
    { from: 'normalize', to: 'is-bug' },
    { from: 'is-bug', to: 'summary', condition: 'true' },
  ],
};

await queue.addWorkflowJob(workflowConfig, { ticket: 'App crashes on login' });
```

| 节点类型 | 配置 |
| --- | --- |
| `agent` | `{ agentConfig, prompt?, outputKey? }` —— 说 `prompt` 支持 `{{path}}` 占位符; 默认提示为 JSON 状态 |
| `transform` | `{ transform, inputKey?, outputKey?, template? }` —— 说 `uppercase`, (中文(简体) ), `lowercase`, (中文(简体) ), `trim`, (中文(简体) ), `json-parse`, (中文(简体) ), `json-stringify`, (中文(简体) ), `template` |
| `condition` | `{ key, operator, value? }` —— 说 `equals`, (中文(简体) ), `not-equals`, (中文(简体) ), `contains`, (中文(简体) ), `exists`, (中文(简体) ), `gt`, (中文(简体) ), `lt`; 向外边缘使用 `condition: 'true' | 'false'` |
| `parallel` | `{}` ——扇出标记;后继者同时运行 |

节点输出存储在状态下 `outputKey` (默认:节点 id). 只能通过未获取条件分支到达的节点被跳过(`{ skipped: true }` 内 `nodeResults`。图在执行前被验证(未知节点、无效的配置、周期)。 运行将自己的账簿保存在状态密钥中( 节点运行, 分支一个条件, 每个代理节点的使用), 起始于 `__cogitator:`,这些被排除在外的工作 `output`输入密钥、节点 ID 和输出密钥 `__cogitator:` 拒绝.

工作流程中有一个检查点, 在每个节点后保存在工作ID和 `runId`因此,工作共享一个 `runId` 永不相续 工作 BullMQ 重试(`attempts`)从它继续,而不是再次运行其成品代理节点. 检查点在运行完成后被删除,一旦没有尝试会跟随失败:在最后一次尝试后,被取消的工作或被暂停的代理运行. 工作停止了关闭 它为接活的工人. `WorkerPool` 将Redis的检查站置于 `<queue prefix>:<queue name>:workflow-checkpoints` (单位:千美元)`cogitator`,或 `{cogitator}` 组合模式)用于 `checkpointTtl` 在最后的保存(默认为7天)之后的第二秒,所以一个运行的检查站,工人被撞倒的,就只能靠自己了. `checkpoints` 在工人配置中需要一个 `CheckpointStore` 自己来的.

**沼泽工作:**

```typescript
const swarmConfig: SerializedSwarm = {
  topology: 'voting',
  agents: [researcherConfig, writerConfig, editorConfig],
  coordinator: coordinatorConfig, // decides when no consensus is reached
  maxRounds: 3,
  consensusThreshold: 0.8,
};

await queue.addSwarmJob(swarmConfig, 'Write an article about AI', {
  priority: 1,
  metadata: { project: 'blog' },
});
```

| 地形学 | 沼泽战略 | 页:1 |
| --- | --- | --- |
| `sequential` | `pipeline` | 每个代理商一个阶段,顺序 |
| `hierarchical` | `hierarchical` | `coordinator` 需要成为主管 |
| `collaborative` | `pipeline` | 每个代理都参与其中 `maxRounds` 回合(默认 3),看迄今为止所有贡献, `coordinator` 结合它们,否则最后的帮助就是答案 |
| `debate` | `debate` | `maxRounds` 子弹, `coordinator` 温和派 |
| `voting` | `consensus` | `consensusThreshold`, (中文(简体) ), `maxRounds`, (中文(简体) ), `coordinator` 断开连接 |

**工作编号 :**

`addAgentJob`, (中文(简体) ), `resumeAgentJob`, (中文(简体) ), `addWorkflowJob`, (中文(简体) ), `addSwarmJob` 财务报告和审定财务报表 `addSwarmAgentJob` 带一个 `jobId`,该工作被添加在其下. 添加一个 ID 队列所持有的工作, 已经让该工作恢复, 而不是添加第二个, 因此一个制作人可以重试一个添加, 如超时后, 不运行两次 :

```typescript
const job = await queue.addWorkflowJob(workflowConfig, input, { jobId: `triage-${ticketId}` });
const again = await queue.addWorkflowJob(workflowConfig, input, { jobId: `triage-${ticketId}` });
// again.id === job.id, one job runs
```

两个加了同样的ID的赛车 都得到了队列所持有的工作。 队列已删除的工作( 过去) `removeOnComplete` 或者说 `removeOnFail`)已经消失,其id增加了一个新的工作. BullMQ 从 id 中构建 Redis 密钥, 所以一个空 id, 一个带有 `:`,一个整数和一个队列自己键的名称(`meta`, (中文(简体) ), `id`, (中文(简体) ), `events`, (中文(简体) ), `stalled-check` 诸如此类的人被拒绝,排队的id是另一种工作(一种代理工作) `addWorkflowJob`说)掷出. 无 `jobId` 每个添加都得到新的id.

### 队列方法

```typescript
const job = await queue.getJob('job-id');

const state: JobState = await queue.getJobState('job-id');
// 'waiting' | 'prioritized' | 'delayed' | 'active' | 'completed' | 'failed'
// | 'waiting-children' | 'unknown'

await queue.resumeAgentJob(agentConfig, pausedResult, { decisions }); // see Job Results
await queue.resumeAgentJob(agentConfig, askedResult, {
  responses: { [askedResult.interrupts![0].id]: 'eu' },
}); // a tool's question

const metrics = await queue.getMetrics();
// { waiting, active, completed, failed, delayed, depth, workerCount }

await queue.pause();
await queue.resume();

await queue.clean(60 * 60 * 1000, 1000, 'completed');
await queue.clean(24 * 60 * 60 * 1000, 100, 'failed');

const bullQueue = queue.getQueue();

await queue.close();
```

工作加一个 `priority` 等待在 BullMQ 中 `prioritized` 状态而不是 `waiting`。 。 。 . `getMetrics()` 数起来 `waiting` 财务报告和审定财务报表 `depth` 都一样,所以 `cogitator_queue_depth` 覆盖所有需要完成的工作. `waiting-children` 是 BullMQ 流量的母体(通过 `getQueue()`等待孩子,而不是像孩子一样被计算在内.

* * *

## 员工池

这个 `WorkerPool` 处理带有可配置货币的工作.

### 创建工人池

```typescript
import { WorkerPool } from '@cogitator-ai/worker';

const pool = new WorkerPool(
  {
    redis: { host: 'localhost', port: 6379 },
    workerCount: 2,
    concurrency: 5,
    lockDuration: 30000,
    stalledInterval: 30000,
  },
  {
    onJobStarted: (jobId, type) => {
      console.log(`Job ${jobId} (${type}) started`);
    },
    onJobCompleted: (jobId, result) => {
      console.log(`Job ${jobId} completed:`, result);
    },
    onJobFailed: (jobId, error) => {
      console.error(`Job ${jobId} failed:`, error);
    },
    onWorkerError: (error) => {
      console.error('Worker error:', error);
    },
  }
);

await pool.start();
```

### 工人配置

```typescript
interface WorkerConfig extends QueueConfig {
  workerCount?: number; // Default: 1
  concurrency?: number; // Default: 5
  lockDuration?: number; // Default: 30000ms
  stalledInterval?: number; // Default: 30000ms
  cogitator?: Cogitator; // Default: new Cogitator()
  tools?: Tool[]; // Tool implementations, resolved by name
  checkpoints?: CheckpointStore; // Where workflow jobs keep their checkpoints. Default: the pool's Redis
  checkpointTtl?: number; // Seconds the default store keeps a checkpoint after its last save. Default: 7 days
  durable?: boolean; // Agent jobs go on from their last step on the next attempt. Default: false
}
```

| 选项 | 默认 | 说明 |
| --- | --- | --- |
| `workerCount` | 1个 | 工人人数 |
| `concurrency` | 页:1 | 每名工人同时从事的工作 |
| `lockDuration` | 3万块 | 在认为工作停顿前锁定超时 |
| `stalledInterval` | 3万块 | 检查已停滞的工作的间隔 |

### 持久代理工作

与 `durable: true` 每个代理工作都是以该工作命名的持久运行(`job:<jobId>`:在每个模型转弯和工具批次后,它都会保留一个检查站. 当工人崩溃或游泳池关闭中产期时,下一次尝试(针对这名工人或另一名工人)会恢复跑道,然后从最后一步继续前进,而不是再次运行整个工作. 一个返回的工具调用保留了它的结果,从未运行两次,一个正在运行的调用到达模型为"结果未知,在重试前核实". 打开 `stop()` 财务报告和审定财务报表 `forceStop()` 终止其持久运行 `RUN_SUSPENDED`所以他们立刻交出 而不是等待租借 当BullMQ将一个被搁置的工作交给它时,当它仍在运行时,这个工作被拖延到运行的租约用完之后,而不是用完一次尝试.

```typescript
const pool = new WorkerPool({ redis, tools, durable: true });
```

所有工人必须共享商店。 泳池的自有相机(无) `cogitator` 给定)将运行在它的 Redis 中保留在队列的前缀下. 一个通过你需要共享商店 `enabled: false` 保持其他运行情况:

```typescript
import { Cogitator, RedisDurableRunStore } from '@cogitator-ai/core';

const cogitator = new Cogitator({
  durable: { store: new RedisDurableRunStore({ client: redisClient }), enabled: false },
});
const pool = new WorkerPool({ redis, cogitator, tools, durable: true });
```

`processAgentJob(payload, runtime, { durable: true })` 当您自己调用处理器时,也会这样做。 见 [持久运行](https://cogitator.app/docs/core/durable-runs)。 。 。 .

### 工人活动

```typescript
interface WorkerPoolEvents {
  onJobStarted?: (jobId: string, type: 'agent' | 'workflow' | 'swarm' | 'swarm-agent') => void;
  onJobCompleted?: (jobId: string, result: JobResult) => void;
  onJobFailed?: (jobId: string, error: Error) => void;
  onWorkerError?: (error: Error) => void;
}
```

### 集合方法

```typescript
await pool.start();

pool.isPoolRunning();

pool.getWorkerCount();

const metrics = await pool.getMetrics(await queue.getMetrics());

// Job duration histogram and per-type counters
pool.metrics.format(await pool.getMetrics(await queue.getMetrics()));

// Cancel a job this pool is running: its agent runs stop and the job fails without retries
pool.cancelJob(jobId, 'No longer needed');

// Graceful shutdown (waits up to 30s for active jobs, then aborts them and force-closes;
// BullMQ hands the aborted jobs to another worker once their locks expire)
await pool.stop(30000);

// Force shutdown (aborts the jobs still running)
await pool.forceStop();
```

每一个工作都得到BullMQ的中止信号, 下到每个代理运行,工作流程节点和swarm转弯. 处理者也拿它: `processAgentJob(payload, runtime, { signal })`,和关于 `processWorkflowJob`, (中文(简体) ), `processSwarmJob` 财务报告和审定财务报表 `executeSwarmAgentJob` (单位:千美元)`processSwarmAgentJob` 读取 `signal` (见a/cn.9/wg.iii/wp.iii/wp.1号文件).

* * *

## 工作处理器

内置处理器处理每个工作类型.

### 直接使用处理器

处理器采用工作有效载荷和可选运行时间(`{ cogitator, tools, checkpoints }`:

```typescript
import { processAgentJob, processWorkflowJob, processSwarmJob } from '@cogitator-ai/worker';

const runtime = { cogitator, tools: [searchTool] };

const agentResult = await processAgentJob(
  { type: 'agent', jobId: 'job-1', agentConfig: myAgentConfig, input: 'Hello!', threadId: 't-1' },
  runtime
);

const workflowResult = await processWorkflowJob(
  { type: 'workflow', jobId: 'job-2', runId: 'run-1', workflowConfig, input: { ticket: '...' } },
  runtime
);

const swarmResult = await processSwarmJob(
  { type: 'swarm', jobId: 'job-3', swarmConfig: mySwarmConfig, input: 'Solve this problem' },
  runtime
);
```

`processWorkflowJob` 除非跑步时间已经过去,否则就不用检查了 `checkpoints` 联合国 `CheckpointStore`) (中文(简体) ). 和他们一起,通过 `{ willRetry }` 作为第三个论点:运行失败之后又发生另一起尝试`willRetry(error)` 返回时 `true`)为它保留检查站,任何其他故障都会删除.

`processSwarmAgentJob(payload, { publisher, isFinalAttempt, ...runtime })` 执行一个分布式群转而将结果(与工作编号相标记)交给协调员 `deliverSwarmJobResult` 联 合 国 `@cogitator-ai/swarms`: 当任务名称 `stateKeys.result` 而出版社已经 `set`,结果保存在那里 `payload.resultTtl` ms(默认为一小时),然后发布到 `payload.stateKeys.results`。 。 。 . `publisher` 是一个 `SwarmResultPublisher`,现在的化名 `SwarmResultClient` (单位:千美元)`publish`选项 `set`) (中文(简体) ). `executeSwarmAgentJob` 返回不发布结果.

* * *

## 分布在沼泽地的工人

以 `distributed.enabled` (见 `@cogitator-ai/swarms`发送每个代理 转向一个雷迪斯队列. `DistributedSwarmWorker` 消耗这些转折 :

```typescript
import { Cogitator } from '@cogitator-ai/core';
import { DistributedSwarmWorker } from '@cogitator-ai/worker';

const worker = new DistributedSwarmWorker(
  {
    redis: { url: 'redis://localhost:6379' }, // the same Redis as the swarm's distributed.redis
    keyPrefix: 'swarm', // must match the swarm's distributed.redis.keyPrefix
    queue: 'swarm-agent-jobs', // must match distributed.queue
    concurrency: 4,
    cogitator: new Cogitator({ llm: { defaultModel: 'ollama/llama3.2' } }),
    tools: [searchTool],
  },
  {
    onJobCompleted: (job) => console.log('done', job.agentName),
    onJobFailed: (job, error) => console.error(job.agentName, error.message),
    onJobCancelled: (job) => console.log('the swarm gave up on', job.jobId),
    onJobsRecovered: (workerId, count) => console.warn(`${count} turns of ${workerId} requeued`),
    onError: (error) => console.error(error),
  }
);

await worker.start();
process.on('SIGTERM', () => void worker.stop()); // waits for in-flight turns
```

失败转弯被报告为错误, 所以它是自己的 `errorHandling` (重试、 故障、 跳过) 应用 。 鼠群放弃的转弯( 它运行已中止或超时, 鼠群关闭, 又一次尝试已回答) 被跳过, 或运行时中止( 检查每个 `cancelCheckInterval` 毫秒,默认为1000),并报告 `onJobCancelled` 而不是发表。 每个转弯运行在模型上 它的代理将使用在流程中,由工人的路由 `cogitator`所以给工人同样的 `llm.backends`,插件和提供者密钥作为运行 sloom 的过程.

`redis` 使用一个 `url` (单位:千美元)`rediss://` 用于 TLS , `host`, (中文(简体) ), `port`, (中文(简体) ), `username`, (中文(简体) ), `password`, (中文(简体) ), `db`, (中文(简体) ), `tls` 或者说 `cluster.nodes`就像排队一样 轮回坐在名单里 `<keyPrefix>:jobs:{<queue>}`,其支架是一个 Redis Cluster 散列标记,将队列及其工人键放在一个槽中。 使用的较早版本 `<keyPrefix>:jobs:<queue>`,因此升级运行群和工人的流程.

队列是可靠的 :

-   工人每转一转 `BLMOVE` (Redis 6.2 或更新) 从队列到自己的列表 `<queue key>:processing:<workerId>`,然后在转弯完成后取出.
-   每个 `heartbeatInterval` ms( 默认 5000) 设置心跳键 `<queue key>:heartbeat:<workerId>`,3个间隔后失效,并保留在组合中 `<queue key>:workers`。 。 。 .
-   打开 `start()` 每拍一拍,一个工人会移动工人的转弯,他们的心跳会回到排队的头部,把工人从场上扔下来,然后报告 `onJobsRecovered(workerId, count)`。 。 。 .
-   工人还检查自己每节节拍的列表:一个转动,工人不会连续跑出两次节拍,例如一次A `BLMOVE` 移动时,一个被丢弃的连接失去了它的答复,回到队列的头部,并用工人自己的身份报告.
-   工人默默地保持了3个间隔,这算不上是消失了,所以一个被停滞的工人可以跑两次.
-   结果是保留在工作之下 `stateKeys.result` 页:1 `resultTtl` 在发布前,协调员在留言被遗漏后发现,一个工人跳过一个结果在那里的工作(例如,一个死去的工人在删除其名单之前回答)。 工人无法控制的结果仍在公布.
-   `stop()` 等待在飞行中转动, 手拿回它清单上剩下的东西, 离开集合并删除它的心跳. `getWorkerId()` {\\fn黑体\\fs22\\bord1\\shad0\\3ahbe\\4ah00\\fscx67\\fscy66\\2chffffff\\3ch808080}给工人们新的身份 `start()`。 。 。 .

* * *

## 任务结果

每个工作类型都返回一个特定的结果结构.

### 代理任务结果

```typescript
interface AgentJobResult {
  type: 'agent';
  output: string;
  structured?: unknown; // the validated answer of a json_schema agent, when it fits
  reasoning?: string; // the reasoning summary, when the agent asked for one
  usage: {
    inputTokens: number;
    outputTokens: number;
    totalTokens: number;
    cost: number; // USD, as the run reported it
    reasoningTokens?: number;
    cachedInputTokens?: number;
  };
  toolCalls: {
    name: string;
    input: unknown;
    output: unknown;
  }[];
  /** 'paused' when tool calls wait for approval: output is then not the answer */
  status?: 'completed' | 'paused';
  pendingApprovals?: ToolApprovalRequest[]; // the calls a paused run waits on
  checkpoint?: RunCheckpoint; // what a paused run continues from, see resumeAgentJob
  /** @deprecated use usage */
  tokenUsage?: {
    prompt: number;
    completion: number;
    total: number;
  };
}
```

为工具批准而暂停的运行将完成它的工作 `status: 'paused'`继续以新的工作决定任何工人:

```typescript
if (result.status === 'paused') {
  await queue.resumeAgentJob(agentConfig, result, {
    decisions: { [result.pendingApprovals![0].toolCallId]: { approved: true } },
  });
}
```

工作流和成群的工作不能等待决定:当其中一名代理人暂停工作时,工作失败而不与一个代理人重复工作 `AgentRunPausedError`。 。 。 .

### 工作流程作业结果

```typescript
interface WorkflowJobResult {
  type: 'workflow';
  output: Record<string, unknown>;
  nodeResults: Record<string, unknown>;
  usage: JobUsage; // what the agent nodes used and cost
  duration: number;
}

interface JobUsage {
  inputTokens?: number; // absent where the runs reported only their totals
  outputTokens?: number;
  totalTokens: number;
  cost: number; // USD
  runs: number; // agent runs counted
}
```

### Swarm 任务结果

```typescript
interface SwarmJobResult {
  type: 'swarm';
  output: string;
  rounds: number;
  // what the swarm's agent runs used and cost, in all and by agent
  usage: JobUsage & { byAgent: Record<string, { tokens: number; cost: number; runs: number }> };
  agentOutputs: {
    agent: string;
    output: string;
  }[];
}
```

一群人只报了总的征兆 所以 `usage` 没有 `inputTokens` 或者说 `outputTokens`。 。 。 .

* * *

## 普罗米修斯量子

用于监测的内置度量衡和碳氢化合物.

### 曝光量表

```typescript
import { JobQueue, WorkerPool } from '@cogitator-ai/worker';
import express from 'express';

const queue = new JobQueue({ redis: { host: 'localhost', port: 6379 } });
const pool = new WorkerPool({ redis: { host: 'localhost', port: 6379 } });
await pool.start();

const app = express();

app.get('/metrics', async (req, res) => {
  const queueMetrics = await queue.getMetrics(); // workerCount = workers connected to the queue
  res.type('text/plain').send(pool.metrics.format(queueMetrics));
});

app.listen(9090);
```

### 可用计量

| 度量衡 | 类型 | 说明 |
| --- | --- | --- |
| `cogitator_queue_depth` | 测量 | 等待工作总数+延迟工作 |
| `cogitator_queue_waiting` | 测量 | 准备工作,包括优先工作 |
| `cogitator_queue_active` | 测量 | 正在处理的工作 |
| `cogitator_queue_completed` | 测量 | 保留在 Redis 的已完成的工作(封顶日期: `removeOnComplete`(中文(简体) ) |
| `cogitator_queue_failed` | 测量 | 保留在 Redis 中的工作失败( 封顶由 `removeOnFail`(中文(简体) ) |
| `cogitator_queue_delayed` | 测量 | 排定/延迟的工作 |
| `cogitator_workers_total` | 测量 | 连接到队列的工人 |
| `cogitator_job_duration_seconds` | 直方图 | 工作处理时间 |
| `cogitator_jobs_by_type_total` | 计数器 | 按类别分列的工作 |
| `cogitator_jobs_failed_total` | 计数器 | 上次尝试失败的工作,通过 `type` |
| `cogitator_job_tokens_total` | 计数器 | 将所用已完成工作的代理运行按 `type` |
| `cogitator_job_input_tokens_total` | 计数器 | 已完成工作的输入符号,按 `type` |
| `cogitator_job_output_tokens_total` | 计数器 | 已完成工作的产出符号,按 `type` |
| `cogitator_job_cost_usd_total` | 计数器 | 已完成工作费用的代理业务,按 `type` |

`cogitator_queue_completed` 财务报告和审定财务报表 `cogitator_queue_failed` 可以去作为 BullMQ 裁剪旧工作, 所以提醒 `cogitator_jobs_failed_total` 改为(例如 `increase(cogitator_jobs_failed_total[5m]) > 5`) (中文(简体) ). 每个类型的计数器都出现在第一种工作之后. 使用计数器 `usage` 用于完成的代理、工作流程、成群和成群代理工作、输入和输出符号,但只当工作分开报告时(成群工作不分开).

### 长度直方图

```typescript
import { DurationHistogram } from '@cogitator-ai/worker';

const histogram = new DurationHistogram('my_duration_seconds', 'Custom duration tracking');

histogram.observe(0.5);
histogram.observe(1.2);
histogram.observe(0.8);

console.log(histogram.format({ queue: 'main' }));

histogram.reset();
```

### 计量收集器

`WorkerPool` 留一个进去 `pool.metrics`: 它召唤 `recordJob(type, durationMs)` 财务报告和审定财务报表 `recordUsage(type, result.usage)` 每一个已完成的工作和 `recordFailure(type)` 对于上一次尝试失败的每个工作(重复尝试不计算). 在池外处理工作时使用自己的收藏器 .

```typescript
import { MetricsCollector } from '@cogitator-ai/worker';

const collector = new MetricsCollector();

collector.recordJob('agent', 1500);
collector.recordJob('workflow', 3200);
collector.recordUsage('workflow', workflowResult.usage); // { inputTokens?, outputTokens?, totalTokens, cost }
collector.recordFailure('agent');

const output = collector.format(queueMetrics, { queue: 'main' });
```

### Kubernetes HPA 示例

```yaml
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: cogitator-workers
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: cogitator-workers
  minReplicas: 1
  maxReplicas: 10
  metrics:
    - type: External
      external:
        metric:
          name: cogitator_queue_depth
        target:
          type: AverageValue
          averageValue: 10
```

* * *

## Redis 配置

### 单节点

```typescript
const queue = new JobQueue({
  redis: {
    host: 'localhost',
    port: 6379,
    password: 'secret',
  },
});
```

### Redis 集群

队列和工人连接到所有集群节点(键使用 `{cogitator}` 散列标签让他们生活在一个位置:

```typescript
const queue = new JobQueue({
  redis: {
    cluster: {
      nodes: [
        { host: 'redis-1', port: 6379 },
        { host: 'redis-2', port: 6379 },
        { host: 'redis-3', port: 6379 },
      ],
    },
    password: 'secret',
  },
});
```

* * *

## 序列类型

Jobs使用可以存储到Redis的序列化配置.

### 序列化 探员

`SerializedAgent` 实值 `AgentWireConfig` 从 `@cogitator-ai/types`数字 :

```typescript
interface SerializedAgent {
  id?: string;
  name: string;
  description?: string;
  instructions: string;
  model: string; // routed like in-process: a known provider prefix picks the provider
  provider?: LLMBackendProvider; // used when model names no provider the worker routes to
  temperature?: number;
  topP?: number;
  maxTokens?: number;
  stopSequences?: string[];
  responseFormat?:
    { type: 'text' } | { type: 'json' } | { type: 'json_schema'; schema: JSONSchema };
  reasoning?: ReasoningConfig;
  maxIterations?: number;
  onIterationLimit?: 'answer' | 'stop';
  timeout?: number;
  tools: ToolSchema[]; // resolved by name against the worker's tools
  handoffs?: { agent: string; toolName?: string; description?: string }[];
  handoffAgents?: Record<string, Omit<SerializedAgent, 'handoffAgents'>>; // handoff targets by name
}
```

拒绝未知的密钥 。 代理工作成果(`AgentJobResult`携带 `output`, (中文(简体) ), `structured`, (中文(简体) ), `structuredError`, (中文(简体) ), `reasoning`, (中文(简体) ), `usage` (电话铃声, `cost`, (中文(简体) ), `duration`、推理和缓存符号, `toolCalls` 及其产出,以及 `truncated`, (中文(简体) ), `blocked` 财务报告和审定财务报表 `iterationLimitReached`。 。 。 .

### 序列化的工作流程

```typescript
interface SerializedWorkflow {
  id: string;
  name: string;
  nodes: SerializedWorkflowNode[];
  edges: SerializedWorkflowEdge[];
}

interface SerializedWorkflowNode {
  id: string;
  type: 'agent' | 'transform' | 'condition' | 'parallel';
  config: Record<string, unknown>;
}

interface SerializedWorkflowEdge {
  from: string;
  to: string;
  condition?: string; // 'true' | 'false' for edges leaving condition nodes
}
```

节点配置类型为 `AgentNodeConfig`, (中文(简体) ), `TransformNodeConfig` 财务报告和审定财务报表 `ConditionNodeConfig`。 。 。 .

### 序列化 沼泽

```typescript
interface SerializedSwarm {
  topology: 'sequential' | 'hierarchical' | 'collaborative' | 'debate' | 'voting';
  agents: SerializedAgent[];
  coordinator?: SerializedAgent;
  maxRounds?: number;
  consensusThreshold?: number;
}
```

* * *

## 实例

### 生产者/消费者

**生产者(生产者:**

```typescript
import { JobQueue } from '@cogitator-ai/worker';

const queue = new JobQueue({
  redis: { host: 'localhost', port: 6379 },
});

async function main() {
  const agentConfig = {
    name: 'Summarizer',
    instructions: 'Summarize the given text concisely.',
    model: 'openai/gpt-6.1-sol',
    provider: 'openai' as const,
    tools: [],
  };

  const texts = [
    'Long article about technology...',
    'Research paper on climate change...',
    'News story about economics...',
  ];

  for (const text of texts) {
    const job = await queue.addAgentJob(agentConfig, text, {
      priority: 1,
    });
    console.log(`Queued job: ${job.id}`);
  }

  await queue.close();
}

main();
```

**消费者(消费者:**

```typescript
import { WorkerPool } from '@cogitator-ai/worker';

const pool = new WorkerPool(
  {
    redis: { host: 'localhost', port: 6379 },
    concurrency: 5,
  },
  {
    onJobStarted: (id, type) => console.log(`Starting ${type} job: ${id}`),
    onJobCompleted: (id, result) => console.log(`Completed: ${id}`, result),
    onJobFailed: (id, error) => console.error(`Failed: ${id}`, error),
  }
);

async function main() {
  await pool.start();
  console.log('Worker pool started');

  process.on('SIGTERM', async () => {
    console.log('Shutting down...');
    await pool.stop(30000);
    process.exit(0);
  });
}

main();
```

### 就业状况监测

```typescript
import { JobQueue } from '@cogitator-ai/worker';

const queue = new JobQueue({
  redis: { host: 'localhost', port: 6379 },
});

async function monitorJob(jobId: string) {
  let lastState = '';

  while (true) {
    const state = await queue.getJobState(jobId);

    if (state !== lastState) {
      console.log(`Job ${jobId}: ${state}`);
      lastState = state;
    }

    if (state === 'completed' || state === 'failed') {
      const job = await queue.getJob(jobId);
      if (job) {
        console.log('Result:', await job.returnvalue);
      }
      break;
    }

    await new Promise((r) => setTimeout(r, 1000));
  }
}
```

### 优先处理

```typescript
await queue.addAgentJob(config, 'Low priority', { priority: 10 });
await queue.addAgentJob(config, 'Medium priority', { priority: 5 });
await queue.addAgentJob(config, 'High priority', { priority: 1 });
await queue.addAgentJob(config, 'Critical', { priority: 0 });
```

### 延迟任务

```typescript
await queue.addAgentJob(config, 'Run in 5 seconds', { delay: 5000 });
await queue.addAgentJob(config, 'Run in 1 minute', { delay: 60000 });
await queue.addAgentJob(config, 'Run in 1 hour', { delay: 3600000 });
```

* * *

## 类型参考

```typescript
import type {
  // Serialized configs
  SerializedAgent,
  SerializedWorkflow,
  SerializedWorkflowNode,
  SerializedWorkflowEdge,
  AgentNodeConfig,
  TransformNodeConfig,
  ConditionNodeConfig,
  SerializedSwarm,

  // Job payloads
  JobPayload,
  AgentJobPayload,
  WorkflowJobPayload,
  SwarmJobPayload,
  SwarmAgentJobPayload,

  // Job results
  JobResult,
  AgentJobResult,
  WorkflowJobResult,
  SwarmJobResult,
  JobUsage,
  SwarmAgentJobResult,

  // Configuration
  QueueConfig,
  WorkerConfig,
  WorkerRuntime,
  QueueMetrics,
  JobState,
  DistributedSwarmWorkerConfig,
  DistributedSwarmWorkerEvents,
} from '@cogitator-ai/worker';
```

* * *

## 许可证

麻省理工学院

## 常见问题

### @cogitator-ai/worker 是什么？

为 Cogitator 代理执行分配的任务队列

### 如何安装 @cogitator-ai/worker？

在项目目录运行 npm install @cogitator-ai/worker，或使用 pnpm add / yarn add / bun add。它还需要宿主项目提供 peer 依赖：ioredis（^5\.0\.0 \|\| ^6\.0\.0）。npm 7 及以上与 pnpm 8 及以上默认会自动安装必需的 peer 依赖，Yarn 需要手动安装。

### @cogitator-ai/worker 支持 TypeScript 吗？

支持。包内自带 TypeScript 类型声明（package\.json 声明了 types 入口），安装后即可获得类型提示，无需再安装 @types 包。

### @cogitator-ai/worker 支持 ESM 还是 CommonJS？

以 ES Module 形式发布，推荐使用 import 引入。在 CommonJS 项目中，Node\.js 20\.19\+ / 22\.12\+ 可以直接 require\(\) 不含顶层 await 的 ES Module，更早的版本需要改用动态 import\(\)。

### @cogitator-ai/worker 需要什么版本的 Node\.js？

package\.json 的 engines 字段要求 Node\.js \>=22\.12\.0。版本不符时 npm 默认只给出警告，开启 engine-strict 后会拒绝安装。

### @cogitator-ai/worker 可以商用吗？

@cogitator-ai/worker 使用 MIT 许可。宽松许可：允许商业使用、修改和再分发，通常只需在分发时保留原版权与许可声明。以上为概括，不构成法律意见，请以许可原文为准。

### @cogitator-ai/worker 的下载量有多少？

根据 npm 官方下载统计，@cogitator-ai/worker 在 2026/10/11 – 2026/11/09 这 30 天内被下载 9,392 次，日均约 313 次。下载量会把 CI 构建、镜像同步和作为其他包依赖的安装一并计入，适合横向比较热度，不能直接等同于使用人数。

### @cogitator-ai/worker 还在维护吗？

最新版本 0\.13\.9 发布于 2026/10/29（12 天前）；至今共发布 112 个版本，其中过去 12 个月发布了 112 个；首次发布于 2025/12/31。发布频率只是参考信号：功能稳定的工具可能很少需要更新，建议结合源码仓库的 issue 和提交记录判断。