@openreachtech/mentsu-agent-loop-renchan-job

@openreachtech/mentsu-agent-loop-core の Agent Loop（任意の Runnable）を、 renchan-job / BullMQ の Worker 内でジョブとして実行する Redis アダプタ です。 キュー・ワーカープール・Redis・リトライ・グレースフルシャットダウン・ログ等の重い処理は renchan-job に全委譲し、本パッケージは橋渡しだけを担います。

目次

• 概要
• 技術仕様
• 利用方法
• ユースケース
• サンプルコード
• 関連パッケージ
• ライセンス
• 開発者
• 著作権

概要

コアの BaseAgentLoop（やパイプライン・合成エージェント）は in-process で動きます。 本番で「長時間ループ・水平スケール・リトライ／レート制御・進捗のリアルタイム配信」が必要になったとき、本アダプタを足すと アクションとループのコードを一切変えずに Redis ジョブ実行へ載せ替えられます。

本アダプタは renchan-job の BaseJobWorker / BaseJobDispatcher を extends するだけ の薄い 3 クラスで構成され、次の橋渡しを既定実装しています。

• executeJob → コアの AgentLoopCtor.create({ context }).run({ input, onProgress })
• onProgress(event) → job.updateProgress(event)（BullMQ の progress イベント化）
• ジョブのライフサイクル（progress / completed / failed）→ コアの AgentTopic で組み立てたチャンネルへ publish
• BullMQ Worker への concurrency / limiter 転送（並列実行数・レート制限の制御）

アプリが書くのは「どのループを」「どの context で」「どの channel で」だけです。

依存方向は一方向（アダプタ → コア）。コアは renchan-job を知りません。

技術仕様

ランタイム / 配布

| 項目 | 値 | | :-- | :-- | | モジュール形式 | ESM ("type": "module") | | Node | ≥ 18 | | エントリ | lib/index.js（barrel） |

peerDependencies

| パッケージ | 範囲 | 備考 | | :-- | :-- | :-- | | @openreachtech/mentsu-agent-loop-core | ^1.0.0 | コア（Loop / Runner / AgentTopic を参照） | | @openreachtech/renchan-job-bullmq | ^1.1.1 | 継承元の Worker / Dispatcher 基底 | | bullmq | ^5 | renchan-job 経由（optional 扱い） |

dependencies ではなく peerDependencies で持ちます（二重インストール＝dual package hazard 回避）。 アプリが既に renchan-job を使っている場合、その 1 コピーを共有でき、追加インストールは不要です。

公開クラス一覧

BaseAgentJobWorker（抽象 / extends renchan-job BaseJobWorker）

ジョブ処理の中で agent-loop を run() する中核。

| メンバー | 区分 | 説明 | | :-- | :-- | :-- | | static get AgentLoopCtor() | 抽象 | 走らせる Runnable（Loop / Composite / Pipeline）の Ctor | | get channel() | 抽象 | 進捗 publish チャンネル名（Subscription resolver と共有） | | createAgentContext({ context, parcel }) | 抽象 | renchan-job context → agent-loop 用 context（DI 口）を構築 | | static get AgentTopicCtor() | フック | 既定 AgentTopic。extends で命名規則を差し替え可 | | buildScope({ jobModel }) | フック | チャンネルを名前空間化する scope（既定 null。ユーザー単位等にする場合 override） | | get subscriptionBroker() | フック | 既定 this.engine.subscriptionBroker（デメテル順守の委譲 getter） | | async executeJob({ body, context, parcel }) | 最終 | AgentLoopCtor.run({ input: body, onProgress })。onProgress を job.updateProgress に束ねる | | buildOptionHash() | 最終 | renchan-job 基底を override し concurrency / limiter を BullMQ Worker へ転送（基底は connection のみ渡すため必須） | | buildTopic({ scope }) | フック | AgentTopicCtor.create({ channel, scope }).value | | publishProgress({ topic, event }) | フック | 既定 this.subscriptionBroker.publish(topic, event) | | onJobProgress / onJobCompleted / onJobFailed / onWorkerError | フック | ライフサイクルコールバック（既定で publish ＋ ログ） |

BaseAgentJobDispatcher（抽象 / extends renchan-job BaseJobDispatcher）

enqueue 側。renchan-job の dispatchJob({ body, optionHash }) を薄くラップ。

| メンバー | 区分 | 説明 | | :-- | :-- | :-- | | static get EngineCtor() | 抽象 | renchan-job Engine（アプリ定義） | | static get ManifestCtor() | 抽象 | 入力スキーマ＋jobName | | async dispatchAgentJob({ input, optionHash }) | 具象 | エージェント語彙のエイリアス。dispatchJob({ body: input, optionHash }) に委譲 |

