@cogitator-ai/ 表达

Express.js 为Cogitator AI运行时间的服务器适配器. 自动生成 REST API 端点用于代理,工作流程,并有 SSE 流出, WebSocket 支持,和 Swagger 文档.

安装

npm install @cogitator-ai/express express
# or
pnpm add @cogitator-ai/express express

快速启动

import express from 'express';
import { Cogitator, Agent, tool } from '@cogitator-ai/core';
import { CogitatorServer } from '@cogitator-ai/express';

const app = express();

const cogitator = new Cogitator({
  llm: {
    defaultModel: 'openai/gpt-6-luna',
    providers: { openai: { apiKey: process.env.OPENAI_API_KEY! } },
  },
});

const chatAgent = new Agent({
  name: 'chat',
  description: 'General-purpose chat assistant',
  instructions: 'You are a helpful assistant.',
  model: 'openai/gpt-6-luna',
});

const server = new CogitatorServer({
  app,
  cogitator,
  agents: { chat: chatAgent },
  config: {
    basePath: '/api',
    enableSwagger: true,
  },
});

await server.init();
app.listen(3000, () => console.log('Server running on http://localhost:3000'));

自动生成的终点

探员

GET    /api/agents                         - List all agents
POST   /api/agents/:name/run               - Run agent (JSON response)
POST   /api/agents/:name/stream            - Run agent (SSE stream)
POST   /api/agents/:name/resume            - Resume a run paused for tool approvals
POST   /api/agents/:name/resume/stream     - Resume a paused run (SSE stream)
POST   /api/agents/:name/agui              - Run or resume for an AG-UI client (AG-UI SSE)

运行/ 流体 : { input: string; context?: object; threadId?: string; multitaskStrategy?: 'enqueue' | 'reject' | 'interrupt' | 'rollback' }。 。 。 . input 必须包含多于白空间和 threadId,发送时必须是非空字符串(400 INVALID_INPUT 否则,在模型被调用之前). 验证符来自 @cogitator-ai/server-shared因此,快活,快活,Hono和Koa拒绝完全相同的尸体。 这个 usage 运行中答案带 inputTokens, (中文(简体) ), outputTokens 财务报告和审定财务报表 totalTokens附加时 reasoningTokens, (中文(简体) ), cachedInputTokens 财务报告和审定财务报表 cacheWriteTokens 当模型报告它们时,形状与其他每个适配器相同. 这个 finish 一个代理流带有相同的 usage (无论运行成本或持续时间). 特工名单曝光 config.description绝不是特工的指示 认证 userId (从) auth)被传递到运行,当客户端断开时运行被中止.

• 运行发送 context 该模型读取 system 线程消息作为操作员指令,所以客户端可以不设置,除非服务器允许: acceptContext 列出 context 密钥客户端可能发送( e)true 接受任何密钥,对于像您自己的代码一样信任的客户端),并 threadMessageRoles 联合国 POST /threads/:id/messages (单位:千美元)user 财务报告和审定财务报表 assistant (默认情况下). 其他的都拒绝 400 INVALID_INPUT在HTTP和WebSocket上一样.
• A 类 POST 身体不是JSON的(Atext/plain拒绝使用 415 UNSUPPORTED_MEDIA_TYPE 在解析前 : 浏览器在不使用 CORS 预飞的情况下发送源代码 .
• POST /agents/:name/run 财务报告和审定财务报表 /resume 回复 toAgentRunResponse() 联 合 国 @cogitator-ai/server-shared数字 : output, (中文(简体) ), threadId, (中文(简体) ), usage, (中文(简体) ), toolCalls (单位:千美元){ id, name, arguments }, (中文(简体) ), status 财务报告和审定财务报表 traceId附加时 reasoning, (中文(简体) ), pendingApprovals, (中文(简体) ), structured, (中文(简体) ), structuredError, (中文(简体) ), truncated, (中文(简体) ), blocked 财务报告和审定财务报表 iterationLimitReached 当它们适用。 网页 complete 事件带有相同的对象,从不提示系统,历史,痕量跨度或暂停运行的检查点.
• 这个 start 代理流命名运行线索的事件(请求的) threadId,或一个新的),以及 finish 以运行结果重复它,所以一个发送无线的客户端继续与这个客户端的对话.
• 一旦运行结束,沼泽路线就会关闭,因此分布的沼泽不会留下雷地斯的连接,其状态会过期. GET /swarms 列出每个代理人、路由器和管道阶段.

