fix(anthropic): cover retry stream recreation

.changeset/orange-squids-cover.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents-plugin-anthropic': patch
+---
+
+added anthropic plugin

plugins/anthropic/README.md

@@ -0,0 +1,22 @@
+# @livekit/agents-plugin-anthropic
+
+Anthropic plugin for LiveKit Node Agents.
+
+## Installation
+
+```bash
+npm install @livekit/agents-plugin-anthropic
+```
+
+## Usage
+
+```typescript
+import { anthropic } from '@livekit/agents-plugin-anthropic';
+
+const agent = new Agent({
+  llm: new anthropic.LLM({
+    model: 'claude-3-5-sonnet-20241022',
+    // caching: 'ephemeral' // uncomment to enable prompt caching
+  }),
+});
+```

plugins/anthropic/api-extractor.json

@@ -0,0 +1,20 @@
+/**
+ * Config file for API Extractor.  For more info, please visit: https://api-extractor.com
+ */
+{
+  "$schema": "https://developer.microsoft.com/json-schemas/api-extractor/v7/api-extractor.schema.json",
+
+  /**
+   * Optionally specifies another JSON config file that this file extends from.  This provides a way for
+   * standard settings to be shared across multiple projects.
+   *
+   * If the path starts with "./" or "../", the path is resolved relative to the folder of the file that contains
+   * the "extends" field.  Otherwise, the first path segment is interpreted as an NPM package name, and will be
+   * resolved using NodeJS require().
+   *
+   * SUPPORTED TOKENS: none
+   * DEFAULT VALUE: ""
+   */
+  "extends": "../../api-extractor-shared.json",
+  "mainEntryPointFilePath": "./dist/index.d.ts"
+}

plugins/anthropic/package.json

@@ -0,0 +1,51 @@
+{
+  "name": "@livekit/agents-plugin-anthropic",
+  "version": "1.0.0",
+  "description": "Anthropic plugin for LiveKit Node Agents",
+  "main": "dist/index.js",
+  "require": "dist/index.cjs",
+  "types": "dist/index.d.ts",
+  "exports": {
+    "import": {
+      "types": "./dist/index.d.ts",
+      "default": "./dist/index.js"
+    },
+    "require": {
+      "types": "./dist/index.d.cts",
+      "default": "./dist/index.cjs"
+    }
+  },
+  "author": "LiveKit",
+  "type": "module",
+  "repository": "git@github.com:livekit/agents-js.git",
+  "license": "Apache-2.0",
+  "files": [
+    "dist",
+    "src",
+    "README.md"
+  ],
+  "scripts": {
+    "build": "tsup --onSuccess \"pnpm build:types\"",
+    "build:types": "tsc --declaration --emitDeclarationOnly && node ../../scripts/copyDeclarationOutput.js",
+    "clean": "rm -rf dist",
+    "clean:build": "pnpm clean && pnpm build",
+    "lint": "eslint -f unix \"src/**/*.{ts,js}\"",
+    "api:check": "api-extractor run --typescript-compiler-folder ../../node_modules/typescript",
+    "api:update": "api-extractor run --local --typescript-compiler-folder ../../node_modules/typescript --verbose"
+  },
+  "devDependencies": {
+    "@livekit/agents": "workspace:*",
+    "@livekit/agents-plugins-test": "workspace:*",
+    "@livekit/rtc-node": "catalog:",
+    "@microsoft/api-extractor": "^7.35.0",
+    "tsup": "^8.3.5",
+    "typescript": "^5.0.0"
+  },
+  "dependencies": {
+    "@anthropic-ai/sdk": "^0.33.1"
+  },
+  "peerDependencies": {
+    "@livekit/agents": "workspace:*",
+    "@livekit/rtc-node": "catalog:"
+  }
+}

plugins/anthropic/src/index.ts

@@ -0,0 +1,19 @@
+// SPDX-FileCopyrightText: 2025 LiveKit, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+import { Plugin } from '@livekit/agents';
+
+export { LLM, LLMStream, type LLMOptions } from './llm.js';
+export * from './models.js';
+
+class AnthropicPlugin extends Plugin {
+  constructor() {
+    super({
+      title: 'anthropic',
+      version: __PACKAGE_VERSION__,
+      package: __PACKAGE_NAME__,
+    });
+  }
+}
+
+Plugin.registerPlugin(new AnthropicPlugin());