dispatchJob / createAsync / teardown は renchan-job 基底をそのまま継承（再実装不要）。

JobAgentRunner（具象 / extends コア BaseAgentRunner）

実行モード切替（要件③）の Redis 実装。enqueue して { accepted, jobId } を即返す。

| メンバー | 区分 | 説明 | | :-- | :-- | :-- | | static create({ DispatcherCtor }) | 具象 | ファクトリ | | async request({ input }) | 具象 | DispatcherCtor.createAsync() → dispatchJob({ body: input }) → teardown() → { accepted, jobId } |

進捗の流れ

loop.onProgress(event)
  → job.updateProgress(event)          （BaseAgentJobWorker が束ねる）
  → BullMQ progress イベント
  → onJobProgress
  → AgentTopic.create({ channel, scope }).value のチャンネルへ publishProgress
  → （GraphQL 等の）Subscription 配信

publish 側（Worker の channel）と subscribe 側（resolver の channel）は、コアの AgentTopic を共有するためチャンネル名が構造的に一致します。

並列実行数の制御

| 層 | 制御内容 | 設定場所 | | :-- | :-- | :-- | | A. Worker 内同時実行 | 1 Worker が同時に走らせる Agent Loop 数 | additionalConfig.concurrency（Worker） | | B. レート制限 | 単位時間あたりの処理数上限 | additionalConfig.limiter = { max, duration }（Worker） | | C. 水平スケール | Worker プロセス／レプリカ数 | デーモン起動数（運用） |

⚠️ renchan-job の基底 buildOptionHash() は connection しか BullMQ に渡しません。 本アダプタが buildOptionHash() を override して concurrency / limiter を転送することで初めて有効になります（アプリは additionalConfig に書くだけ）。

利用方法

インストール

npm install @openreachtech/mentsu-agent-loop-renchan-job
# peer（アプリに無ければ）
npm install @openreachtech/mentsu-agent-loop-core @openreachtech/renchan-job-bullmq bullmq

構成

コアの アクション・ループはそのまま流用 し、以下の薄い配線クラスを足します。

1. renchan-job の Engine / Share / Context（アプリで 1 組。既存があれば再利用）
2. Manifest（入力スキーマ＝ループ入力）
3. Worker（BaseAgentJobWorker を継承）
4. Dispatcher（BaseAgentJobDispatcher を継承）
5. デーモン起動 ＋ enqueue

既存の renchan-job 環境への相乗り: 本アダプタは BaseJobWorker/BaseJobDispatcher を extends するだけなので、既存の Engine / Share / Daemon にそのまま乗ります。注意点は 2 つ — renchan-job を peer で共有して二重インストールを避けること、jobName を既存ジョブと重複させないこと（例 agentLoop/videoSearch）。

ユースケース

• 本番の長時間ループ — タイムアウトしやすい同期 API ではなく、ジョブ化して非同期に回す。
• 水平スケール — Worker プロセス／レプリカを増やしてスループットを上げる（層 C）。
• リトライ / レート制御 — optionHash.attempts / backoff でジョブ再試行、limiter で AI API のレート上限を吸収。
• 進捗のリアルタイム配信 — Worker から AgentTopic チャンネルへ publish し、GraphQL Subscription 等でフロントに流す。
• 既存ジョブ基盤への相乗り — 既存の renchan-job アプリに、エージェントジョブを追加するだけ。
• 実行モード切替 — JobAgentRunner（Redis）と コアの InlineAgentRunner（in-process）を env で差し替え、呼び出し側はモードを意識しない。

サンプルコード

題材は「動画 AI 検索エージェント」。コアの VideoSearchAgentLoop をジョブ実行します。

1. Engine / Share / Context（renchan-job・アプリで 1 組）

// app/jobs/AppJobEngine.js
import BaseJobEngine from '@openreachtech/renchan-job-bullmq/lib/BaseJobEngine.js'

import env from '../globals/env.js'
import rootPath from '../globals/root-path.js'

import AppJobShare from './contexts/AppJobShare.js'
import AppJobContext from './contexts/AppJobContext.js'

/**
 * @extends {BaseJobEngine}
 */
export default class AppJobEngine extends BaseJobEngine {
  /** @override */
  static get config () {
    return {
      workersPath: rootPath.to('app/jobs'),
      redisConfig: {
        host: env.REDIS_HOST,
        port: Number(env.REDIS_PORT),
      },
    }
  }

  /** @override */
  static get ShareCtor () {
    return AppJobShare
  }

  /** @override */
  static get ContextCtor () {
    return AppJobContext
  }