线索( 记忆)

GET    /api/threads/:id               - Get thread messages, a page at a time with ?limit&cursor
POST   /api/threads/:id/messages      - Add message to thread
DELETE /api/threads/:id               - Delete thread

路线 cogitator.getMemory(),它连接了首先使用的已配置的内存适配器,因此在任何代理运行之前,线程可以在新服务器上被读取. 仅a Cogitator 无 memory 配置的答案 503 UNAVAILABLE。需要新信息 role (单位:千美元)user 或者说 assistant,则与 threadMessageRoles和一个非空字符串 content选项 metadata 存储在内存条目上.

GET /threads/:id 默认回答整个线程 。 与 ?limit=50 它回答了最新50条信息(最老的首页)和a nextCursor,以及 ?limit=50&cursor=<nextCursor> 在他们面前的50个,直到一个答复没有 nextCursor。用同样毫秒写出的信件从不跳过或跨行重复。 A 类 limit 1至 1000 之外或服务器从未给予的光标被拒绝 400 INVALID_INPUT。 。 。 .

let cursor: string | undefined;
do {
  const query = new URLSearchParams({ limit: '50', ...(cursor && { cursor }) });
  const page = await (await fetch(`/api/threads/${threadId}?${query}`)).json();
  render(page.messages);
  cursor = page.nextCursor;
} while (cursor);

工作流量

GET    /api/workflows                 - List all workflows
POST   /api/workflows/:name/run            - Run workflow
POST   /api/workflows/:name/stream         - Stream workflow events
POST   /api/workflows/:name/resume         - Resume a workflow from a checkpoint
POST   /api/workflows/:name/resume/stream  - Resume a workflow and stream its events

车身 : { input?: object; options?: { maxConcurrency?: number; maxIterations?: number; checkpoint?: boolean } }。 。 。 . maxConcurrency编号 :maxIterations 必须是正整数(另作400个)。 与 limits.workflowMaxIterations 或者说 limits.workflowMaxConcurrency 设置时,一个较大的值被维持到极限,一个要求无一物的请求则得到极限。 其他选择被放弃,以及 checkpoint: true 除非服务器已经 workflowCheckpoints (见 检查站和恢复) (中文(简体) ). 一个失败的工作流程会回答它失败的错误, 像其它运行错误一样映射( 点击的节点) LLM_RATE_LIMITED 答题 429 与 Retry-After,内部错误 500 INTERNAL_ERROR),其溪流以 error 带有该代码的事件代替 workflow_completed。断开连接会中止工作流程.

溪流传出 { type: 'workflow', event, data } 事件: node_started, (中文(简体) ), node_completed, (中文(简体) ), node_error, (中文(简体) ), node_progress 终于 workflow_completed (单位:千美元){ workflowId, duration, checkpointId? }),或者,当运行失败时, workflow_failed (单位:千美元){ checkpointId? }紧接着 error 活动.

检查站和恢复

config.workflowCheckpoints 使用任意 CheckpointStore 联 合 国 @cogitator-ai/workflows,例如 PostgresCheckpointStore 或者说 RedisCheckpointStore带着它,跑步 options.checkpoint: true 把它的检查站作为呼叫者 userId 那个 auth 和答复 workflow_completed 最后一个事件为 checkpointId。 。 。 .

POST /workflows/:name/resume 带 { checkpointId: string; options?: { maxConcurrency?: number; maxIterations?: number } },通过完成的检查站的节点并运行其余部分。 答案是说 /run,以及 /resume/stream 溪流类 /stream编辑 limits 等一会 /run,而恢复运行时会不断节省检查站,以便再次恢复运行。 缺少、属于另一个用户或属于另一个工作流程答案的检查点 404 CHECKPOINT_NOT_FOUND 之前什么运行,运行 /resume/stream 流开前. 无 workflowCheckpoints 每一个回复 404 CHECKPOINT_NOT_FOUND。失败的运行会用它保存的最后一个检查点来回答它的错误 checkpointId旁边 code 财务报告和审定财务报表 message (单位:千美元){ error: { message, code, checkpointId } }),且故障流发送 workflow_failed 事件为 { checkpointId } 紧接着 error ,所以客户端从那里恢复失败运行 。 复职报告再次没有将检查站命名为要恢复的检查站,这是它自己的最后一个检查站,或者,当它没有拯救任何检查站时,就是它开始的检查站.

