@cogitator-ai/ 工人(% 1)

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

安装

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

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

特征

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

快速启动

生产者:添加工作

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}`);

消费者:流程就业

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 班级管理创造就业和跟踪状况.

创建队列

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,
  },
});

队列配置

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,因此工人验证结构化的答案的方式与进程运行时一样. 使用工人不知道的钥匙的配置被拒绝,而不是在没有这种设置的情况下运行:

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 }

或者用手书写序列式:

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,超过共享的状态对象 节点以回合的形式运行,每个节点一次地运行,在节点全部落定之前,每个节点都有一个有多个根的图从所有节点开始.

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' });

节点输出存储在状态下 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 自己来的.

沼泽工作:

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' },
});

工作编号 :

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

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.

队列方法

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 处理带有可配置货币的工作.

创建工人池

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();

工人配置

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
}

持久代理工作

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

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

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

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 }) 当您自己调用处理器时,也会这样做。 见 持久运行。 。 。 .

工人活动

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;
}

集合方法

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 }:

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 消耗这些转折 :

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()。 。 。 .

任务结果

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

代理任务结果

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'继续以新的工作决定任何工人:

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

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

工作流程作业结果

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 任务结果

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。 。 。 .

普罗米修斯量子

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

曝光量表

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

长度直方图

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) 对于上一次尝试失败的每个工作(重复尝试不计算). 在池外处理工作时使用自己的收藏器 .

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 示例

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 配置

单节点

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

Redis 集群

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

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数字 :

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。 。 。 .

序列化的工作流程

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。 。 。 .

序列化 沼泽

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

实例

生产者/消费者

生产者(生产者:

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();

消费者(消费者:

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();

就业状况监测

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));
  }
}

优先处理

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 });

延迟任务

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 });

类型参考

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';

许可证

麻省理工学院