  /** @override */
  static get standardErrorCodeHash () {
    return {
      Unknown: '100.X000.001',
      InvalidRequest: '103.X000.001',
    }
  }

  // 委譲getter（デメテル順守）。Worker から this.engine.subscriptionBroker を 1 ホップで参照可に。
  /** @override */
  get subscriptionBroker () {
    return this.share.subscriptionBroker
  }
}

// app/jobs/contexts/AppJobShare.js
import BaseJobShare from '@openreachtech/renchan-job-bullmq/lib/contexts/BaseJobShare.js'

import elasticsearchClient from '../../clients/elasticsearchClient.js'
import aiAgent from '../../clients/aiAgent.js'
import subscriptionBroker from '../../clients/subscriptionBroker.js'

/**
 * デーモン全体で共有するリソースの注入点。
 *
 * @extends {BaseJobShare}
 */
export default class AppJobShare extends BaseJobShare {
  /** @override */
  get elasticsearchClient () {
    return elasticsearchClient
  }

  /** @override */
  get aiAgent () {
    return aiAgent
  }

  /** @override */
  get subscriptionBroker () {
    return subscriptionBroker
  }
}

// app/jobs/contexts/AppJobContext.js
import BaseJobContext from '@openreachtech/renchan-job-bullmq/lib/contexts/BaseJobContext.js'

/**
 * @extends {BaseJobContext}
 */
export default class AppJobContext extends BaseJobContext {
  // 委譲getter（デメテル順守）。Worker の createAgentContext が context.elasticsearchClient を 1 ホップで参照可に。
  /** @override */
  get elasticsearchClient () {
    return this.share.elasticsearchClient
  }

  /** @override */
  get aiAgent () {
    return this.share.aiAgent
  }
}

2. Manifest（入力スキーマ ＝ ループ入力）

// app/jobs/videoSearch/VideoSearchJobManifest.js
import BaseJobManifest from '@openreachtech/renchan-job-bullmq/lib/BaseJobManifest.js'

import { ScalarHash } from '@openreachtech/mentsu-schema'

const {
  Text,
  Integer,
} = ScalarHash

/**
 * @extends {BaseJobManifest}
 */
export default class VideoSearchJobManifest extends BaseJobManifest {
  /** @override */
  static get jobName () {
    return 'videoSearch'
  }

  /** @override */
  static get bodySchema () {
    return {
      keyword: Text,
      userId: Integer,
    }
  }
}

3. Worker（アダプタ継承 ＝ ジョブ内でループを実行）

// app/jobs/videoSearch/VideoSearchJobWorker.js
import { BaseAgentJobWorker } from '@openreachtech/mentsu-agent-loop-renchan-job'

import VideoSearchAgentLoop from '../../agents/videoSearch/VideoSearchAgentLoop.js'
import VideoSearchJobManifest from './VideoSearchJobManifest.js'

/**
 * @extends {BaseAgentJobWorker}
 */
export default class VideoSearchJobWorker extends BaseAgentJobWorker {
  /** @override */
  static get ManifestCtor () {
    return VideoSearchJobManifest
  }

  /** @override */
  static get AgentLoopCtor () {
    return VideoSearchAgentLoop
  }

  /** @override */
  static get additionalConfig () {
    return {
      concurrency: 4, // ← 1 Worker が同時に走らせる Agent Loop 数（アダプタが BullMQ へ転送）
      limiter: { // ← レート制限（AI API 上限対策・任意）
        max: 10,
        duration: 1000,
      },
    }
  }

  /**
   * renchan-job の context から agent-loop 用 context（DI口）を構築。
   *
   * @override
   */
  createAgentContext ({
    context,
  }) {
    return {
      // context が委譲getterを公開しているので 1 ホップ（デメテル順守）
      elasticsearchClient: context.elasticsearchClient,
      aiAgent: context.aiAgent,
    }
  }

  /** @override */
  get channel () {
    return 'videoSearchProgress' // ← Subscription resolver と同じ
  }
}

executeJob 本体（loop.run ＋ onProgress → job.updateProgress）も、publishProgress（既定 this.subscriptionBroker.publish(...)）も、subscriptionBroker（既定 this.engine.subscriptionBroker）もアダプタ基底が実装済みです。

4. Dispatcher（enqueue 側）

// app/jobs/AppBaseJobDispatcher.js
import { BaseAgentJobDispatcher } from '@openreachtech/mentsu-agent-loop-renchan-job'

import AppJobEngine from './AppJobEngine.js'

/**
 * アプリ共通の Dispatcher 基底（EngineCtor を束ねる）。
 *
 * @extends {BaseAgentJobDispatcher}
 */