import { PostgresCheckpointStore } from '@cogitator-ai/workflows';

const server = new CogitatorServer({
  app,
  cogitator,
  workflows: { shipping },
  config: { auth, workflowCheckpoints: new PostgresCheckpointStore({ client: pool }) },
});

斯瓦尔斯

GET    /api/swarms                    - List all swarms
POST   /api/swarms/:name/run          - Run swarm
POST   /api/swarms/:name/stream       - Stream swarm events
GET    /api/swarms/:name/blackboard   - Get configured blackboard sections

运行/ 流体 : { input: string; context?: object; threadId?: string; timeout?: number } (单位:千美元)timeout 必须是积极的,最多 2147483647 (中文(简体) ). 与 limits.swarmTimeout 设定,会保持更长的超时 和运行,没有人得到它, 和一团流出的时间答案 504 RUN_TIMEOUT。断开连接会中止群。 溪流传出 { type: 'swarm', event, data } 活动(agent_start, (中文(简体) ), agent_complete, (中文(简体) ), agent_error, (中文(简体) ), message最终 swarm_completed 产出和使用情况。 每次运行都会产生出一团新鲜,因此黑板端点返回所配置的初始区段.

工具文档( D)

GET    /api/tools                     - List all tools
GET    /api/health                    - Health check ({ status, uptime, timestamp })
GET    /api/ready                     - Readiness probe ({ status, checks })
GET    /api/docs                      - Swagger UI
GET    /api/openapi.json              - OpenAPI spec

配置

const server = new CogitatorServer({
  app,
  cogitator,
  agents: { chat: chatAgent, research: researchAgent },
  workflows: { 'code-review': codeReviewWorkflow },
  swarms: { 'dev-team': devTeamSwarm },
  config: {
    basePath: '/cogitator',
    enableWebSocket: true,
    enableSwagger: true,

    // Authentication — throw to reject (401); the returned context is available
    // as req.cogitator.auth and its userId is passed to agent runs.
    // Also applied to WebSocket upgrade requests.
    auth: async (req) => {
      const token = req.headers.authorization?.replace('Bearer ', '');
      const user = await validateToken(token);
      return { userId: user.id, roles: user.roles };
    },

    // Rate limiting (in-memory, per process). windowMs must be positive and max
    // non-negative, otherwise the constructor throws. Requests without a client
    // address share a single bucket.
    rateLimit: {
      windowMs: 60000, // 1 minute
      max: 100, // 100 requests per window
      trustProxy: false, // true: key by the nearest X-Forwarded-For hop
    },

    // CORS — credentials default to true for explicit origins and to false for '*'.
    // `origin: '*'` with `credentials: true` reflects any origin with credentials;
    // only do that for trusted, non-cookie setups.
    cors: {
      origin: ['https://myapp.com'],
      credentials: true,
    },

    // Swagger customization
    swagger: {
      title: 'My AI API',
      description: 'AI-powered API endpoints',
      version: '1.0.0',
    },

    // Caps on what a request may ask a run for. A larger value is held to the cap,
    // a request without one gets the cap. Checked when the server is created.
    limits: { swarmTimeout: 120_000, workflowMaxIterations: 50, workflowMaxConcurrency: 4 },

    // 'deny' answers 401 to callers without a userId. Health, ready and the docs stay open.
    anonymous: 'deny',

    // Called with the auth context and the thread of every request that names one.
    // false answers 403 THREAD_FORBIDDEN and no run starts.
    authorizeThread: (auth, threadId) => threadId.startsWith(`${auth?.userId}:`),

    // Dependencies for the tools of HTTP agent runs, streams and resumes (context.deps).
    // Built from the request on the server, never read from the body.
    deps: async (req) => ({ db: await tenantDb(req.get('x-tenant')) }),

    // WebSocket runs build theirs from the upgrade request, a plain Node IncomingMessage.
    websocket: { deps: async (req) => ({ db: await tenantDb(req.headers['x-tenant']) }) },
  },
});

多个用户

何时 auth 返回 a userId呼叫者用线条做的每件事都被设定范围:

• 代理运行和流 (REST 和 WebSocket) 通过 userId 改为 cogitator.run(),所以运行创建的线程属于该用户,并且是一个 threadId 由他人拥有的被拒绝.
• GET /threads/:id, (中文(简体) ), POST /threads/:id/messages 财务报告和审定财务报表 DELETE /threads/:id 先检查线的主人; POST 到未知的id 创建了呼叫者拥有的线程.
• 另一个用户的线索答案 403 有代码 THREAD_ACCESS_DENIED,其信息既不返回也不更改.
• 线索创建较早, 没有所有者( 无) userId仅对无 userId,例如没有 auth 已配置 .
• authorizeThread(auth, threadId) 在所有权之上增加你自己的规则,用于共享或团队线程. 它运行在每一条线条命名的路线上(代理跑道,流道和恢复道,群跑道和流道, GET, (中文(简体) ), DELETE 财务报告和审定财务报表 POST /threads/:id/messages)和在WebSocket上运行 与一个 threadId,在尸体验证后,在核心检查所有者之前。 答复 false 拒绝使用 403 THREAD_FORBIDDEN,没有运行开始.
• anonymous: 'deny' 拒绝每个来电者 auth 给予无 userId 与 401 UNAUTHORIZED,包括了WebSocket的升级. /health, (中文(简体) ), /ready, (中文(简体) ), /openapi.json 财务报告和审定财务报表 /docs 保持开放。 基础路径答案下的未知路径 401 所以匿名电话员无法查清有哪条路 404 NOT_FOUND 一旦打电话者有 userId默认, 'allow',让匿名的来电者像以前一样进入,以及未知路径的答案 404 NOT_FOUND 在错误信封中.

限制和道具

limits 上限客户端可能要求运行的: swarmTimeout (单位:百万美元), workflowMaxIterations 财务报告和审定财务报表 workflowMaxConcurrency。从不因一个限制而拒绝请求:它得到的要求和限制的比HTTP和SSE都小,或者在它要求无时得到的限制。 OpenAPI spec显示每个限制作为其域的默认值. 每个限制必须是正整数,或者构造器所抛出.

deps: (req) => deps 将依赖关系代理工具构建为 context.deps (数据库控件、租户、客户端) HTTP 运行、流出或恢复 deps。它们被建在服务器上,用于每个请求,从不来自身体。 钩子所抛出的错误,在流开始前如跑出错误一样被回答.

WebSocket 运行并恢复,从 websocket.deps: (req) => deps,在每次运行时都随升级请求一起发出。 这个要求是一个简单的节点 IncomingMessage 从未通过快递,所以它已经 headers, (中文(简体) ), url 财务报告和审定财务报表 socket 但是没有 req.get() 中间的器件也没有添加任何内容 这就是为什么它得到自己的钩子。 无 websocket.deps WebSocket的运行没有道具. 它所抛出的错误( 或一个) authorizeThread 答案 { type: 'error', id, error } 并开始不跑.

核准

一个工具与 requiresApproval (单位:千美元)true,或参数的函数)在转弯执行前暂停运行:

1. POST /api/agents/:name/run 答案为 status: 'paused' 财务报告和审定财务报表 pendingApprovals (单位:千美元){ toolCallId, toolName, arguments, description, sideEffects? }; 已完成运行 status: 'completed'。运行的检查点停留在服务器上,从不发送给客户端.
2. POST /api/agents/:name/stream 发射 { type: 'approval-required', threadId, approvals } 刚刚 finish。 。 。 .
3. 客户端显示待决的电话,并将答案发送给 POST /api/agents/:name/resume数字 :

await fetch('/api/agents/support/resume', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({
    threadId,
    decisions: { [toolCallId]: { approved: true } },
    defaultDecision: { approved: false, reason: 'Not approved' },
  }),
});

躯体是 { threadId, decisions?, defaultDecision? },在决定是 { approved: boolean, reason?: string, updatedArguments?: object, remember?: boolean | { rule?: string, layer?: string } } (见 权限); 畸形身体获得 400 INVALID_INPUT。反应的形状与 /run 并可能再次暂停 电话没有决定。 运行恢复为认证 userId: 无暂停的线索运行答案 409 RUN_NOT_PAUSED,另一个用户暂停运行 403 THREAD_ACCESS_DENIED通过 WebSocket, 发送 { type: 'resume', id, payload: { name, threadId, decisions?, defaultDecision? } }; 暂停运行 complete 事件携带 status 财务报告和审定财务报表 pendingApprovals。 。 。 .