plugins/anthropic/src/llm.test.ts

@@ -0,0 +1,241 @@
+// SPDX-FileCopyrightText: 2025 LiveKit, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+import type Anthropic from '@anthropic-ai/sdk';
+import { llm } from '@livekit/agents';
+import { describe, expect, it } from 'vitest';
+import { LLM } from './llm.js';
+
+function messageStartEvent(): Anthropic.MessageStreamEvent {
+  return {
+    type: 'message_start',
+    message: {
+      id: 'msg_123',
+      usage: { input_tokens: 0, output_tokens: 0 },
+    },
+  } as Anthropic.MessageStreamEvent;
+}
+
+function textDeltaEvent(text: string): Anthropic.MessageStreamEvent {
+  return {
+    type: 'content_block_delta',
+    delta: { type: 'text_delta', text },
+  } as Anthropic.MessageStreamEvent;
+}
+
+async function collectTextFromEvents(
+  events: Anthropic.MessageStreamEvent[],
+  toolCtx?: llm.ToolContext,
+): Promise<string> {
+  const client = {
+    messages: {
+      create: async () =>
+        (async function* (): AsyncGenerator<Anthropic.MessageStreamEvent> {
+          yield* events;
+        })(),
+    },
+  } as unknown as Anthropic;
+  const anthropicLlm = new LLM({
+    apiKey: 'dummy',
+    client,
+    model: 'claude-3-5-sonnet-20241022',
+  });
+  const chatCtx = new llm.ChatContext();
+  chatCtx.addMessage({ role: 'user', content: 'Hello, world!' });
+
+  const stream = anthropicLlm.chat({ chatCtx, toolCtx });
+  const textChunks: string[] = [];
+  for await (const chunk of stream) {
+    if (chunk.delta?.content) {
+      textChunks.push(chunk.delta.content);
+    }
+  }
+  return textChunks.join('');
+}
+
+describe('Anthropic LLM', () => {
+  it('correctly maps ChatContext to Anthropic system and messages arrays', () => {
+    const anthropicLlm = new LLM({ apiKey: 'dummy', model: 'claude-3-5-sonnet-20241022' });
+
+    const chatCtx = new llm.ChatContext();
+    chatCtx.addMessage({
+      role: 'system',
+      content: 'You are a mock agent.',
+    });
+    chatCtx.addMessage({
+      role: 'user',
+      content: 'Hello, world!',
+    });
+
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    const { system, messages } = (anthropicLlm as any)._buildAnthropicContext(chatCtx);
+
+    // Assert that system prompts were correctly isolated
+    expect(system).toHaveLength(1);
+    expect(system[0].text).toBe('You are a mock agent.');
+
+    // Assert that strictly user messages ended up in the messages payload
+    expect(messages).toHaveLength(1);
+    expect(messages[0].role).toBe('user');
+    expect(messages[0].content).toBe('Hello, world!');
+  });
+
+  it('merges consecutive same-role messages', () => {
+    const anthropicLlm = new LLM({ apiKey: 'dummy', model: 'claude-3-5-sonnet-20241022' });
+
+    const chatCtx = new llm.ChatContext();
+    chatCtx.addMessage({ role: 'user', content: 'First message' });
+    chatCtx.addMessage({ role: 'user', content: 'Second message' });
+    chatCtx.addMessage({ role: 'assistant', content: 'Reply' });
+
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    const { messages } = (anthropicLlm as any)._buildAnthropicContext(chatCtx);
+
+    // Two user messages should be merged into one with array content
+    expect(messages).toHaveLength(2);
+    expect(messages[0].role).toBe('user');
+    expect(Array.isArray(messages[0].content)).toBe(true);
+    expect(messages[0].content).toHaveLength(2);
+    expect(messages[1].role).toBe('assistant');
+  });
+
+  it('injects a dummy user message if conversation starts with assistant', () => {
+    const anthropicLlm = new LLM({ apiKey: 'dummy', model: 'claude-3-5-sonnet-20241022' });
+
+    const chatCtx = new llm.ChatContext();
+    chatCtx.addMessage({ role: 'system', content: 'System prompt' });
+    chatCtx.addMessage({ role: 'assistant', content: 'I start talking' });
+
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    const { messages } = (anthropicLlm as any)._buildAnthropicContext(chatCtx);
+
+    // Should have injected a dummy user message at the start
+    expect(messages[0].role).toBe('user');
+    expect(messages[0].content).toBe('(empty)');
+    expect(messages[1].role).toBe('assistant');
+  });
+
+  it('handles function_call and function_call_output items', () => {
+    const anthropicLlm = new LLM({ apiKey: 'dummy', model: 'claude-3-5-sonnet-20241022' });
+
+    const chatCtx = new llm.ChatContext();
+    chatCtx.addMessage({ role: 'user', content: 'What is the weather?' });
+
+    // Simulate a function call from the assistant
+    chatCtx.items.push(
+      new llm.FunctionCall({
+        callId: 'call_123',
+        name: 'get_weather',
+        args: '{"city":"London"}',
+      }),
+    );
+
+    // Simulate the tool result
+    chatCtx.items.push(
+      new llm.FunctionCallOutput({
+        callId: 'call_123',
+        name: 'get_weather',
+        output: '{"temp": 20}',
+        isError: false,
+      }),
+    );
+
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    const { messages } = (anthropicLlm as any)._buildAnthropicContext(chatCtx);
+
+    // user message, then assistant tool_use, then user tool_result
+    expect(messages).toHaveLength(3);
+    expect(messages[0].role).toBe('user');
+    expect(messages[1].role).toBe('assistant');
+    expect(Array.isArray(messages[1].content)).toBe(true);
+    expect(messages[1].content[0].type).toBe('tool_use');
+    expect(messages[1].content[0].id).toBe('call_123');
+    expect(messages[2].role).toBe('user');
+    expect(Array.isArray(messages[2].content)).toBe(true);
+    expect(messages[2].content[0].type).toBe('tool_result');
+  });
+
+  it('creates a fresh stream on retry', async () => {
+    let calls = 0;
+    const client = {
+      messages: {
+        create: async () => {
+          calls += 1;
+          if (calls === 1) {
+            throw new Error('transient connect failure');
+          }
+          return (async function* (): AsyncGenerator<Anthropic.MessageStreamEvent> {})();
+        },
+      },
+    } as unknown as Anthropic;
+    const anthropicLlm = new LLM({
+      apiKey: 'dummy',
+      client,
+      model: 'claude-3-5-sonnet-20241022',
+    });
+    anthropicLlm.on('error', () => {});
+    const chatCtx = new llm.ChatContext();
+    chatCtx.addMessage({ role: 'user', content: 'Hello, world!' });
+
+    const stream = anthropicLlm.chat({
+      chatCtx,
+      connOptions: { maxRetry: 1, retryIntervalMs: 0, timeoutMs: 1000 },
+    });
+    const chunks: llm.ChatChunk[] = [];
+    for await (const chunk of stream) {
+      chunks.push(chunk);
+    }
+
+    expect(calls).toBe(2);
+    expect(chunks.at(-1)?.usage).toEqual({
+      completionTokens: 0,
+      promptTokens: 0,
+      promptCachedTokens: 0,
+      totalTokens: 0,
+    });
+  });
+
+  it('filters thinking blocks when tools are active', async () => {
+    const text = await collectTextFromEvents(
+      [
+        messageStartEvent(),
+        textDeltaEvent('<thinking>hidden'),
+        textDeltaEvent('still hidden</thinking>visible'),
+      ],
+      {},
+    );
+
+    expect(text).toBe('visible');
+  });
+
+  it('does not filter thinking blocks without tools', async () => {
+    const text = await collectTextFromEvents([
+      messageStartEvent(),
+      textDeltaEvent('<thinking>visible without tools</thinking>'),
+    ]);
+
+    expect(text).toBe('<thinking>visible without tools</thinking>');
+  });
+
+  it('preserves text around same-delta thinking blocks', async () => {
+    const text = await collectTextFromEvents(
+      [messageStartEvent(), textDeltaEvent('before <thinking>hidden</thinking> after')],
+      {},
+    );
+
+    expect(text).toBe('before  after');
+  });
+
+  it('preserves text before split thinking blocks', async () => {
+    const text = await collectTextFromEvents(
+      [
+        messageStartEvent(),
+        textDeltaEvent('before <thinking>hidden'),
+        textDeltaEvent('still hidden</thinking> after'),
+      ],
+      {},
+    );
+
+    expect(text).toBe('before  after');
+  });
+});