export default class AppBaseJobDispatcher extends BaseAgentJobDispatcher {
  /** @override */
  static get EngineCtor () {
    return AppJobEngine
  }
}

// app/jobs/videoSearch/VideoSearchJobDispatcher.js
import AppBaseJobDispatcher from '../AppBaseJobDispatcher.js'

import VideoSearchJobManifest from './VideoSearchJobManifest.js'

/**
 * @extends {AppBaseJobDispatcher}
 */
export default class VideoSearchJobDispatcher extends AppBaseJobDispatcher {
  /** @override */
  static get ManifestCtor () {
    return VideoSearchJobManifest
  }
}

5. デーモン起動 ＋ ジョブ投入

// scripts/start-daemon.js
import JobWorkersDaemon from '@openreachtech/renchan-job-bullmq/lib/JobWorkersDaemon.js'
import Timber from '@openreachtech/renchan-job-bullmq/lib/tools/Timber.js'

import AppJobEngine from '../app/jobs/AppJobEngine.js'

JobWorkersDaemon.createAsync({
  EngineCtor: AppJobEngine,
})
  .then(daemon => daemon.startDaemon())
  .then(workers => {
    Timber.log(`Started with ${workers.length} worker(s).`)
  })
  .catch(error => {
    Timber.error('Failed to start:', error.message)

    process.exit(1)
  })

// app/jobs/videoSearch/enqueueVideoSearch.js
import VideoSearchJobDispatcher from './VideoSearchJobDispatcher.js'

/**
 * 動画検索ジョブを投入する。
 *
 * @param {{ keyword: string, userId: number }} params
 */
export async function enqueueVideoSearch ({
  keyword,
  userId,
}) {
  const dispatcher = await VideoSearchJobDispatcher.createAsync()

  const response = await dispatcher.dispatchJob({
    body: { keyword, userId },
    optionHash: {
      attempts: 3, // ← QC（ジョブ軸: リトライ回数。ループの maxIterations とは別物）
      backoff: {
        type: 'exponential',
        delay: 1000,
      },
    },
  })

  await dispatcher.teardown()

  return {
    accepted: response.hasResponse(),
    jobId: response.idKey,
  }
}

6. 実行モード切替（in-process ⇔ Redis）

呼び出し側を Runner にのみ依存させ、env で実体を差し替えます。アクション・ループは一切変更しません。

// app/agents/videoSearch/runner.js
import { InlineAgentRunner } from '@openreachtech/mentsu-agent-loop-core'
import { JobAgentRunner } from '@openreachtech/mentsu-agent-loop-renchan-job'

import VideoSearchAgentLoop from './VideoSearchAgentLoop.js'
import VideoSearchJobDispatcher from '../../jobs/videoSearch/VideoSearchJobDispatcher.js'

import env from '../../globals/env.js'
import agentContext from './agentContext.js'

const inlineAgentRunner = InlineAgentRunner.create({
  AgentLoopCtor: VideoSearchAgentLoop,
  context: agentContext,
})

const jobAgentRunner = JobAgentRunner.create({
  DispatcherCtor: VideoSearchJobDispatcher,
})

export const videoSearchRunner = env.USE_REDIS
  ? jobAgentRunner    // Redis（renchan-job 経由・非同期）
  : inlineAgentRunner // in-process（Redis 不要・同期）

// 呼び出し側はモードを意識しない
const result = await videoSearchRunner.request({
  input: { keyword, userId },
  onProgress: event => { /* 非Redis では即時、Redis では Subscription 経由 */ },
})

| モード | request の戻り | 進捗 | 用途 | | :-- | :-- | :-- | :-- | | InlineAgentRunner（コア） | ループの最終結果 | in-process（onProgress 即時） | dev / test / CLI / Lambda / 同期 API | | JobAgentRunner（本パッケージ） | { accepted, jobId } | Subscription 経由 | 本番・長時間・水平スケール |

関連パッケージ

| パッケージ | 役割 | | :-- | :-- | | @openreachtech/mentsu-agent-loop-core | コア（Action / Loop / 合成 / Runner / AgentTopic）。本パッケージの peer | | @openreachtech/mentsu-agent-loop-renchan-job | 本パッケージ（Redis 実行アダプタ） | | @openreachtech/mentsu-agent-loop-graphql | GraphQL リゾルバ基底（進捗購読の subscribe 側）※実装予定 | | @openreachtech/renchan-job-bullmq | 継承元のジョブ基盤（Worker / Dispatcher / Engine）。本パッケージの peer |

ライセンス

本プロジェクトは Apache License 2.0 で公開されています。

詳細は LICENSE ファイル を参照してください。

開発者

Open Reach Tech Inc.

著作権

© 2026 Open Reach Tech Inc.