工具也可以向用户提问 context.interrupt(),它暂停运行的方式相同。 答案随即 interrupts (单位:千美元){ id, kind, reason, message, responseSchema?, expiresAt?, toolCallId?, toolName?, details? }, (中文(简体) ), kind 一个 input, (中文(简体) ), plan 财务报告和审定财务报表 budget),溪流发出 { type: 'interrupt-required', threadId, interrupts } 改为 approval-required,客户端则通过中断id恢复回答: { threadId, responses: { [id]: answer } },在HTTP和WebSocket上同样使用(payload: { name, threadId, responses }) (中文(简体) ). 与其 responseSchema 拒绝使用 400 INTERRUPT_ANSWER_INVALID后发送 expiresAt 与 410 INTERRUPT_EXPIRED (单位:千美元)/resume/stream 两者都作为 error 与此相关的事件 code) (中文(简体) ).

运行或流体也可以命名为 multitaskStrategy 对于已经运行在这个 Cogitator 中的线程 : enqueue 等待运行, reject 拒绝新运行 THREAD_BUSY, (中文(简体) ), interrupt 停止运行( 它失败) RUN_INTERRUPTED和 rollback 停止它和从它开始的树枝. POST /agents/:name/run 以 409。一条流已回答 200 发送 start 当它的运行遇到线条的其他运行, 所以它得到它们作为一个 error 事件代替( e){ type: 'error', message, code: 'THREAD_BUSY' }),和一个WebSocket run 作为 error 边框。 没有一只,就是猎人 multitaskStrategy ,并拒绝使用 400 INVALID_INPUT。网络安全套 run 载荷也带走了 沼泽路线没有线条队列并忽略 multitaskStrategy。 。 。 .

POST /agents/:name/resume/stream 带着同样的身体和溪流 其他跑步一样 /stream。 。 。 . start 财务报告和审定财务报表 finish 将恢复的线程命名,在溪流打开前拒绝的一具尸体、代理或线被回答为JSON,就像 /resume。没有暂停运行的线程以一个 error 事件, code: 'RUN_NOT_PAUSED',而断开连接会中止恢复.

SSE 流线

这个 /agents/:name/stream 端点返回 Server- Sent 事件 :

// Client-side
const response = await fetch('/api/agents/chat/stream', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({ input: 'Hello!' }),
});

const reader = response.body!.getReader();
const decoder = new TextDecoder();

while (true) {
  const { done, value } = await reader.read();
  if (done) break;

  const lines = decoder.decode(value).split('\n');
  for (const line of lines) {
    if (line.startsWith('data: ') && line !== 'data: [DONE]') {
      const event = JSON.parse(line.slice(6));
      console.log(event);
      // { type: 'text-delta', id: '...', delta: 'Hello' }
    }
  }
}

当SSE流打开时,它写着 : keep-alive 注释每 sseHeartbeatMs (默认5 s, 0 关闭)),由SSE客户端跳过,因此一个代理或负载平衡器不会关闭在慢工具或模型上等待的流(nginx关闭连接静默60 s).

活动 : start, (中文(简体) ), text-start编号 :text-delta编号 :text-end, (中文(简体) ), tool-call-start编号 :tool-call-delta编号 :tool-call-end (与模型的工具叫id), tool-result (单位:千美元)toolCallId 与呼叫相匹配, approval-required, (中文(简体) ), interrupt-required, (中文(简体) ), error (单位:千美元){ message, code }),以及 finish 和运行的 usage (与JSON 答案相同的对象),然后 data: [DONE]。 。 。 .

当代理设置 reasoning: { summary: true } 并且提供者返回一个推理摘要, 它作为自己的部分流出 : reasoning-start, (中文(简体) ), reasoning-delta (单位:千美元){ id, delta }, (中文(简体) ), reasoning-end。一个推理部分总是在文本或工具调用开始前关闭,因此推理部分和文本部分永远不会互通.

可恢复流

与 resumableStreams 一个客户端丢失了 /agents/:name/stream 或者说 /agents/:name/resume/stream (a flashy network,一个去睡觉的标签) 得到它错过的东西, 运行在它离开的时候进行:

const server = new CogitatorServer({
  app,
  cogitator,
  agents: { chat: chatAgent },
  config: {
    resumableStreams: true,
  },
});

• 每个事件都带有SSE id, (中文(简体) ), <streamId>:<seq> 与 seq 从1起。 心跳的评论没有.
• 一个断开连接的客户端不再停止运行. 它的事件不断进入商店.
• 同样的路线 Last-Event-ID: <id> 重新连接而不是启动运行, 并且身体没有被读取: 答案用其 id 重放之后的事件, 然后跟踪运行到 finish (和 data: [DONE])或(或) error. 若干客户可能遵循一个流程. ?lastEventId=<id> 对无法设置头部的客户端也起作用,但它将ID放入了URL和访问日志,因此更倾向于头部.
• DELETE /agents/:name/stream/:streamId 取消运行( e)204:它的溪流以一个 error 事件 STREAM_CANCELLED。溪流 id 是事件 ID 在上次事件之前的一部分 :。 。 。 .
• 一条溪流与发起它的呼唤者相连: auth 返回(用户代号、角色、权限、元数据,例如租户),所以没有用户代号的呼叫者也被分开。 给 owner: (auth) => key 将其绑入更少的字段,例如当身份带有符号过期时,仅用户的ID。 任何人,未知或已过期的溪流和事件 ID该溪流从来没有发送得到 404 STREAM_NOT_FOUND 在任何 SSE 标题之前, 代码不是 <streamId>:<seq> 获得 400 INVALID_INPUT。 。 。 . authorizeThread 每次重新连接和取消时, 都会询问溪流的线程( t)403 THREAD_FORBIDDEN) (中文(简体) ).
• 不再保存的事件后重新连接( 多于 maxEvents 从此以后 error 事件为 STREAM_RESUME_GAP:取出线程(GET /threads/:id)来追. 一个读取缓慢的客户端最多只能持有 maxEvents 事件和追赶 从商店.
• 跑步没人跟着 detachedTimeoutMs (默认为5分钟)被停止,所以一个再也没有回来的客户不会运行成本上升,每一次运行在 until 中止( 通过信号, 您的服务器停止, 并提前中止) server.close()等待着溪流的开通。 这样的流以 error 事件 STREAM_ABORTED。 。 。 .
• 一条奔跑的溪流永远不会被赶走: maxStreams 溪流都继续,一个新的 拒绝与 503 STREAM_LIMIT_REACHED。 。 。 .

true 将事件保存在进程记忆中:每条溪流有1000个事件,一个完成的溪流5分钟,1000个溪流一次(maxEvents, (中文(简体) ), retentionMs, (中文(简体) ), maxStreams) (中文(简体) ). 负载平衡器后面有多个进程,通过 a store 共享(StreamReplayStore 从 @cogitator-ai/server-shared, 由 Redis 或 类似 的 pub/ sub 支持), 因此重新连接或取消, 在另一个进程上着陆到达了溪流及其运行. 默认关闭 : 没有 ID, 而客户端中断运行 .

WebSocket 支持

启用实时双向通信( 需要可选 ws 软件包 :

import { createServer } from 'http';

const server = new CogitatorServer({
  // ...
  config: {
    enableWebSocket: true,
    websocket: { path: '/api/ws', pingInterval: 30000, maxPayloadSize: 1024 * 1024 },
  },
});

await server.init();

const httpServer = createServer(app);
await server.attachWebSocket(httpServer); // defaults to `${basePath}/ws`
httpServer.listen(3000);

attachWebSocket 丢出,如果 enableWebSocket 关闭, init() 未运行,或 ws 未安装。 配置 auth 函数运行在升级请求上(在扔出时以401被拒绝,或当它给出不时) userId 下级 anonymous: 'deny'A级 run 或者说 resume 在线条上 authorizeThread 拒绝答案 { type: 'error', id, error } 没有开始,和 websocket.deps 与升级请求一起调用。 对于定制设置, setupWebSocket(httpServer, routeContext, config) 也被导出.

议定书:

只有代理运行在WebSocket上得到支持,使用HTTP端点进行工作流程和群集. 一个新的 run 从同一连接中止上一个连接。 运行失败报告 a CogitatorError'消息,任何其他错误都会被记录并发送为 Internal server error。服务器按 pingInterval 和终止客户 错过了一磅.

订阅者只得到运行时的事件 userId 由召来者发起 authorizeThread 设置, 当 authorizeThread(subscriberAuth, threadId) 允许在运行的线条上进行 threadId 或为它而创造的一个核心。 答案是每个用户一次被问到并运行,一个用户拒绝,或丢弃,没有获得运行事件. 无 authorizeThread,没有 userId 分享彼此运行的事件, 所以设置 authorizeThread 或者说 anonymous: 'deny' 匿名者不得见面.

客户端使用量 :

const ws = new WebSocket('ws://localhost:3000/api/ws');

ws.onmessage = (event) => {
  const message = JSON.parse(event.data);
  console.log(message);
};

// Run agent
ws.send(
  JSON.stringify({
    type: 'run',
    id: 'req-1',
    payload: {
      type: 'agent',
      name: 'chat',
      input: 'Hello!',
    },
  })
);

自定义中端软件

使用内置的中件工厂或创建自己的:

import {
  createAuthMiddleware,
  createRateLimitMiddleware,
  createCorsMiddleware,
  errorHandler,
} from '@cogitator-ai/express';

// Use individually
app.use(
  '/api',
  createAuthMiddleware(async (req) => {
    // Custom auth logic
  })
);

app.use(
  '/api',
  createRateLimitMiddleware({
    windowMs: 60000,
    max: 100,
    keyGenerator: (req) => req.headers['x-api-key'] as string,
    trustProxy: false, // set true if behind a trusted reverse proxy
  })
);

自定义路由

创建有流支持的自定义路由 :

import { Router } from 'express';
import { ExpressStreamWriter, setupSSEHeaders, generateId } from '@cogitator-ai/express';

const router = Router();

router.post('/custom/stream', async (req, res) => {
  setupSSEHeaders(res);
  const writer = new ExpressStreamWriter(res);
  const messageId = generateId('msg');
  const textId = generateId('txt');

  // Use res.on('close') (not req.on('close'), which fires once the body is read)
  res.on('close', () => writer.close());

  writer.start(messageId);
  writer.textStart(textId);

  // Your streaming logic
  writer.textDelta(textId, 'Hello ');
  writer.textDelta(textId, 'World!');

  writer.textEnd(textId);
  writer.finish(messageId);
  writer.close();
});

API 参考

协作者服务员

class CogitatorServer {
  constructor(options: CogitatorServerConfig);
  init(): Promise<void>;
  attachWebSocket(server: http.Server): Promise<WebSocketServer>;
  readonly isInitialized: boolean;
}

协作者服务器配置

interface CogitatorServerConfig {
  app: Router;
  cogitator: Cogitator;
  agents?: Record<string, Agent>;
  workflows?: Record<string, Workflow>;
  swarms?: Record<string, SwarmConfig>;
  config?: {
    basePath?: string; // Default: '/cogitator'
    enableWebSocket?: boolean; // Default: false
    enableSwagger?: boolean; // Default: true
    auth?: AuthFunction;
    rateLimit?: RateLimitConfig;
    cors?: CorsConfig;
    swagger?: SwaggerConfig;
    websocket?: WebSocketConfig;
    sseHeartbeatMs?: number; // Default: 5000, 0 turns SSE heartbeats off
    acceptContext?: boolean | string[]; // context keys clients may send: none by default, a list, or true for any
    threadMessageRoles?: ('user' | 'assistant' | 'system')[]; // roles of POST /threads/:id/messages, default user and assistant
    limits?: RunLimits; // { swarmTimeout?, workflowMaxIterations?, workflowMaxConcurrency? }
    anonymous?: 'allow' | 'deny'; // Default: 'allow'
    authorizeThread?: (
      auth: AuthContext | undefined,
      threadId: string
    ) => boolean | Promise<boolean>;
    deps?: (req: Request) => unknown | Promise<unknown>; // HTTP runs, WebSocket runs use websocket.deps
    workflowCheckpoints?: CheckpointStore; // workflow checkpoints and /resume, none by default
    resumableStreams?: boolean | ResumableStreamOptions; // see Resumable streams, off by default
  };
}

interface WebSocketConfig {
  path?: string; // Default: `${basePath}/ws`
  pingInterval?: number; // Default: 30000
  maxPayloadSize?: number; // Default: 1 MiB
  deps?: (req: IncomingMessage) => unknown | Promise<unknown>; // deps of WebSocket runs, from the upgrade request
}

CogitatorServer 当控制台启动时, 它不会写给它 . init() 和 WebSocket 设置报告 debug 职等(getLogger() 从 @cogitator-ai/core, (中文(简体) ), LOG_LEVEL=debug 看到他们,或 setLogger() 引导它们进入自己的日志。 错误仍在记录中 .

快报

class ExpressStreamWriter {
  constructor(res: Response);
  start(messageId: string): void;
  textStart(id: string): void;
  textDelta(id: string, delta: string): void;
  textEnd(id: string): void;
  reasoningStart(id: string): void;
  reasoningDelta(id: string, delta: string): void;
  reasoningEnd(id: string): void;
  toolCallStart(id: string, toolName: string): void;
  toolCallDelta(id: string, argsTextDelta: string): void;
  toolCallEnd(id: string): void;
  toolResult(id: string, toolCallId: string, result: unknown): void;
  approvalRequired(threadId: string, approvals: readonly PendingApproval[]): void;
  workflowEvent(event: string, data: unknown): void;
  swarmEvent(event: string, data: unknown): void;
  error(message: string, code?: string): void;
  finish(messageId: string, usage?: Usage): void;
  close(): void;
  readonly isClosed: boolean;
}

回复结束后的写作(或客户端已断开)被忽略.

处理错误

所有端点返回一致的错误反应 :

{
  "error": {
    "message": "Agent 'unknown' not found",
    "code": "NOT_FOUND"
  }
}

HTTP 状态代码错误代码映射 :

• INVALID_INPUT - 400个(鉴定错误和JSON身体缺陷)
• UNAUTHORIZED 第401条
• THREAD_ACCESS_DENIED ^ 403(另一个用户的线程)
• THREAD_FORBIDDEN ——403(authorizeThread 拒绝线索)
• NOT_FOUND 第404条
• RUN_NOT_PAUSED ~ 409 (恢复线程而不暂停运行)
• PAYLOAD_TOO_LARGE − 413个
• RATE_LIMIT_EXCEEDED - 429个
• INTERNAL_ERROR ——500个
• UNIMPLEMENTED + 501(缺少可选工作流程/群包)
• UNAVAILABLE ~ 503 (没有配置内存)
• RUN_TIMEOUT • 504(一群人没有时间)

A 类 CogitatorError 保留它的信息、代码和答案 statusCode,即代码的映射状态,除非错误被赋予了另一个(例如) LLM_RATE_LIMITED - 429页)。 当它说什么时候重试(retryAfter,正如供应商费率限制所做的那样,答案带有 Retry-After 在整秒钟。 地图是 describeHttpError 联 合 国 @cogitator-ai/server-shared,所以快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快取快 服务器上记录了任何其他错误并回答为 500 Internal server error,因此连接字符串,文件路径和其他内部内容永远无法到达客户端. 同样的映射适用于一个失败的工作流程, 同样的遮盖到SSE error 事件、 node_error编号 :agent_error 流事件和 WebSocket 运行错误.

文档

全向导 : cogitator.app/docs/服务器-适应器/表达式. 工具核准: 核准/文件/工具/核准。 。 。 .

许可证

麻省理工学院