From e44335c2e00c89e3f89fb7b063e4b33298d1a60c Mon Sep 17 00:00:00 2001 From: andrewmadson-pplx Date: Fri, 28 Aug 2026 15:14:24 -0700 Subject: [PATCH 1/8] feat(perplexity): add Agent API presets --- packages/baseai/src/data/models.ts | 34 +- packages/baseai/src/dev/data/models.ts | 34 +- packages/baseai/src/dev/llms/call-llm.ts | 3 +- .../baseai/src/dev/llms/call-perplexity.ts | 19 +- .../perplexity/agentResponse.test.ts | 468 ++++++++++ .../dev/providers/perplexity/agentResponse.ts | 808 ++++++++++++++++++ .../providers/perplexity/agentStream.test.ts | 305 +++++++ .../src/dev/providers/perplexity/api.ts | 3 +- .../providers/perplexity/catalog.node.test.ts | 65 ++ .../providers/perplexity/legacyStream.test.ts | 82 ++ .../providers/perplexity/perplexity.test.ts | 89 ++ .../baseai/src/dev/routes/v1/pipes/run.ts | 2 +- .../baseai/src/dev/utils/stream/stream.ts | 59 +- packages/baseai/types/model.ts | 4 + packages/baseai/types/providers.ts | 3 + packages/baseai/vitest.edge.config.js | 20 + packages/baseai/vitest.node.config.js | 20 + packages/core/src/data/models.ts | 34 +- packages/core/types/model.ts | 4 + packages/core/types/pipes.ts | 12 + 20 files changed, 2025 insertions(+), 43 deletions(-) create mode 100644 packages/baseai/src/dev/providers/perplexity/agentResponse.test.ts create mode 100644 packages/baseai/src/dev/providers/perplexity/agentResponse.ts create mode 100644 packages/baseai/src/dev/providers/perplexity/agentStream.test.ts create mode 100644 packages/baseai/src/dev/providers/perplexity/catalog.node.test.ts create mode 100644 packages/baseai/src/dev/providers/perplexity/legacyStream.test.ts create mode 100644 packages/baseai/src/dev/providers/perplexity/perplexity.test.ts create mode 100644 packages/baseai/vitest.edge.config.js create mode 100644 packages/baseai/vitest.node.config.js diff --git a/packages/baseai/src/data/models.ts b/packages/baseai/src/data/models.ts index d8c4e055..c325ac7b 100644 --- a/packages/baseai/src/data/models.ts +++ b/packages/baseai/src/data/models.ts @@ -96,9 +96,9 @@ export const OLLAMA: string = 'ollama'; interface Model { id: string; provider: string; - promptCost: number; - completionCost: number; - requestCost?: number; + promptCost: number | null; + completionCost: number | null; + requestCost?: number | null; toolSupport?: { toolChoice: boolean; parallelToolCalls: boolean; @@ -577,6 +577,34 @@ export const modelsByProvider: ModelsByProviderInclCosts = { } ], [PERPLEXITY]: [ + { + id: 'fast', + provider: PERPLEXITY, + promptCost: null, + completionCost: null, + requestCost: null + }, + { + id: 'low', + provider: PERPLEXITY, + promptCost: null, + completionCost: null, + requestCost: null + }, + { + id: 'medium', + provider: PERPLEXITY, + promptCost: null, + completionCost: null, + requestCost: null + }, + { + id: 'high', + provider: PERPLEXITY, + promptCost: null, + completionCost: null, + requestCost: null + }, { id: 'llama-3.1-sonar-huge-128k-online', provider: PERPLEXITY, diff --git a/packages/baseai/src/dev/data/models.ts b/packages/baseai/src/dev/data/models.ts index b82060c5..e3ee0c28 100644 --- a/packages/baseai/src/dev/data/models.ts +++ b/packages/baseai/src/dev/data/models.ts @@ -16,9 +16,9 @@ export const X_AI: string = 'xAI'; interface Model { id: string; provider: string; - promptCost: number; - completionCost: number; - requestCost?: number; + promptCost: number | null; + completionCost: number | null; + requestCost?: number | null; toolSupport?: { toolChoice: boolean; parallelToolCalls: boolean; @@ -497,6 +497,34 @@ export const modelsByProvider: ModelsByProviderInclCosts = { } ], [PERPLEXITY]: [ + { + id: 'fast', + provider: PERPLEXITY, + promptCost: null, + completionCost: null, + requestCost: null + }, + { + id: 'low', + provider: PERPLEXITY, + promptCost: null, + completionCost: null, + requestCost: null + }, + { + id: 'medium', + provider: PERPLEXITY, + promptCost: null, + completionCost: null, + requestCost: null + }, + { + id: 'high', + provider: PERPLEXITY, + promptCost: null, + completionCost: null, + requestCost: null + }, { id: 'llama-3.1-sonar-huge-128k-online', provider: PERPLEXITY, diff --git a/packages/baseai/src/dev/llms/call-llm.ts b/packages/baseai/src/dev/llms/call-llm.ts index dcbf9b74..f1da0801 100644 --- a/packages/baseai/src/dev/llms/call-llm.ts +++ b/packages/baseai/src/dev/llms/call-llm.ts @@ -158,7 +158,8 @@ export async function callLLM({ pipe, messages, llmApiKey, - stream + stream, + paramsTools }); } diff --git a/packages/baseai/src/dev/llms/call-perplexity.ts b/packages/baseai/src/dev/llms/call-perplexity.ts index 5be492d7..46dd827e 100644 --- a/packages/baseai/src/dev/llms/call-perplexity.ts +++ b/packages/baseai/src/dev/llms/call-perplexity.ts @@ -3,22 +3,39 @@ import transformToProviderRequest from '../utils/provider-handlers/transfrom-to- import { handleProviderRequest } from '../utils/provider-handlers/provider-request-handler'; import { PERPLEXITY } from '../data/models'; +import { + callPerplexityAgent, + getPerplexityTransport +} from '../providers/perplexity/agentResponse'; import { handleLlmError } from './utils'; import type { Message, Pipe } from 'types/pipe'; import type { ModelParams } from 'types/providers'; +import type { PipeTool } from 'types/tools'; export async function callPerplexity({ pipe, messages, llmApiKey, - stream + stream, + paramsTools }: { pipe: Pipe; llmApiKey: string; stream: boolean; messages: Message[]; + paramsTools?: PipeTool[]; }) { try { + if (getPerplexityTransport(pipe.model) === 'agentResponse') { + return await callPerplexityAgent({ + pipe, + messages, + llmApiKey, + stream, + paramsTools + }); + } + const modelParams = buildModelParams(pipe, stream, messages); // Transform params according to provider's format diff --git a/packages/baseai/src/dev/providers/perplexity/agentResponse.test.ts b/packages/baseai/src/dev/providers/perplexity/agentResponse.test.ts new file mode 100644 index 00000000..1c477b84 --- /dev/null +++ b/packages/baseai/src/dev/providers/perplexity/agentResponse.test.ts @@ -0,0 +1,468 @@ +import { describe, expect, it, vi } from 'vitest'; +import type { Pipe } from 'types/pipe'; +import type { ProviderMessage } from 'types/providers'; +import { + buildPerplexityAgentRequest, + callPerplexityAgent, + getPerplexityAgentURL, + transformPerplexityAgentResponse, + type PerplexityAgentResponse +} from './agentResponse'; + +function scaffoldPipe(overrides: Partial = {}): Pipe { + return { + name: 'perplexity-agent', + description: 'Perplexity Agent API example', + status: 'private', + model: 'perplexity:fast', + stream: true, + json: false, + store: true, + moderate: true, + top_p: 1, + max_tokens: 1000, + temperature: 0.7, + presence_penalty: 1, + frequency_penalty: 1, + stop: [], + tool_choice: 'auto', + parallel_tool_calls: true, + messages: [], + variables: [], + memory: [], + tools: [], + ...overrides + }; +} + +const messages: ProviderMessage[] = [ + { role: 'system', content: 'Be concise.' }, + { role: 'user', content: 'What changed?' }, + { role: 'assistant', content: 'I will check.' } +]; + +const completedResponse: PerplexityAgentResponse = { + id: 'resp_agent_123', + object: 'response', + created_at: 1787928000, + status: 'completed', + model: 'perplexity/sonar', + output: [ + { + type: 'search_results', + results: [ + { id: 1, url: 'https://example.com/one' }, + { id: 2, url: 'https://example.com/two' } + ] + }, + { + type: 'message', + id: 'msg_123', + role: 'assistant', + status: 'completed', + content: [ + { + type: 'output_text', + text: 'Alpha[1] ', + annotations: [ + { + type: 'url_citation', + start_index: 0, + end_index: 5, + url: 'https://example.com/direct' + } + ] + }, + { + type: 'output_text', + text: 'Beta [web:2].' + } + ] + } + ], + usage: { + input_tokens: 12, + output_tokens: 8, + total_tokens: 20 + } +}; + +describe('Perplexity Agent request adapter', () => { + it('maps a standard scaffold changed only to perplexity:fast', () => { + const request = buildPerplexityAgentRequest({ + pipe: scaffoldPipe(), + messages, + stream: true + }); + + expect(request).toEqual({ + preset: 'fast', + input: [ + { type: 'message', role: 'system', content: 'Be concise.' }, + { + type: 'message', + role: 'user', + content: 'What changed?' + }, + { + type: 'message', + role: 'assistant', + content: 'I will check.' + } + ], + max_output_tokens: 1000, + stream: true, + temperature: 0.7, + top_p: 1 + }); + expect(request).not.toHaveProperty('model'); + expect(request).not.toHaveProperty('store'); + expect(request).not.toHaveProperty('moderate'); + expect(request).not.toHaveProperty('tools'); + }); + + it('maps supported text and image content parts', () => { + const request = buildPerplexityAgentRequest({ + pipe: scaffoldPipe(), + messages: [ + { + role: 'user', + content: [ + { type: 'text', text: 'Describe this.' }, + { + type: 'image_url', + image_url: { url: 'https://example.com/image.png' } + } + ] + } + ], + stream: false + }); + + expect(request.input[0].content).toEqual([ + { type: 'input_text', text: 'Describe this.' }, + { + type: 'input_image', + image_url: 'https://example.com/image.png' + } + ]); + }); + + it('preserves store and moderate as platform behavior without forwarding them', () => { + const request = buildPerplexityAgentRequest({ + pipe: scaffoldPipe({ store: false, moderate: false }), + messages, + stream: false + }); + expect(request).not.toHaveProperty('store'); + expect(request).not.toHaveProperty('moderate'); + }); + + it.each([ + ['presence_penalty', { presence_penalty: 0 }], + ['frequency_penalty', { frequency_penalty: 0 }], + ['stop', { stop: ['END'] }], + ['json', { json: true }], + ['tool_choice', { tool_choice: 'required' }], + ['parallel_tool_calls', { parallel_tool_calls: false }], + ['max_tokens', { max_tokens: 0 }] + ])('rejects unsupported %s values by field name', (field, overrides) => { + expect(() => + buildPerplexityAgentRequest({ + pipe: scaffoldPipe(overrides as Partial), + messages, + stream: false + }) + ).toThrow(field); + }); + + it('rejects Pipe tools and request tools', () => { + const tool = { + type: 'function', + function: { name: 'lookup', parameters: {} }, + run: vi.fn() + }; + expect(() => + buildPerplexityAgentRequest({ + pipe: scaffoldPipe({ tools: [tool] as Pipe['tools'] }), + messages, + stream: false + }) + ).toThrow('tools'); + expect(() => + buildPerplexityAgentRequest({ + pipe: scaffoldPipe(), + messages, + stream: false, + paramsTools: [tool] as Pipe['tools'] + }) + ).toThrow('tools'); + }); + + it.each([ + [ + 'messages[0].role', + { role: 'tool', content: 'result', tool_call_id: 'call_1' } + ], + [ + 'messages[0].tool_calls', + { role: 'assistant', content: null, tool_calls: [] } + ], + [ + 'messages[0].name', + { role: 'assistant', content: 'hello', name: 'named' } + ] + ])('rejects unsupported history at %s', (field, message) => { + expect(() => + buildPerplexityAgentRequest({ + pipe: scaffoldPipe(), + messages: [message as ProviderMessage], + stream: false + }) + ).toThrow(field); + }); + + it('rejects unsupported content parts by field path', () => { + expect(() => + buildPerplexityAgentRequest({ + pipe: scaffoldPipe(), + messages: [ + { + role: 'user', + content: [{ type: 'file_url' }] as never + } + ], + stream: false + }) + ).toThrow('messages[0].content[0].type'); + }); + + it('rejects non-empty tools before network I/O', async () => { + const fetcher = vi.fn(); + await expect( + callPerplexityAgent( + { + pipe: scaffoldPipe(), + messages, + llmApiKey: 'test-key', + stream: false, + paramsTools: [{ type: 'function' }] as never + }, + { fetcher: fetcher as typeof fetch } + ) + ).rejects.toThrow('tools'); + expect(fetcher).not.toHaveBeenCalled(); + }); +}); + +describe('Perplexity Agent HTTP adapter', () => { + it('joins Agent URLs with and without trailing slashes', () => { + expect(getPerplexityAgentURL('https://api.perplexity.ai')).toBe( + 'https://api.perplexity.ai/v1/agent' + ); + expect(getPerplexityAgentURL('https://api.perplexity.ai/')).toBe( + 'https://api.perplexity.ai/v1/agent' + ); + }); + + it('sends the exact method, URL, headers, preset, and request fields', async () => { + const fetcher = vi.fn( + async () => + new Response(JSON.stringify(completedResponse), { + status: 200, + headers: { 'content-type': 'application/json' } + }) + ); + + await callPerplexityAgent( + { + pipe: scaffoldPipe(), + messages, + llmApiKey: 'test-key', + stream: false + }, + { + baseURL: 'https://api.perplexity.ai/', + fetcher: fetcher as typeof fetch + } + ); + + expect(fetcher).toHaveBeenCalledOnce(); + const [url, init] = fetcher.mock.calls[0] as unknown as [ + string, + NonNullable[1]> + ]; + expect(url).toBe('https://api.perplexity.ai/v1/agent'); + expect(init.method).toBe('POST'); + expect(init.headers).toEqual({ + Authorization: 'Bearer test-key', + 'content-type': 'application/json', + accept: 'application/json' + }); + expect(JSON.parse(init.body as string)).toEqual({ + preset: 'fast', + input: [ + { type: 'message', role: 'system', content: 'Be concise.' }, + { + type: 'message', + role: 'user', + content: 'What changed?' + }, + { + type: 'message', + role: 'assistant', + content: 'I will check.' + } + ], + max_output_tokens: 1000, + stream: false, + temperature: 0.7, + top_p: 1 + }); + }); + + it('uses only safe Agent error fields', async () => { + const fetcher = vi.fn( + async () => + new Response( + JSON.stringify({ + error: { + message: 'Preset is invalid', + type: 'invalid_request', + private_prompt: 'must not be included' + } + }), + { status: 400 } + ) + ); + + await expect( + callPerplexityAgent( + { + pipe: scaffoldPipe(), + messages, + llmApiKey: 'test-key', + stream: false + }, + { fetcher: fetcher as typeof fetch } + ) + ).rejects.toThrow('Preset is invalid'); + }); +}); + +describe('Perplexity Agent response adapter', () => { + it('maps the complete BaseAI response envelope and citations', () => { + expect( + transformPerplexityAgentResponse( + completedResponse, + 'perplexity:fast' + ) + ).toEqual({ + id: 'resp_agent_123', + object: 'chat.completion', + created: 1787928000, + model: 'perplexity:fast', + provider: 'Perplexity', + choices: [ + { + message: { + role: 'assistant', + content: 'Alpha[1] Beta [web:2].', + citationMetadata: { + citationSources: [ + { + startIndex: 0, + endIndex: 5, + uri: 'https://example.com/direct' + }, + { + startIndex: 5, + endIndex: 8, + uri: 'https://example.com/one' + }, + { + startIndex: 14, + endIndex: 21, + uri: 'https://example.com/two' + } + ] + } + }, + index: 0, + logprobs: null, + finish_reason: 'stop' + } + ], + usage: { + prompt_tokens: 12, + completion_tokens: 8, + total_tokens: 20 + } + }); + }); + + it('uses a frozen zero-usage rule when Agent usage is absent', () => { + const response = { ...completedResponse, usage: undefined }; + expect( + transformPerplexityAgentResponse(response, 'perplexity:low').usage + ).toEqual({ + prompt_tokens: 0, + completion_tokens: 0, + total_tokens: 0 + }); + }); + + it('maps documented token exhaustion to length', () => { + const response: PerplexityAgentResponse = { + ...completedResponse, + status: 'incomplete', + incomplete_details: { reason: 'max_output_tokens' } + }; + expect( + transformPerplexityAgentResponse(response, 'perplexity:medium') + .choices[0].finish_reason + ).toBe('length'); + }); + + it('rejects undocumented incomplete reasons', () => { + expect(() => + transformPerplexityAgentResponse( + { + ...completedResponse, + status: 'incomplete', + incomplete_details: { reason: 'max_tokens' } + }, + 'perplexity:medium' + ) + ).toThrow(); + }); + + it.each([ + ['incomplete', undefined], + ['failed', { message: 'upstream failed' }], + ['cancelled', undefined], + ['in_progress', undefined] + ])('rejects %s responses outside token exhaustion', (status, error) => { + expect(() => + transformPerplexityAgentResponse( + { ...completedResponse, status, error }, + 'perplexity:high' + ) + ).toThrow(); + }); + + it('rejects malformed and empty completed responses', () => { + expect(() => + transformPerplexityAgentResponse( + { ...completedResponse, id: undefined }, + 'perplexity:fast' + ) + ).toThrow('missing identity'); + expect(() => + transformPerplexityAgentResponse( + { ...completedResponse, output: [] }, + 'perplexity:fast' + ) + ).toThrow('empty completed output'); + }); +}); diff --git a/packages/baseai/src/dev/providers/perplexity/agentResponse.ts b/packages/baseai/src/dev/providers/perplexity/agentResponse.ts new file mode 100644 index 00000000..5eb16694 --- /dev/null +++ b/packages/baseai/src/dev/providers/perplexity/agentResponse.ts @@ -0,0 +1,808 @@ +import { PERPLEXITY } from '@/dev/data/models'; +import { ApiError } from '@/dev/hono/errors'; +import { Stream } from '@/dev/utils/stream/stream'; +import type { Pipe } from 'types/pipe'; +import type { + ChatCompletionResponse, + CitationSource, + ContentType, + ProviderMessage +} from 'types/providers'; +import type { PipeTool } from 'types/tools'; +import PerplexityAIApiConfig from './api'; + +export const PERPLEXITY_AGENT_PRESETS = [ + 'fast', + 'low', + 'medium', + 'high' +] as const; + +export type PerplexityAgentPreset = (typeof PERPLEXITY_AGENT_PRESETS)[number]; + +type AgentInputContentPart = + | { type: 'input_text'; text: string } + | { type: 'input_image'; image_url: string }; + +interface AgentInputMessage { + type: 'message'; + role: 'system' | 'user' | 'assistant'; + content: string | AgentInputContentPart[]; +} + +export interface PerplexityAgentRequest { + input: AgentInputMessage[]; + max_output_tokens: number; + preset: PerplexityAgentPreset; + stream: boolean; + temperature: number; + top_p: number; +} + +interface AgentAnnotation { + end_index?: number; + id?: number | string; + search_result_id?: number | string; + source_id?: number | string; + start_index?: number; + type?: string; + uri?: string; + url?: string; +} + +interface AgentOutputText { + annotations?: AgentAnnotation[]; + text: string; + type: 'output_text'; +} + +interface AgentMessageOutput { + content: AgentOutputText[]; + id: string; + role: 'assistant'; + status: string; + type: 'message'; +} + +interface AgentSearchResult { + id: number | string; + url: string; +} + +interface AgentSearchResultsOutput { + results: AgentSearchResult[]; + type: 'search_results'; +} + +type AgentOutputItem = + | AgentMessageOutput + | AgentSearchResultsOutput + | { type: string; [key: string]: unknown }; + +interface AgentUsage { + input_tokens?: number; + output_tokens?: number; + total_tokens?: number; +} + +interface AgentErrorInfo { + code?: string; + message?: string; + type?: string; +} + +export interface PerplexityAgentResponse { + created_at?: number; + error?: AgentErrorInfo | null; + id?: string; + incomplete_details?: { + reason?: string; + } | null; + model?: string; + object?: string; + output?: AgentOutputItem[]; + status?: string; + usage?: AgentUsage; +} + +interface AgentStreamEvent { + content_index?: number; + delta?: string; + error?: AgentErrorInfo; + event?: string; + item_id?: string; + output_index?: number; + response?: PerplexityAgentResponse; + sequence_number?: number; + type?: string; + data?: AgentStreamEvent; +} + +interface AgentCallOptions { + baseURL?: string; + fetcher?: typeof fetch; +} + +const AGENT_OBJECT = 'chat.completion'; +const AGENT_STREAM_OBJECT = 'chat.completion.chunk'; + +export function isPerplexityAgentModel(model: string): boolean { + const preset = model.startsWith('perplexity:') + ? model.slice('perplexity:'.length) + : model; + return PERPLEXITY_AGENT_PRESETS.includes(preset as PerplexityAgentPreset); +} + +export function getPerplexityTransport( + model: string +): 'agentResponse' | 'chatComplete' { + return isPerplexityAgentModel(model) ? 'agentResponse' : 'chatComplete'; +} + +function getAgentPreset(model: string): PerplexityAgentPreset { + const preset = model.startsWith('perplexity:') + ? model.slice('perplexity:'.length) + : model; + if (!isPerplexityAgentModel(preset)) { + throw new ApiError({ + code: 'BAD_REQUEST', + message: `Unsupported Perplexity Agent preset: ${model}` + }); + } + return preset as PerplexityAgentPreset; +} + +function unsupportedField(field: string, detail: string): never { + throw new ApiError({ + code: 'BAD_REQUEST', + message: `Perplexity Agent API field "${field}" is unsupported: ${detail}` + }); +} + +function validateAgentPipeFields( + pipe: Pipe, + paramsTools: PipeTool[] | undefined +) { + if (pipe.presence_penalty !== 1) { + unsupportedField( + 'presence_penalty', + 'only the BaseAI scaffold default of 1 can be omitted' + ); + } + if (pipe.frequency_penalty !== 1) { + unsupportedField( + 'frequency_penalty', + 'only the BaseAI scaffold default of 1 can be omitted' + ); + } + if (pipe.stop.length > 0) { + unsupportedField('stop', 'non-empty stop sequences are not supported'); + } + if (pipe.json) { + unsupportedField('json', 'structured output is outside this release'); + } + + const hasPipeTools = pipe.tools.length > 0; + const hasParamTools = Boolean(paramsTools?.length); + if (hasPipeTools || hasParamTools) { + unsupportedField( + 'tools', + 'custom tools cannot preserve Agent thought signatures yet' + ); + } + if (pipe.tool_choice !== 'auto') { + unsupportedField( + 'tool_choice', + 'only the BaseAI scaffold default of "auto" can be omitted' + ); + } + if (pipe.parallel_tool_calls !== true) { + unsupportedField( + 'parallel_tool_calls', + 'only the BaseAI scaffold default of true can be omitted' + ); + } + if (pipe.max_tokens < 1) { + unsupportedField('max_tokens', 'the value must be at least 1'); + } +} + +function convertContentPart( + part: ContentType, + messageIndex: number, + partIndex: number +): AgentInputContentPart { + const path = `messages[${messageIndex}].content[${partIndex}]`; + if (part.type === 'text' || part.type === 'input_text') { + if (typeof part.text !== 'string') { + unsupportedField(`${path}.text`, 'text content must be a string'); + } + return { type: 'input_text', text: part.text }; + } + + if (part.type === 'image_url' || part.type === 'input_image') { + const image = part.image_url as { url?: string } | string | undefined; + const imageURL = typeof image === 'string' ? image : image?.url; + if (!imageURL) { + unsupportedField( + `${path}.image_url`, + 'an HTTPS URL or data URI is required' + ); + } + return { type: 'input_image', image_url: imageURL }; + } + + unsupportedField(`${path}.type`, `content type "${part.type}"`); +} + +function convertMessage( + message: ProviderMessage, + messageIndex: number +): AgentInputMessage { + const path = `messages[${messageIndex}]`; + if (!['system', 'user', 'assistant'].includes(message.role)) { + unsupportedField( + `${path}.role`, + `role "${message.role}" cannot be replayed` + ); + } + if (message.function_call !== undefined) { + unsupportedField( + `${path}.function_call`, + 'function history is unsupported' + ); + } + if (message.tool_calls !== undefined) { + unsupportedField( + `${path}.tool_calls`, + 'tool-call history is unsupported' + ); + } + if (message.tool_call_id !== undefined) { + unsupportedField(`${path}.tool_call_id`, 'tool history is unsupported'); + } + if (message.name !== undefined) { + unsupportedField(`${path}.name`, 'named messages are unsupported'); + } + if (message.content === null || message.content === undefined) { + unsupportedField(`${path}.content`, 'message content is required'); + } + + const content = Array.isArray(message.content) + ? message.content.map((part, partIndex) => + convertContentPart(part, messageIndex, partIndex) + ) + : message.content; + + return { + type: 'message', + role: message.role as AgentInputMessage['role'], + content + }; +} + +export function buildPerplexityAgentRequest({ + pipe, + messages, + stream, + paramsTools +}: { + pipe: Pipe; + messages: ProviderMessage[]; + stream: boolean; + paramsTools?: PipeTool[]; +}): PerplexityAgentRequest { + validateAgentPipeFields(pipe, paramsTools); + + return { + preset: getAgentPreset(pipe.model), + input: messages.map(convertMessage), + max_output_tokens: pipe.max_tokens, + stream, + temperature: pipe.temperature, + top_p: pipe.top_p + }; +} + +export function getPerplexityAgentURL( + baseURL: string, + endpoint = PerplexityAIApiConfig.agentResponse || '/v1/agent' +): string { + return `${baseURL.replace(/\/+$/, '')}/${endpoint.replace(/^\/+/, '')}`; +} + +function getSearchResultURLs(output: AgentOutputItem[]): Map { + const urls = new Map(); + for (const item of output) { + if (item.type !== 'search_results') continue; + for (const result of (item as AgentSearchResultsOutput).results || []) { + if (result.url) urls.set(String(result.id), result.url); + } + } + return urls; +} + +function citationIdFromMarker(marker: string): string | undefined { + return marker.match(/^\[(?:web:)?(\d+)\]$/i)?.[1]; +} + +function extractTextAndCitations(output: AgentOutputItem[]): { + text: string; + citations: CitationSource[]; +} { + const searchResultURLs = getSearchResultURLs(output); + const textParts: string[] = []; + const citations: CitationSource[] = []; + const seenCitations = new Set(); + let textOffset = 0; + + const addCitation = (citation: CitationSource) => { + const key = `${citation.startIndex ?? ''}:${citation.endIndex ?? ''}:${citation.uri ?? ''}`; + if (seenCitations.has(key)) return; + seenCitations.add(key); + citations.push(citation); + }; + + for (const item of output) { + if (item.type !== 'message') continue; + const message = item as AgentMessageOutput; + if (message.role !== 'assistant') { + throw new Error( + 'Invalid Perplexity Agent response: non-assistant output' + ); + } + + for (const part of message.content || []) { + if (part.type !== 'output_text' || typeof part.text !== 'string') { + throw new Error( + 'Invalid Perplexity Agent response: malformed output text' + ); + } + textParts.push(part.text); + + for (const annotation of part.annotations || []) { + const start = annotation.start_index; + const end = annotation.end_index; + const marker = + typeof start === 'number' && typeof end === 'number' + ? part.text.slice(start, end) + : ''; + const resultId = + annotation.search_result_id ?? + annotation.source_id ?? + annotation.id ?? + citationIdFromMarker(marker); + const uri = + annotation.url ?? + annotation.uri ?? + (resultId === undefined + ? undefined + : searchResultURLs.get(String(resultId))); + + if (!uri) continue; + addCitation({ + startIndex: + typeof start === 'number' + ? textOffset + start + : undefined, + endIndex: + typeof end === 'number' ? textOffset + end : undefined, + uri + }); + } + + const markerPattern = /\[(?:web:)?(\d+)\]/gi; + for (const marker of part.text.matchAll(markerPattern)) { + const uri = searchResultURLs.get(marker[1]); + if (!uri || marker.index === undefined) continue; + addCitation({ + startIndex: textOffset + marker.index, + endIndex: textOffset + marker.index + marker[0].length, + uri + }); + } + + textOffset += part.text.length; + } + } + + return { text: textParts.join(''), citations }; +} + +function responseFailureMessage(response: PerplexityAgentResponse): string { + return ( + response.error?.message || + `Perplexity Agent response ended with status "${response.status || 'unknown'}"` + ); +} + +export function transformPerplexityAgentResponse( + response: PerplexityAgentResponse, + requestedModel: string +): ChatCompletionResponse & { provider: string } { + const status = response.status; + const isTokenLimit = + status === 'incomplete' && + response.incomplete_details?.reason === 'max_output_tokens'; + if (status !== 'completed' && !isTokenLimit) { + throw new Error(responseFailureMessage(response)); + } + if (!response.id || typeof response.created_at !== 'number') { + throw new Error('Invalid Perplexity Agent response: missing identity'); + } + if (!Array.isArray(response.output)) { + throw new Error('Invalid Perplexity Agent response: missing output'); + } + + const { text, citations } = extractTextAndCitations(response.output); + if (!text) { + throw new Error( + 'Invalid Perplexity Agent response: empty completed output' + ); + } + + const inputTokens = response.usage?.input_tokens ?? 0; + const outputTokens = response.usage?.output_tokens ?? 0; + const totalTokens = + response.usage?.total_tokens ?? inputTokens + outputTokens; + + return { + id: response.id, + object: AGENT_OBJECT, + created: response.created_at, + model: requestedModel, + provider: PERPLEXITY, + choices: [ + { + message: { + role: 'assistant', + content: text, + ...(citations.length > 0 && { + citationMetadata: { citationSources: citations } + }) + }, + index: 0, + logprobs: null, + finish_reason: isTokenLimit ? 'length' : 'stop' + } + ], + usage: { + prompt_tokens: inputTokens, + completion_tokens: outputTokens, + total_tokens: totalTokens + } + }; +} + +function normalizeStreamEvent(event: AgentStreamEvent): AgentStreamEvent { + if (event.event && event.data) { + return { + ...event.data, + type: event.data.type || event.event + }; + } + return event; +} + +function streamError(event: AgentStreamEvent): Error { + const message = + event.error?.message || + event.response?.error?.message || + `Perplexity Agent stream ended with event "${event.type || 'unknown'}"`; + return new Error(message); +} + +function makeStreamChunk({ + id, + created, + model, + delta, + finishReason +}: { + id: string; + created: number; + model: string; + delta: Record; + finishReason: string | null; +}) { + return `data: ${JSON.stringify({ + id, + object: AGENT_STREAM_OBJECT, + created, + model, + provider: PERPLEXITY, + choices: [ + { + delta, + index: 0, + finish_reason: finishReason + } + ] + })}\n\n`; +} + +export function transformPerplexityAgentStream( + response: Response, + requestedModel: string, + abortController = new AbortController() +): ReadableStream { + const events = Stream.fromSSEResponse( + response, + abortController + ); + const encoder = new TextEncoder(); + + return new ReadableStream({ + async start(controller) { + let responseId: string | undefined; + let createdAt: number | undefined; + let lastSequence = -1; + let terminalSeen = false; + let roleEmitted = false; + let textSeen = false; + let lastOutputIndex = -1; + let lastContentIndex = -1; + const itemByPosition = new Map(); + + try { + for await (const rawEvent of events) { + if ( + !rawEvent || + typeof rawEvent !== 'object' || + Array.isArray(rawEvent) + ) { + throw new Error( + 'Perplexity Agent stream event is malformed' + ); + } + const event = normalizeStreamEvent(rawEvent); + if (typeof event.type !== 'string') { + throw new Error( + 'Perplexity Agent stream event is malformed' + ); + } + if (terminalSeen) { + throw new Error( + 'Perplexity Agent stream emitted data after a terminal event' + ); + } + + if (typeof event.sequence_number === 'number') { + if (event.sequence_number <= lastSequence) { + throw new Error( + 'Perplexity Agent stream sequence is duplicate or out of order' + ); + } + lastSequence = event.sequence_number; + } + + if ( + event.type === 'response.created' || + event.type === 'response.in_progress' + ) { + const id = event.response?.id; + const created = event.response?.created_at; + if (!id || typeof created !== 'number') { + throw new Error( + 'Perplexity Agent stream identity event is malformed' + ); + } + if (responseId && responseId !== id) { + throw new Error( + 'Perplexity Agent stream response identity changed' + ); + } + if (createdAt !== undefined && createdAt !== created) { + throw new Error( + 'Perplexity Agent stream response identity changed' + ); + } + responseId = id; + createdAt = created; + continue; + } + + if (event.type === 'response.output_text.delta') { + if (!responseId || createdAt === undefined) { + throw new Error( + 'Perplexity Agent stream emitted text before identity' + ); + } + if ( + typeof event.delta !== 'string' || + typeof event.item_id !== 'string' || + typeof event.output_index !== 'number' || + typeof event.content_index !== 'number' + ) { + throw new Error( + 'Perplexity Agent stream text delta is malformed' + ); + } + + const isEarlierPosition = + event.output_index < lastOutputIndex || + (event.output_index === lastOutputIndex && + event.content_index < lastContentIndex); + if (isEarlierPosition) { + throw new Error( + 'Perplexity Agent stream text indexes are out of order' + ); + } + + const position = `${event.output_index}:${event.content_index}`; + const knownItem = itemByPosition.get(position); + if (knownItem && knownItem !== event.item_id) { + throw new Error( + 'Perplexity Agent stream item identity changed' + ); + } + itemByPosition.set(position, event.item_id); + lastOutputIndex = event.output_index; + lastContentIndex = event.content_index; + + if (!roleEmitted) { + controller.enqueue( + encoder.encode( + makeStreamChunk({ + id: responseId, + created: createdAt, + model: requestedModel, + delta: { + role: 'assistant', + content: '' + }, + finishReason: null + }) + ) + ); + roleEmitted = true; + } + + controller.enqueue( + encoder.encode( + makeStreamChunk({ + id: responseId, + created: createdAt, + model: requestedModel, + delta: { content: event.delta }, + finishReason: null + }) + ) + ); + textSeen ||= event.delta.length > 0; + continue; + } + + if ( + event.type === 'response.failed' || + event.type === 'response.incomplete' || + event.type === 'response.cancelled' + ) { + terminalSeen = true; + throw streamError(event); + } + + if (event.type === 'response.completed') { + terminalSeen = true; + if ( + event.response?.status !== 'completed' || + !responseId || + createdAt === undefined || + event.response.id !== responseId || + event.response.created_at !== createdAt + ) { + throw new Error( + 'Perplexity Agent stream completion event is malformed' + ); + } + continue; + } + } + + if (abortController.signal.aborted) return; + if (!terminalSeen) { + throw new Error( + 'Perplexity Agent stream ended before response.completed' + ); + } + if (!textSeen || !responseId || createdAt === undefined) { + throw new Error( + 'Perplexity Agent stream completed without output text' + ); + } + + controller.enqueue( + encoder.encode( + makeStreamChunk({ + id: responseId, + created: createdAt, + model: requestedModel, + delta: {}, + finishReason: 'stop' + }) + ) + ); + controller.enqueue(encoder.encode('data: [DONE]\n\n')); + controller.close(); + } catch (error) { + controller.error( + error instanceof Error + ? error + : new Error('Perplexity Agent stream failed') + ); + } + }, + cancel() { + abortController.abort(); + } + }); +} + +async function getSafeAgentError(response: Response): Promise { + try { + const body = (await response.json()) as { error?: AgentErrorInfo }; + return body.error?.message || `HTTP ${response.status}`; + } catch { + return `HTTP ${response.status}`; + } +} + +export async function callPerplexityAgent( + { + pipe, + messages, + llmApiKey, + stream, + paramsTools + }: { + pipe: Pipe; + messages: ProviderMessage[]; + llmApiKey: string; + stream: boolean; + paramsTools?: PipeTool[]; + }, + options: AgentCallOptions = {} +) { + const request = buildPerplexityAgentRequest({ + pipe, + messages, + stream, + paramsTools + }); + const baseURL = + options.baseURL || + PerplexityAIApiConfig.baseURL || + 'https://api.perplexity.ai'; + const url = getPerplexityAgentURL(baseURL); + const abortController = new AbortController(); + const fetcher = options.fetcher || fetch; + const response = await fetcher(url, { + method: 'POST', + headers: PerplexityAIApiConfig.headers(llmApiKey), + body: JSON.stringify(request), + signal: abortController.signal + }); + + if (!response.ok) { + throw new ApiError({ + code: 'BAD_REQUEST', + message: await getSafeAgentError(response) + }); + } + + if (stream) { + return transformPerplexityAgentStream( + response, + pipe.model, + abortController + ); + } + + let body: PerplexityAgentResponse; + try { + body = (await response.json()) as PerplexityAgentResponse; + } catch { + throw new Error('Invalid Perplexity Agent response: malformed JSON'); + } + return transformPerplexityAgentResponse(body, pipe.model); +} diff --git a/packages/baseai/src/dev/providers/perplexity/agentStream.test.ts b/packages/baseai/src/dev/providers/perplexity/agentStream.test.ts new file mode 100644 index 00000000..fed8d668 --- /dev/null +++ b/packages/baseai/src/dev/providers/perplexity/agentStream.test.ts @@ -0,0 +1,305 @@ +import { describe, expect, it, vi } from 'vitest'; +import { transformPerplexityAgentStream } from './agentResponse'; + +const created = { + type: 'response.created', + sequence_number: 0, + response: { + id: 'resp_stream_123', + created_at: 1787928000, + status: 'in_progress' + } +}; + +const delta = (text: string, sequenceNumber = 1) => ({ + type: 'response.output_text.delta', + sequence_number: sequenceNumber, + item_id: 'msg_stream_123', + output_index: 0, + content_index: 0, + delta: text +}); + +const completed = (sequenceNumber = 2) => ({ + type: 'response.completed', + sequence_number: sequenceNumber, + response: { + id: 'resp_stream_123', + created_at: 1787928000, + status: 'completed' + } +}); + +function sse( + events: object[], + lineEnding: '\n' | '\r\n' = '\n', + includeDone = true +): string { + const body = events + .map( + event => `data: ${JSON.stringify(event)}${lineEnding}${lineEnding}` + ) + .join(''); + return includeDone ? `${body}data: [DONE]${lineEnding}${lineEnding}` : body; +} + +function responseFromChunks(chunks: Uint8Array[]): Response { + return new Response( + new ReadableStream({ + start(controller) { + for (const chunk of chunks) controller.enqueue(chunk); + controller.close(); + } + }), + { status: 200, headers: { 'content-type': 'text/event-stream' } } + ); +} + +function responseFromText(text: string): Response { + return responseFromChunks([new TextEncoder().encode(text)]); +} + +async function readText(stream: ReadableStream): Promise { + const reader = stream.getReader(); + const decoder = new TextDecoder(); + let text = ''; + while (true) { + const { done, value } = await reader.read(); + if (done) return text + decoder.decode(); + text += decoder.decode(value, { stream: true }); + } +} + +describe('Perplexity Agent stream reducer', () => { + it.each(['\n', '\r\n'] as const)( + 'accepts %j SSE delimiters and emits one terminal and one DONE', + async lineEnding => { + const output = await readText( + transformPerplexityAgentStream( + responseFromText( + sse([created, delta('Hello'), completed()], lineEnding) + ), + 'perplexity:fast' + ) + ); + + const frames = output + .split('\n\n') + .filter(Boolean) + .map(frame => frame.replace(/^data: /, '')); + expect(frames).toHaveLength(4); + expect(JSON.parse(frames[0]).choices[0]).toEqual({ + delta: { role: 'assistant', content: '' }, + index: 0, + finish_reason: null + }); + expect(JSON.parse(frames[1]).choices[0].delta).toEqual({ + content: 'Hello' + }); + expect(JSON.parse(frames[2]).choices[0]).toEqual({ + delta: {}, + index: 0, + finish_reason: 'stop' + }); + expect(frames[3]).toBe('[DONE]'); + expect(output.match(/finish_reason":"stop"/g)).toHaveLength(1); + expect(output.match(/data: \[DONE\]/g)).toHaveLength(1); + } + ); + + it('preserves fragmented UTF-8 text', async () => { + const source = new TextEncoder().encode( + sse([created, delta('café 🛰️'), completed()]) + ); + const splitAt = source.findIndex(byte => byte === 0xf0); + const response = responseFromChunks([ + source.slice(0, splitAt + 1), + source.slice(splitAt + 1, splitAt + 3), + source.slice(splitAt + 3) + ]); + + const output = await readText( + transformPerplexityAgentStream(response, 'perplexity:medium') + ); + expect(output).toContain('café 🛰️'); + expect(output).not.toContain('�'); + }); + + it('accepts multiline SSE data fields', async () => { + const createdJSON = JSON.stringify(created); + const splitAt = createdJSON.indexOf(',"response"') + 1; + const multilineCreated = + `data: ${createdJSON.slice(0, splitAt)}\n` + + `data: ${createdJSON.slice(splitAt)}\n\n`; + const response = responseFromText( + multilineCreated + sse([delta('Hello'), completed()]) + ); + + await expect( + readText(transformPerplexityAgentStream(response, 'perplexity:low')) + ).resolves.toContain('Hello'); + }); + + it('accepts named SSE events whose data omits type', async () => { + const withoutType = ({ + type: _type, + ...event + }: T) => event; + const source = + `event: response.created\ndata: ${JSON.stringify(withoutType(created))}\n\n` + + `event: response.output_text.delta\ndata: ${JSON.stringify( + withoutType(delta('Named event')) + )}\n\n` + + `event: response.completed\ndata: ${JSON.stringify( + withoutType(completed()) + )}\n\n` + + 'data: [DONE]\n\n'; + + await expect( + readText( + transformPerplexityAgentStream( + responseFromText(source), + 'perplexity:high' + ) + ) + ).resolves.toContain('Named event'); + }); + + it('redacts malformed SSE content from logging and errors', async () => { + const consoleError = vi + .spyOn(console, 'error') + .mockImplementation(() => undefined); + const response = responseFromText( + 'data: {"private_prompt":"do not log"\n\n' + ); + + await expect( + readText( + transformPerplexityAgentStream(response, 'perplexity:fast') + ) + ).rejects.toThrow('Could not parse SSE message as JSON'); + expect(consoleError).not.toHaveBeenCalled(); + consoleError.mockRestore(); + }); + + it.each(['{}', '42'])( + 'rejects structurally malformed SSE data: %s', + async data => { + const response = responseFromText(`data: ${data}\n\n`); + await expect( + readText( + transformPerplexityAgentStream(response, 'perplexity:fast') + ) + ).rejects.toThrow('stream event is malformed'); + } + ); + + it.each([ + { + type: 'response.failed', + sequence_number: 1, + error: { message: 'upstream failed' } + }, + { + type: 'response.incomplete', + sequence_number: 1, + response: { error: { message: 'incomplete response' } } + } + ])('fails on $type without success terminal output', async terminal => { + const response = responseFromText(sse([created, terminal])); + await expect( + readText( + transformPerplexityAgentStream(response, 'perplexity:fast') + ) + ).rejects.toThrow(); + }); + + it('fails on duplicate terminal events', async () => { + const response = responseFromText( + sse([created, delta('Hello'), completed(), completed(3)]) + ); + await expect( + readText( + transformPerplexityAgentStream(response, 'perplexity:fast') + ) + ).rejects.toThrow('after a terminal event'); + }); + + it('fails on out-of-order terminal sequences', async () => { + const response = responseFromText( + sse([created, delta('Hello', 3), completed(2)]) + ); + await expect( + readText( + transformPerplexityAgentStream(response, 'perplexity:fast') + ) + ).rejects.toThrow('duplicate or out of order'); + }); + + it('fails on out-of-order output indexes', async () => { + const laterOutput = { + ...delta('later', 1), + output_index: 1, + item_id: 'msg_later' + }; + const earlierOutput = { + ...delta('earlier', 2), + output_index: 0 + }; + const response = responseFromText( + sse([created, laterOutput, earlierOutput, completed(3)]) + ); + await expect( + readText( + transformPerplexityAgentStream(response, 'perplexity:fast') + ) + ).rejects.toThrow('text indexes are out of order'); + }); + + it('fails when response identity changes', async () => { + const changedIdentity = { + type: 'response.in_progress', + sequence_number: 1, + response: { + id: 'resp_stream_123', + created_at: 1787928001, + status: 'in_progress' + } + }; + const response = responseFromText(sse([created, changedIdentity])); + await expect( + readText( + transformPerplexityAgentStream(response, 'perplexity:fast') + ) + ).rejects.toThrow('response identity changed'); + }); + + it('fails on premature EOF and premature DONE', async () => { + const prematureEOF = responseFromText( + sse([created, delta('partial')], '\n', false) + ); + await expect( + readText( + transformPerplexityAgentStream(prematureEOF, 'perplexity:fast') + ) + ).rejects.toThrow('before response.completed'); + + const prematureDone = responseFromText( + sse([created, delta('partial')]) + ); + await expect( + readText( + transformPerplexityAgentStream(prematureDone, 'perplexity:fast') + ) + ).rejects.toThrow('before response.completed'); + }); + + it('fails when a completed stream contains no text', async () => { + const response = responseFromText(sse([created, completed(1)])); + await expect( + readText( + transformPerplexityAgentStream(response, 'perplexity:fast') + ) + ).rejects.toThrow('without output text'); + }); +}); diff --git a/packages/baseai/src/dev/providers/perplexity/api.ts b/packages/baseai/src/dev/providers/perplexity/api.ts index 59b5b849..7c8c125c 100644 --- a/packages/baseai/src/dev/providers/perplexity/api.ts +++ b/packages/baseai/src/dev/providers/perplexity/api.ts @@ -8,7 +8,8 @@ const PerplexityAIApiConfig: ProviderAPIConfig = { accept: 'application/json' }; }, - chatComplete: '/chat/completions' + chatComplete: '/chat/completions', + agentResponse: '/v1/agent' }; export default PerplexityAIApiConfig; diff --git a/packages/baseai/src/dev/providers/perplexity/catalog.node.test.ts b/packages/baseai/src/dev/providers/perplexity/catalog.node.test.ts new file mode 100644 index 00000000..3c6ea45f --- /dev/null +++ b/packages/baseai/src/dev/providers/perplexity/catalog.node.test.ts @@ -0,0 +1,65 @@ +import { readFile } from 'node:fs/promises'; +import { fileURLToPath } from 'node:url'; +import { describe, expect, it } from 'vitest'; +import { + PERPLEXITY as cliPerplexity, + modelsByProvider as cliModels +} from '@/data/models'; +import { + PERPLEXITY as devPerplexity, + modelsByProvider as devModels +} from '@/dev/data/models'; +import { + PERPLEXITY as corePerplexity, + modelsByProvider as coreModels +} from '../../../../../core/src/data/models'; + +const expectedIds = [ + 'fast', + 'low', + 'medium', + 'high', + 'llama-3.1-sonar-huge-128k-online', + 'llama-3.1-sonar-large-128k-online', + 'llama-3.1-sonar-small-128k-online', + 'llama-3.1-sonar-large-128k-chat', + 'llama-3.1-sonar-small-128k-chat' +]; + +describe('Perplexity model catalogs', () => { + it('keeps all three catalog copies in parity', () => { + const catalogs = [ + cliModels[cliPerplexity], + devModels[devPerplexity], + coreModels[corePerplexity] + ]; + for (const catalog of catalogs) { + expect(catalog.map(model => model.id)).toEqual(expectedIds); + } + }); + + it('represents dynamic Agent preset pricing explicitly', () => { + for (const catalog of [ + cliModels[cliPerplexity], + devModels[devPerplexity], + coreModels[corePerplexity] + ]) { + for (const model of catalog.slice(0, 4)) { + expect(model).toMatchObject({ + promptCost: null, + completionCost: null, + requestCost: null + }); + } + } + }); + + it('leaves the global Pipe scaffold default unchanged', async () => { + const file = fileURLToPath( + new URL('../../../pipe/index.ts', import.meta.url) + ); + const source = await readFile(file, 'utf8'); + expect(source).toContain("model: 'openai:gpt-4o-mini'"); + expect(source).not.toContain("model: 'perplexity:fast'"); + }); +}); diff --git a/packages/baseai/src/dev/providers/perplexity/legacyStream.test.ts b/packages/baseai/src/dev/providers/perplexity/legacyStream.test.ts new file mode 100644 index 00000000..c35b9c6c --- /dev/null +++ b/packages/baseai/src/dev/providers/perplexity/legacyStream.test.ts @@ -0,0 +1,82 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { OPEN_AI, PERPLEXITY } from '@/dev/data/models'; +import { handleStreamingMode } from '@/dev/utils/provider-handlers/response-handler-utils'; +import { PerplexityAIChatCompleteStreamChunkTransform } from './chatComplete'; + +async function readText(stream: ReadableStream): Promise { + const reader = stream.getReader(); + const decoder = new TextDecoder(); + let text = ''; + while (true) { + const { done, value } = await reader.read(); + if (done) return text + decoder.decode(); + text += decoder.decode(value, { stream: true }); + } +} + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe('existing provider streaming regressions', () => { + it('preserves the legacy Perplexity Chat Completions chunk envelope', async () => { + vi.spyOn(Date, 'now').mockReturnValue(1787928000000); + const upstreamChunk = { + id: 'legacy-123', + object: 'chat.completion.chunk', + created: 1787927999, + model: 'llama-3.1-sonar-small-128k-chat', + usage: { + prompt_tokens: 0, + completion_tokens: 0, + total_tokens: 0 + }, + choices: [ + { + message: { role: 'assistant', content: '' }, + delta: { role: 'assistant', content: 'Hello' }, + index: 0, + finish_reason: null + } + ] + }; + const upstream = `data: ${JSON.stringify(upstreamChunk)}\r\n\r\n`; + const stream = await handleStreamingMode( + new Response(upstream), + PERPLEXITY, + PerplexityAIChatCompleteStreamChunkTransform, + 'https://api.perplexity.ai/chat/completions' + ); + + expect(await readText(stream)).toBe( + `data: ${JSON.stringify({ + id: 'legacy-123', + object: 'chat.completion.chunk', + created: 1787928000, + model: 'llama-3.1-sonar-small-128k-chat', + provider: 'Perplexity', + choices: [ + { + delta: { role: 'assistant', content: 'Hello' }, + index: 0, + finish_reason: null + } + ] + })}\n\n` + ); + }); + + it('passes another existing provider stream through byte for byte', async () => { + const upstream = + 'data: {"id":"openai-1","delta":"one"}\n\n' + + 'data: {"id":"openai-1","delta":"two"}\n\n'; + const stream = await handleStreamingMode( + new Response(upstream), + OPEN_AI, + undefined, + 'https://api.openai.com/v1/chat/completions' + ); + + expect(await readText(stream)).toBe(upstream); + }); +}); diff --git a/packages/baseai/src/dev/providers/perplexity/perplexity.test.ts b/packages/baseai/src/dev/providers/perplexity/perplexity.test.ts new file mode 100644 index 00000000..bb1d274b --- /dev/null +++ b/packages/baseai/src/dev/providers/perplexity/perplexity.test.ts @@ -0,0 +1,89 @@ +import { describe, expect, it } from 'vitest'; +import type { Pipe } from 'types/pipe'; +import { validateRequestBody } from '@/dev/routes/v1/pipes/run'; +import { toOldPipeFormat } from '@/utils/to-old-pipe-format'; +import { + PERPLEXITY_AGENT_PRESETS, + buildPerplexityAgentRequest, + getPerplexityTransport +} from './agentResponse'; + +const legacyModels = [ + 'perplexity:llama-3.1-sonar-huge-128k-online', + 'perplexity:llama-3.1-sonar-large-128k-online', + 'perplexity:llama-3.1-sonar-small-128k-online', + 'perplexity:llama-3.1-sonar-large-128k-chat', + 'perplexity:llama-3.1-sonar-small-128k-chat' +] as const; + +function scaffoldPipe(model: Pipe['model']): Pipe { + return { + name: 'perplexity-agent', + description: 'Perplexity Agent API example', + status: 'private', + model, + stream: true, + json: false, + store: true, + moderate: true, + top_p: 1, + max_tokens: 1000, + temperature: 0.7, + presence_penalty: 1, + frequency_penalty: 1, + stop: [], + tool_choice: 'auto', + parallel_tool_calls: true, + messages: [{ role: 'system', content: 'Be concise.' }], + variables: [], + memory: [], + tools: [] + }; +} + +describe('Perplexity transport dispatch', () => { + it.each(legacyModels)('keeps %s on Chat Completions', model => { + expect(getPerplexityTransport(model)).toBe('chatComplete'); + }); + + it.each(PERPLEXITY_AGENT_PRESETS)( + 'routes perplexity:%s to the matching Agent preset', + preset => { + const model = `perplexity:${preset}` as Pipe['model']; + expect(getPerplexityTransport(model)).toBe('agentResponse'); + expect( + buildPerplexityAgentRequest({ + pipe: scaffoldPipe(model), + messages: [{ role: 'user', content: 'Hello' }], + stream: false + }).preset + ).toBe(preset); + } + ); +}); + +describe('Perplexity model string compatibility', () => { + it.each(PERPLEXITY_AGENT_PRESETS)( + 'preserves perplexity:%s through Pipe parsing and toOldPipeFormat', + preset => { + const model = `perplexity:${preset}` as Pipe['model']; + const pipe = scaffoldPipe(model); + const parsed = validateRequestBody({ + pipe, + stream: false, + messages: [], + llmApiKey: 'test-key' + }); + expect(parsed.pipe.model).toBe(model); + + const oldPipe = toOldPipeFormat(pipe); + expect(oldPipe.config.model).toMatchObject({ + name: preset, + provider: 'Perplexity' + }); + expect( + `${oldPipe.config.model.provider.toLowerCase()}:${oldPipe.config.model.name}` + ).toBe(model); + } + ); +}); diff --git a/packages/baseai/src/dev/routes/v1/pipes/run.ts b/packages/baseai/src/dev/routes/v1/pipes/run.ts index c58d0db2..aaa8d97f 100644 --- a/packages/baseai/src/dev/routes/v1/pipes/run.ts +++ b/packages/baseai/src/dev/routes/v1/pipes/run.ts @@ -51,7 +51,7 @@ const RequestBodySchema = z.object({ type RequestBody = z.infer; // Helper functions -const validateRequestBody = (body: unknown): RequestBody => { +export const validateRequestBody = (body: unknown): RequestBody => { const result = RequestBodySchema.safeParse(body); if (!result.success) { throw new ApiErrorZod({ diff --git a/packages/baseai/src/dev/utils/stream/stream.ts b/packages/baseai/src/dev/utils/stream/stream.ts index dcdfb739..1086cbcc 100644 --- a/packages/baseai/src/dev/utils/stream/stream.ts +++ b/packages/baseai/src/dev/utils/stream/stream.ts @@ -47,16 +47,13 @@ export class Stream implements AsyncIterable { try { data = JSON.parse(sse.data); - } catch (e) { - console.error( - `Could not parse message into JSON:`, - sse.data + } catch { + throw new Error( + 'Could not parse SSE message as JSON' ); - console.error(`From chunk:`, sse.raw); - throw e; } - if (data && data.error) { + if (data && data.error && !data.type) { throw new Error(data.error); } @@ -65,13 +62,10 @@ export class Stream implements AsyncIterable { let data; try { data = JSON.parse(sse.data); - } catch (e) { - console.error( - `Could not parse message into JSON:`, - sse.data + } catch { + throw new Error( + 'Could not parse SSE message as JSON' ); - console.error(`From chunk:`, sse.raw); - throw e; } // TODO: Is this where the error should be thrown? if (sse.event == 'error') { @@ -237,6 +231,9 @@ export async function* _iterSSEMessages( const sse = sseDecoder.decode(line); if (sse) yield sse; } + + const finalEvent = sseDecoder.flush(); + if (finalEvent) yield finalEvent; } /** @@ -284,7 +281,7 @@ function findDoubleNewlineIndex(buffer: Uint8Array): number { const newline = 0x0a; // \n const carriage = 0x0d; // \r - for (let i = 0; i < buffer.length - 2; i++) { + for (let i = 0; i < buffer.length - 1; i++) { if (buffer[i] === newline && buffer[i + 1] === newline) { // \n\n return i + 2; @@ -361,6 +358,20 @@ class SSEDecoder { return null; } + + flush(): ServerSentEvent | null { + if (!this.event && !this.data.length) return null; + + const sse: ServerSentEvent = { + event: this.event, + data: this.data.join('\n'), + raw: this.chunks + }; + this.event = null; + this.data = []; + this.chunks = []; + return sse; + } } /** @@ -431,25 +442,10 @@ class LineDecoder { if (bytes == null) return ''; if (typeof bytes === 'string') return bytes; - // Node: - if (typeof Buffer !== 'undefined') { - if (bytes instanceof Buffer) { - return bytes.toString(); - } - if (bytes instanceof Uint8Array) { - return Buffer.from(bytes).toString(); - } - - throw new Error( - `Unexpected: received non-Uint8Array (${bytes.constructor.name}) stream chunk in an environment with a global "Buffer" defined, which this library assumes to be Node. Please report this error.` - ); - } - - // Browser if (typeof TextDecoder !== 'undefined') { if (bytes instanceof Uint8Array || bytes instanceof ArrayBuffer) { this.textDecoder ??= new TextDecoder('utf8'); - return this.textDecoder.decode(bytes); + return this.textDecoder.decode(bytes, { stream: true }); } throw new Error( @@ -465,6 +461,9 @@ class LineDecoder { } flush(): string[] { + const remainingText = this.textDecoder?.decode() || ''; + if (remainingText) this.buffer.push(remainingText); + if (!this.buffer.length && !this.trailingCR) { return []; } diff --git a/packages/baseai/types/model.ts b/packages/baseai/types/model.ts index 4ec94b0f..7bc15ab5 100644 --- a/packages/baseai/types/model.ts +++ b/packages/baseai/types/model.ts @@ -67,6 +67,10 @@ export type FireworksAIModels = | 'fireworks:llama-v3p3-70b-instruct'; export type PerplexityModels = + | 'perplexity:fast' + | 'perplexity:low' + | 'perplexity:medium' + | 'perplexity:high' | 'perplexity:llama-3.1-sonar-huge-128k-online' | 'perplexity:llama-3.1-sonar-large-128k-online' | 'perplexity:llama-3.1-sonar-small-128k-online' diff --git a/packages/baseai/types/providers.ts b/packages/baseai/types/providers.ts index 7e7cfff3..552ee868 100644 --- a/packages/baseai/types/providers.ts +++ b/packages/baseai/types/providers.ts @@ -98,6 +98,8 @@ export interface ProviderAPIConfig { 'stream-complete'?: string; /** The endpoint for the 'chatComplete' function. */ chatComplete?: string; + /** The endpoint for the Perplexity Agent API operation. */ + agentResponse?: string; /** The endpoint for the 'stream-chatComplete' function. */ 'stream-chatComplete'?: string; /** The endpoint for the 'embed' function. */ @@ -121,6 +123,7 @@ export interface ProviderAPIConfig { export type endpointStrings = | 'complete' | 'chatComplete' + | 'agentResponse' | 'embed' | 'rerank' | 'moderate' diff --git a/packages/baseai/vitest.edge.config.js b/packages/baseai/vitest.edge.config.js new file mode 100644 index 00000000..0fdb6c90 --- /dev/null +++ b/packages/baseai/vitest.edge.config.js @@ -0,0 +1,20 @@ +import { fileURLToPath } from 'node:url'; +import { defineConfig } from 'vitest/config'; + +export default defineConfig({ + resolve: { + alias: { + '@': fileURLToPath(new URL('./src', import.meta.url)), + types: fileURLToPath(new URL('./types', import.meta.url)) + } + }, + test: { + environment: 'edge-runtime', + globals: true, + include: ['**/*.test.ts'], + exclude: ['**/*.node.test.ts', '**/node_modules/**'], + typecheck: { + enabled: true + } + } +}); diff --git a/packages/baseai/vitest.node.config.js b/packages/baseai/vitest.node.config.js new file mode 100644 index 00000000..94f5e73a --- /dev/null +++ b/packages/baseai/vitest.node.config.js @@ -0,0 +1,20 @@ +import { fileURLToPath } from 'node:url'; +import { defineConfig } from 'vitest/config'; + +export default defineConfig({ + resolve: { + alias: { + '@': fileURLToPath(new URL('./src', import.meta.url)), + types: fileURLToPath(new URL('./types', import.meta.url)) + } + }, + test: { + environment: 'node', + globals: true, + include: ['**/*.test.ts'], + exclude: ['**/*.edge.test.ts', '**/node_modules/**'], + typecheck: { + enabled: true + } + } +}); diff --git a/packages/core/src/data/models.ts b/packages/core/src/data/models.ts index be39bcd2..641b04d7 100644 --- a/packages/core/src/data/models.ts +++ b/packages/core/src/data/models.ts @@ -16,9 +16,9 @@ export const X_AI: string = 'xAI'; interface Model { id: string; provider: string; - promptCost: number; - completionCost: number; - requestCost?: number; + promptCost: number | null; + completionCost: number | null; + requestCost?: number | null; } interface ModelsByProviderInclCosts { @@ -377,6 +377,34 @@ export const modelsByProvider: ModelsByProviderInclCosts = { }, ], [PERPLEXITY]: [ + { + id: 'fast', + provider: PERPLEXITY, + promptCost: null, + completionCost: null, + requestCost: null, + }, + { + id: 'low', + provider: PERPLEXITY, + promptCost: null, + completionCost: null, + requestCost: null, + }, + { + id: 'medium', + provider: PERPLEXITY, + promptCost: null, + completionCost: null, + requestCost: null, + }, + { + id: 'high', + provider: PERPLEXITY, + promptCost: null, + completionCost: null, + requestCost: null, + }, { id: 'llama-3.1-sonar-huge-128k-online', provider: PERPLEXITY, diff --git a/packages/core/types/model.ts b/packages/core/types/model.ts index 960cdaad..5d3a609f 100644 --- a/packages/core/types/model.ts +++ b/packages/core/types/model.ts @@ -67,6 +67,10 @@ export type FireworksAIModels = | 'fireworks:llama-v3p3-70b-instruct'; export type PerplexityModels = + | 'perplexity:fast' + | 'perplexity:low' + | 'perplexity:medium' + | 'perplexity:high' | 'perplexity:llama-3.1-sonar-huge-128k-online' | 'perplexity:llama-3.1-sonar-large-128k-online' | 'perplexity:llama-3.1-sonar-small-128k-online' diff --git a/packages/core/types/pipes.ts b/packages/core/types/pipes.ts index d4390e86..c8c39e99 100644 --- a/packages/core/types/pipes.ts +++ b/packages/core/types/pipes.ts @@ -25,9 +25,21 @@ export interface ToolCallResult { function: Function; } +export interface CitationSource { + startIndex?: number; + endIndex?: number; + uri?: string; + license?: string; +} + +export interface CitationMetadata { + citationSources?: CitationSource[]; +} + export interface Message { role: MessageRole; content: string | null; + citationMetadata?: CitationMetadata; name?: string; tool_call_id?: string; tool_calls?: ToolCallResult[]; From 30d7f786efeeb4e87d45e6296da88d9813d3dcc6 Mon Sep 17 00:00:00 2001 From: andrewmadson-pplx Date: Fri, 28 Aug 2026 15:14:30 -0700 Subject: [PATCH 2/8] docs(perplexity): add Agent example and Sonar notice --- .../docs/supported-models-and-providers.mdx | 29 +++++++++++++++-- .../nodejs/baseai/pipes/perplexity-agent.ts | 32 +++++++++++++++++++ .../nodejs/examples/pipe.perplexity-agent.ts | 22 +++++++++++++ examples/nodejs/package.json | 1 + examples/nodejs/readme.md | 1 + 5 files changed, 83 insertions(+), 2 deletions(-) create mode 100644 examples/nodejs/baseai/pipes/perplexity-agent.ts create mode 100644 examples/nodejs/examples/pipe.perplexity-agent.ts diff --git a/apps/baseai.dev/content/docs/docs/supported-models-and-providers.mdx b/apps/baseai.dev/content/docs/docs/supported-models-and-providers.mdx index 5ffae1a5..a055797a 100644 --- a/apps/baseai.dev/content/docs/docs/supported-models-and-providers.mdx +++ b/apps/baseai.dev/content/docs/docs/supported-models-and-providers.mdx @@ -7,7 +7,7 @@ tags: - llm providers - models published: 2024-09-24 -modified: 2024-09-24 +modified: 2026-08-28 --- # Supported LLM models and providers @@ -150,7 +150,32 @@ Learn more about [using Ollama models](/docs/guides/using-ollama-models) in Base ### Perplexity -| Model | Provider | Owner | Context | Cost* | +Use a Perplexity Agent API preset for new local BaseAI integrations. Presets select and evolve their underlying model and built-in search tools dynamically. + +| Agent preset | BaseAI ID | Best for | Cost | +|--------------|--------------------------------------------|--------------------------------------------------|---------| +| Fast | | Quick lookups and short summaries with citations | Dynamic | +| Low | | Everyday, light multi-step research | Dynamic | +| Medium | | Multi-hop research across many sources | Dynamic | +| High | | Deep research and broad source coverage | Dynamic | + +This first Agent integration supports system, user, and assistant text; non-streaming responses; streaming answer text; and preset search. Custom tools, tool-result replay, background runs, file outputs, and explicit Agent model selection are not supported. Citation metadata is included on non-streaming responses; streamed citation metadata is not exposed because the existing BaseAI stream chunk contract has no compatible citation field. + + +These Agent identifiers are supported by the public BaseAI local runtime. Langbase-hosted model availability and rollout are managed separately. + + + +The following legacy Perplexity Sonar identifiers use Chat Completions and are scheduled to stop working on **September 27, 2026**. Existing Pipes are not migrated automatically. Use a Perplexity Agent preset and follow the [Perplexity Sonar migration guide](https://docs.perplexity.ai/docs/agent-api/migrate-from-sonar/how-to). + +- `perplexity:llama-3.1-sonar-huge-128k-online` +- `perplexity:llama-3.1-sonar-large-128k-online` +- `perplexity:llama-3.1-sonar-small-128k-online` +- `perplexity:llama-3.1-sonar-large-128k-chat` +- `perplexity:llama-3.1-sonar-small-128k-chat` + + +| Legacy model | Provider | Owner | Context | Cost* | |----------------------------------------------------------------------------------------------------------|------------|-------|---------|--------------------------------------------| | llama-3.1-sonar-huge-128k-online
ID: | Perplexity | Meta | 127,072 | $5 prompt
$5 completion | | llama-3.1-sonar-large-128k-online
ID: | Perplexity | Meta | 127,072 | $1 prompt
$1 completion | diff --git a/examples/nodejs/baseai/pipes/perplexity-agent.ts b/examples/nodejs/baseai/pipes/perplexity-agent.ts new file mode 100644 index 00000000..c09a0280 --- /dev/null +++ b/examples/nodejs/baseai/pipes/perplexity-agent.ts @@ -0,0 +1,32 @@ +import {PipeI} from '@baseai/core'; + +const perplexityAgentPipe = (): PipeI => ({ + apiKey: process.env.LANGBASE_API_KEY!, + name: 'perplexity-agent', + description: 'Research with a Perplexity Agent API preset.', + status: 'private', + model: 'perplexity:fast', + stream: true, + json: false, + store: true, + moderate: true, + top_p: 1, + max_tokens: 1000, + temperature: 0.7, + presence_penalty: 1, + frequency_penalty: 1, + stop: [], + tool_choice: 'auto', + parallel_tool_calls: true, + messages: [ + { + role: 'system', + content: 'Answer concisely and cite sources for factual claims.', + }, + ], + variables: [], + memory: [], + tools: [], +}); + +export default perplexityAgentPipe; diff --git a/examples/nodejs/examples/pipe.perplexity-agent.ts b/examples/nodejs/examples/pipe.perplexity-agent.ts new file mode 100644 index 00000000..49e12dfc --- /dev/null +++ b/examples/nodejs/examples/pipe.perplexity-agent.ts @@ -0,0 +1,22 @@ +import 'dotenv/config'; +import {Pipe} from '@baseai/core'; +import perplexityAgentPipe from '../baseai/pipes/perplexity-agent'; + +const pipe = new Pipe(perplexityAgentPipe()); + +async function main() { + const response = await pipe.run({ + messages: [ + { + role: 'user', + content: + 'What changed in the latest JavaScript language release?', + }, + ], + }); + + console.log(response.completion); + console.log(response.choices[0]?.message.citationMetadata); +} + +main(); diff --git a/examples/nodejs/package.json b/examples/nodejs/package.json index 5be76560..d2ac131a 100644 --- a/examples/nodejs/package.json +++ b/examples/nodejs/package.json @@ -10,6 +10,7 @@ "pipe.run": "npx tsx ./examples/pipe.run.ts", "pipe.run.stream": "npx tsx ./examples/pipe.run.stream.ts", "pipe.run.stream.loop": "npx tsx ./examples/pipe.run.stream.loop.ts", + "pipe.perplexity.agent": "npx tsx ./examples/pipe.perplexity-agent.ts", "pipe.generate.text": "npx tsx ./examples/pipe.generate.text.ts", "pipe.stream.text": "npx tsx ./examples/pipe.stream.text.ts" }, diff --git a/examples/nodejs/readme.md b/examples/nodejs/readme.md index 93fc48f6..54f74a16 100644 --- a/examples/nodejs/readme.md +++ b/examples/nodejs/readme.md @@ -22,6 +22,7 @@ npx baseai pipe npm run pipe.run npm run pipe.run.stream npm run pipe.run.stream.loop +npm run pipe.perplexity.agent npm run pipe.generate.text npm run pipe.stream.text ``` From 58a44bf0b0d0e95bf8a89518af59c09e31c9c5e1 Mon Sep 17 00:00:00 2001 From: andrewmadson-pplx Date: Fri, 28 Aug 2026 15:14:33 -0700 Subject: [PATCH 3/8] chore: document Perplexity Agent API release --- .changeset/strong-agents-replace-sonar.md | 9 +++++++++ packages/baseai/CHANGELOG.md | 11 +++++++++++ packages/core/CHANGELOG.md | 7 +++++++ 3 files changed, 27 insertions(+) create mode 100644 .changeset/strong-agents-replace-sonar.md diff --git a/.changeset/strong-agents-replace-sonar.md b/.changeset/strong-agents-replace-sonar.md new file mode 100644 index 00000000..8be32e7e --- /dev/null +++ b/.changeset/strong-agents-replace-sonar.md @@ -0,0 +1,9 @@ +--- +'baseai': minor +'@baseai/core': minor +--- + +Add Perplexity Agent API support with the `fast`, `low`, `medium`, and `high` +presets for local BaseAI Pipes. The existing Perplexity Sonar identifiers remain +on Chat Completions but are deprecated ahead of their scheduled September 27, +2026 shutdown; existing Pipes are not migrated automatically. diff --git a/packages/baseai/CHANGELOG.md b/packages/baseai/CHANGELOG.md index d1b20baa..b90cbddc 100644 --- a/packages/baseai/CHANGELOG.md +++ b/packages/baseai/CHANGELOG.md @@ -1,5 +1,16 @@ # baseai +## Unreleased + +### Minor Changes + +- Add local Perplexity Agent API support with the `fast`, `low`, `medium`, and + `high` presets, including text streaming and citation mapping. +- Deprecate the five legacy `perplexity:llama-3.1-sonar-*` identifiers ahead + of their scheduled September 27, 2026 shutdown. Existing Pipes are not + migrated automatically; follow the [Perplexity Sonar migration + guide](https://docs.perplexity.ai/docs/agent-api/migrate-from-sonar/how-to). + ## 0.9.44 ### Patch Changes diff --git a/packages/core/CHANGELOG.md b/packages/core/CHANGELOG.md index 7d6cc9b1..311101a6 100644 --- a/packages/core/CHANGELOG.md +++ b/packages/core/CHANGELOG.md @@ -1,5 +1,12 @@ # `baseai` SDK +## Unreleased + +### Minor Changes + +- Add the `perplexity:fast`, `perplexity:low`, `perplexity:medium`, and + `perplexity:high` model identifiers for local Perplexity Agent API Pipes. + ## 0.9.43 ### Patch Changes From 7895b9335806d849b7b4a614bf8dd67a05453612 Mon Sep 17 00:00:00 2001 From: andrewmadson-pplx Date: Fri, 28 Aug 2026 15:17:47 -0700 Subject: [PATCH 4/8] docs(perplexity): clarify Sonar migration notice --- .../content/docs/docs/supported-models-and-providers.mdx | 4 +++- packages/baseai/CHANGELOG.md | 2 +- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/apps/baseai.dev/content/docs/docs/supported-models-and-providers.mdx b/apps/baseai.dev/content/docs/docs/supported-models-and-providers.mdx index a055797a..f2f9084d 100644 --- a/apps/baseai.dev/content/docs/docs/supported-models-and-providers.mdx +++ b/apps/baseai.dev/content/docs/docs/supported-models-and-providers.mdx @@ -166,7 +166,9 @@ These Agent identifiers are supported by the public BaseAI local runtime. Langba -The following legacy Perplexity Sonar identifiers use Chat Completions and are scheduled to stop working on **September 27, 2026**. Existing Pipes are not migrated automatically. Use a Perplexity Agent preset and follow the [Perplexity Sonar migration guide](https://docs.perplexity.ai/docs/agent-api/migrate-from-sonar/how-to). +Sonar Chat Completions is now [**Agent API**](https://docs.perplexity.ai/docs/agent-api/quickstart). Migrate by September 27, 2026. View the [**Migration Guide**](https://docs.perplexity.ai/docs/agent-api/migrate-from-sonar/overview). + +The following legacy Perplexity Sonar identifiers remain routed through Chat Completions. Existing Pipes are not migrated automatically. - `perplexity:llama-3.1-sonar-huge-128k-online` - `perplexity:llama-3.1-sonar-large-128k-online` diff --git a/packages/baseai/CHANGELOG.md b/packages/baseai/CHANGELOG.md index b90cbddc..9ba8eefd 100644 --- a/packages/baseai/CHANGELOG.md +++ b/packages/baseai/CHANGELOG.md @@ -9,7 +9,7 @@ - Deprecate the five legacy `perplexity:llama-3.1-sonar-*` identifiers ahead of their scheduled September 27, 2026 shutdown. Existing Pipes are not migrated automatically; follow the [Perplexity Sonar migration - guide](https://docs.perplexity.ai/docs/agent-api/migrate-from-sonar/how-to). + guide](https://docs.perplexity.ai/docs/agent-api/migrate-from-sonar/overview). ## 0.9.44 From 7070498df2db1e2cca6d1ce15cfb02abe79a5165 Mon Sep 17 00:00:00 2001 From: andrewmadson-pplx Date: Fri, 28 Aug 2026 16:44:03 -0700 Subject: [PATCH 5/8] fix(core): return tool-free streams immediately --- packages/core/CHANGELOG.md | 2 + .../core/src/pipes/pipes.streaming.test.ts | 61 +++++++++++++++++++ packages/core/src/pipes/pipes.ts | 5 +- 3 files changed, 67 insertions(+), 1 deletion(-) create mode 100644 packages/core/src/pipes/pipes.streaming.test.ts diff --git a/packages/core/CHANGELOG.md b/packages/core/CHANGELOG.md index 311101a6..7634f12c 100644 --- a/packages/core/CHANGELOG.md +++ b/packages/core/CHANGELOG.md @@ -6,6 +6,8 @@ - Add the `perplexity:fast`, `perplexity:low`, `perplexity:medium`, and `perplexity:high` model identifiers for local Perplexity Agent API Pipes. +- Return tool-free streams without pre-reading them, expose Perplexity + citation metadata, and allow provider responses to omit unavailable usage. ## 0.9.43 diff --git a/packages/core/src/pipes/pipes.streaming.test.ts b/packages/core/src/pipes/pipes.streaming.test.ts new file mode 100644 index 00000000..d8849419 --- /dev/null +++ b/packages/core/src/pipes/pipes.streaming.test.ts @@ -0,0 +1,61 @@ +import {describe, expect, it, vi} from 'vitest'; +import type {Pipe as PipeConfig} from '../../types/pipes'; +import {Pipe, type RunResponseStream} from './pipes'; + +function toolFreePipe(): PipeConfig { + return { + apiKey: 'test-api-key', + name: 'perplexity-agent', + description: 'Perplexity Agent streaming test', + status: 'private', + model: 'perplexity:fast', + stream: true, + json: false, + store: true, + moderate: true, + top_p: 1, + max_tokens: 1000, + temperature: 0.7, + presence_penalty: 1, + frequency_penalty: 1, + stop: [], + tool_choice: 'auto', + parallel_tool_calls: true, + messages: [], + variables: [], + memory: [], + tools: [], + }; +} + +describe('Pipe tool-free streaming', () => { + it('returns the stream before the upstream source finishes', async () => { + let terminalReached = false; + let finishSource!: () => void; + const source = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('first chunk')); + finishSource = () => { + terminalReached = true; + controller.close(); + }; + }, + }); + const response: RunResponseStream = {stream: source, threadId: null}; + const pipe = new Pipe({...toolFreePipe(), prod: true}); + const post = vi.fn().mockResolvedValue(response); + (pipe as unknown as {request: {post: typeof post}}).request = {post}; + + const delayedTerminal = setTimeout(() => finishSource(), 100); + const result = await pipe.run({messages: [], stream: true}); + + expect(result).toBe(response); + expect(terminalReached).toBe(false); + const reader = result.stream.getReader(); + const first = await reader.read(); + expect(new TextDecoder().decode(first.value)).toBe('first chunk'); + reader.releaseLock(); + clearTimeout(delayedTerminal); + finishSource(); + }); +}); diff --git a/packages/core/src/pipes/pipes.ts b/packages/core/src/pipes/pipes.ts index 410d74d4..97fbe43d 100644 --- a/packages/core/src/pipes/pipes.ts +++ b/packages/core/src/pipes/pipes.ts @@ -49,7 +49,7 @@ export interface RunResponse { created: number; model: string; choices: ChoiceGenerate[]; - usage: Usage; + usage?: Usage; system_fingerprint: string | null; rawResponse?: { headers: Record; @@ -312,6 +312,9 @@ export class Pipe { if (Object.entries(response).length === 0) { return {} as RunResponse | RunResponseStream; } + if (stream && !this.hasTools) { + return response as RunResponseStream; + } if (!runTools) { if (!stream) { From ea4c33b22f6b8e95f734d7707d01037bffafd195 Mon Sep 17 00:00:00 2001 From: andrewmadson-pplx Date: Fri, 28 Aug 2026 16:44:09 -0700 Subject: [PATCH 6/8] fix(perplexity): harden Agent API transport --- .../baseai/src/dev/llms/call-perplexity.ts | 2 + .../perplexity/agentResponse.test.ts | 235 +++++++-------- .../dev/providers/perplexity/agentResponse.ts | 236 ++++++++------- .../src/dev/providers/perplexity/agentSSE.ts | 110 +++++++ .../providers/perplexity/agentStream.test.ts | 117 ++++---- .../fixtures/documented-agent-response.json | 36 +++ .../fixtures/documented-agent-stream.json | 60 ++++ .../routes/v1/pipes/run.perplexity.test.ts | 276 ++++++++++++++++++ .../baseai/src/dev/utils/stream/stream.ts | 59 ++-- 9 files changed, 805 insertions(+), 326 deletions(-) create mode 100644 packages/baseai/src/dev/providers/perplexity/agentSSE.ts create mode 100644 packages/baseai/src/dev/providers/perplexity/fixtures/documented-agent-response.json create mode 100644 packages/baseai/src/dev/providers/perplexity/fixtures/documented-agent-stream.json create mode 100644 packages/baseai/src/dev/routes/v1/pipes/run.perplexity.test.ts diff --git a/packages/baseai/src/dev/llms/call-perplexity.ts b/packages/baseai/src/dev/llms/call-perplexity.ts index 46dd827e..f86f1ff9 100644 --- a/packages/baseai/src/dev/llms/call-perplexity.ts +++ b/packages/baseai/src/dev/llms/call-perplexity.ts @@ -8,6 +8,7 @@ import { getPerplexityTransport } from '../providers/perplexity/agentResponse'; import { handleLlmError } from './utils'; +import { ApiError } from '../hono/errors'; import type { Message, Pipe } from 'types/pipe'; import type { ModelParams } from 'types/providers'; import type { PipeTool } from 'types/tools'; @@ -54,6 +55,7 @@ export async function callPerplexity({ transformedRequestParams }); } catch (error: any) { + if (error instanceof ApiError) throw error; handleLlmError({ error, provider: PERPLEXITY }); } } diff --git a/packages/baseai/src/dev/providers/perplexity/agentResponse.test.ts b/packages/baseai/src/dev/providers/perplexity/agentResponse.test.ts index 1c477b84..e9b908d1 100644 --- a/packages/baseai/src/dev/providers/perplexity/agentResponse.test.ts +++ b/packages/baseai/src/dev/providers/perplexity/agentResponse.test.ts @@ -8,6 +8,7 @@ import { transformPerplexityAgentResponse, type PerplexityAgentResponse } from './agentResponse'; +import documentedAgentResponse from './fixtures/documented-agent-response.json'; function scaffoldPipe(overrides: Partial = {}): Pipe { return { @@ -41,51 +42,7 @@ const messages: ProviderMessage[] = [ { role: 'assistant', content: 'I will check.' } ]; -const completedResponse: PerplexityAgentResponse = { - id: 'resp_agent_123', - object: 'response', - created_at: 1787928000, - status: 'completed', - model: 'perplexity/sonar', - output: [ - { - type: 'search_results', - results: [ - { id: 1, url: 'https://example.com/one' }, - { id: 2, url: 'https://example.com/two' } - ] - }, - { - type: 'message', - id: 'msg_123', - role: 'assistant', - status: 'completed', - content: [ - { - type: 'output_text', - text: 'Alpha[1] ', - annotations: [ - { - type: 'url_citation', - start_index: 0, - end_index: 5, - url: 'https://example.com/direct' - } - ] - }, - { - type: 'output_text', - text: 'Beta [web:2].' - } - ] - } - ], - usage: { - input_tokens: 12, - output_tokens: 8, - total_tokens: 20 - } -}; +const completedResponse = documentedAgentResponse as PerplexityAgentResponse; describe('Perplexity Agent request adapter', () => { it('maps a standard scaffold changed only to perplexity:fast', () => { @@ -121,33 +78,6 @@ describe('Perplexity Agent request adapter', () => { expect(request).not.toHaveProperty('tools'); }); - it('maps supported text and image content parts', () => { - const request = buildPerplexityAgentRequest({ - pipe: scaffoldPipe(), - messages: [ - { - role: 'user', - content: [ - { type: 'text', text: 'Describe this.' }, - { - type: 'image_url', - image_url: { url: 'https://example.com/image.png' } - } - ] - } - ], - stream: false - }); - - expect(request.input[0].content).toEqual([ - { type: 'input_text', text: 'Describe this.' }, - { - type: 'input_image', - image_url: 'https://example.com/image.png' - } - ]); - }); - it('preserves store and moderate as platform behavior without forwarding them', () => { const request = buildPerplexityAgentRequest({ pipe: scaffoldPipe({ store: false, moderate: false }), @@ -222,7 +152,7 @@ describe('Perplexity Agent request adapter', () => { ).toThrow(field); }); - it('rejects unsupported content parts by field path', () => { + it('rejects non-string content by field path', () => { expect(() => buildPerplexityAgentRequest({ pipe: scaffoldPipe(), @@ -234,7 +164,7 @@ describe('Perplexity Agent request adapter', () => { ], stream: false }) - ).toThrow('messages[0].content[0].type'); + ).toThrow('messages[0].content'); }); it('rejects non-empty tools before network I/O', async () => { @@ -348,42 +278,82 @@ describe('Perplexity Agent HTTP adapter', () => { ) ).rejects.toThrow('Preset is invalid'); }); + + it.each([ + [400, 'BAD_REQUEST'], + [422, 'BAD_REQUEST'], + [401, 'UNAUTHORIZED'], + [403, 'FORBIDDEN'], + [404, 'NOT_FOUND'], + [429, 'RATE_LIMITED'], + [500, 'INTERNAL_SERVER_ERROR'] + ] as const)('preserves HTTP %i as %s', async (status, code) => { + const fetcher = vi.fn( + async () => + new Response( + JSON.stringify({ error: { message: 'Provider error' } }), + { status } + ) + ); + + await expect( + callPerplexityAgent( + { + pipe: scaffoldPipe(), + messages, + llmApiKey: 'test-key', + stream: false + }, + { fetcher: fetcher as typeof fetch } + ) + ).rejects.toMatchObject({ status, code }); + }); + + it('distinguishes a transport failure from a provider bad request', async () => { + const fetcher = vi.fn(async () => { + throw new TypeError('private transport detail'); + }); + + await expect( + callPerplexityAgent( + { + pipe: scaffoldPipe(), + messages, + llmApiKey: 'test-key', + stream: false + }, + { fetcher: fetcher as unknown as typeof fetch } + ) + ).rejects.toMatchObject({ + status: 500, + code: 'INTERNAL_SERVER_ERROR', + message: 'Unable to reach the Perplexity Agent API' + }); + }); }); describe('Perplexity Agent response adapter', () => { it('maps the complete BaseAI response envelope and citations', () => { - expect( - transformPerplexityAgentResponse( - completedResponse, - 'perplexity:fast' - ) - ).toEqual({ - id: 'resp_agent_123', + const content = + 'A model card documents intended use and limitations.[web:4]'; + const markerStart = content.indexOf('[web:4]'); + expect(transformPerplexityAgentResponse(completedResponse)).toEqual({ + id: 'resp_docs_model_card', object: 'chat.completion', - created: 1787928000, - model: 'perplexity:fast', + created: 1784292159, + model: 'openai/gpt-5.4-mini', provider: 'Perplexity', choices: [ { message: { role: 'assistant', - content: 'Alpha[1] Beta [web:2].', + content, citationMetadata: { citationSources: [ { - startIndex: 0, - endIndex: 5, - uri: 'https://example.com/direct' - }, - { - startIndex: 5, - endIndex: 8, - uri: 'https://example.com/one' - }, - { - startIndex: 14, - endIndex: 21, - uri: 'https://example.com/two' + startIndex: markerStart, + endIndex: markerStart + '[web:4]'.length, + uri: 'https://arxiv.org/abs/1810.03993' } ] } @@ -394,22 +364,18 @@ describe('Perplexity Agent response adapter', () => { } ], usage: { - prompt_tokens: 12, - completion_tokens: 8, - total_tokens: 20 + prompt_tokens: 120, + completion_tokens: 12, + total_tokens: 132 } }); }); - it('uses a frozen zero-usage rule when Agent usage is absent', () => { + it('omits usage when Agent usage is absent', () => { const response = { ...completedResponse, usage: undefined }; - expect( - transformPerplexityAgentResponse(response, 'perplexity:low').usage - ).toEqual({ - prompt_tokens: 0, - completion_tokens: 0, - total_tokens: 0 - }); + expect(transformPerplexityAgentResponse(response)).not.toHaveProperty( + 'usage' + ); }); it('maps documented token exhaustion to length', () => { @@ -419,21 +385,17 @@ describe('Perplexity Agent response adapter', () => { incomplete_details: { reason: 'max_output_tokens' } }; expect( - transformPerplexityAgentResponse(response, 'perplexity:medium') - .choices[0].finish_reason + transformPerplexityAgentResponse(response).choices[0].finish_reason ).toBe('length'); }); it('rejects undocumented incomplete reasons', () => { expect(() => - transformPerplexityAgentResponse( - { - ...completedResponse, - status: 'incomplete', - incomplete_details: { reason: 'max_tokens' } - }, - 'perplexity:medium' - ) + transformPerplexityAgentResponse({ + ...completedResponse, + status: 'incomplete', + incomplete_details: { reason: 'max_tokens' } + }) ).toThrow(); }); @@ -444,25 +406,38 @@ describe('Perplexity Agent response adapter', () => { ['in_progress', undefined] ])('rejects %s responses outside token exhaustion', (status, error) => { expect(() => - transformPerplexityAgentResponse( - { ...completedResponse, status, error }, - 'perplexity:high' - ) + transformPerplexityAgentResponse({ + ...completedResponse, + status, + error + }) ).toThrow(); }); it('rejects malformed and empty completed responses', () => { expect(() => - transformPerplexityAgentResponse( - { ...completedResponse, id: undefined }, - 'perplexity:fast' - ) + transformPerplexityAgentResponse({ + ...completedResponse, + id: undefined + }) ).toThrow('missing identity'); expect(() => - transformPerplexityAgentResponse( - { ...completedResponse, output: [] }, - 'perplexity:fast' - ) + transformPerplexityAgentResponse({ + ...completedResponse, + output: [] + }) ).toThrow('empty completed output'); + expect(() => + transformPerplexityAgentResponse({ + ...completedResponse, + model: undefined + }) + ).toThrow('missing model'); + expect(() => + transformPerplexityAgentResponse({ + ...completedResponse, + usage: { input_tokens: 1, output_tokens: 2 } + }) + ).toThrow('malformed usage'); }); }); diff --git a/packages/baseai/src/dev/providers/perplexity/agentResponse.ts b/packages/baseai/src/dev/providers/perplexity/agentResponse.ts index 5eb16694..4d4ac497 100644 --- a/packages/baseai/src/dev/providers/perplexity/agentResponse.ts +++ b/packages/baseai/src/dev/providers/perplexity/agentResponse.ts @@ -1,15 +1,15 @@ import { PERPLEXITY } from '@/dev/data/models'; import { ApiError } from '@/dev/hono/errors'; -import { Stream } from '@/dev/utils/stream/stream'; +import type { StatusCode } from 'hono/utils/http-status'; import type { Pipe } from 'types/pipe'; import type { ChatCompletionResponse, CitationSource, - ContentType, ProviderMessage } from 'types/providers'; import type { PipeTool } from 'types/tools'; import PerplexityAIApiConfig from './api'; +import { iterateAgentSSE } from './agentSSE'; export const PERPLEXITY_AGENT_PRESETS = [ 'fast', @@ -20,14 +20,10 @@ export const PERPLEXITY_AGENT_PRESETS = [ export type PerplexityAgentPreset = (typeof PERPLEXITY_AGENT_PRESETS)[number]; -type AgentInputContentPart = - | { type: 'input_text'; text: string } - | { type: 'input_image'; image_url: string }; - interface AgentInputMessage { type: 'message'; role: 'system' | 'user' | 'assistant'; - content: string | AgentInputContentPart[]; + content: string; } export interface PerplexityAgentRequest { @@ -109,13 +105,11 @@ interface AgentStreamEvent { content_index?: number; delta?: string; error?: AgentErrorInfo; - event?: string; item_id?: string; output_index?: number; response?: PerplexityAgentResponse; sequence_number?: number; type?: string; - data?: AgentStreamEvent; } interface AgentCallOptions { @@ -207,34 +201,6 @@ function validateAgentPipeFields( } } -function convertContentPart( - part: ContentType, - messageIndex: number, - partIndex: number -): AgentInputContentPart { - const path = `messages[${messageIndex}].content[${partIndex}]`; - if (part.type === 'text' || part.type === 'input_text') { - if (typeof part.text !== 'string') { - unsupportedField(`${path}.text`, 'text content must be a string'); - } - return { type: 'input_text', text: part.text }; - } - - if (part.type === 'image_url' || part.type === 'input_image') { - const image = part.image_url as { url?: string } | string | undefined; - const imageURL = typeof image === 'string' ? image : image?.url; - if (!imageURL) { - unsupportedField( - `${path}.image_url`, - 'an HTTPS URL or data URI is required' - ); - } - return { type: 'input_image', image_url: imageURL }; - } - - unsupportedField(`${path}.type`, `content type "${part.type}"`); -} - function convertMessage( message: ProviderMessage, messageIndex: number @@ -264,20 +230,14 @@ function convertMessage( if (message.name !== undefined) { unsupportedField(`${path}.name`, 'named messages are unsupported'); } - if (message.content === null || message.content === undefined) { - unsupportedField(`${path}.content`, 'message content is required'); + if (typeof message.content !== 'string') { + unsupportedField(`${path}.content`, 'message content must be a string'); } - const content = Array.isArray(message.content) - ? message.content.map((part, partIndex) => - convertContentPart(part, messageIndex, partIndex) - ) - : message.content; - return { type: 'message', role: message.role as AgentInputMessage['role'], - content + content: message.content }; } @@ -417,8 +377,7 @@ function responseFailureMessage(response: PerplexityAgentResponse): string { } export function transformPerplexityAgentResponse( - response: PerplexityAgentResponse, - requestedModel: string + response: PerplexityAgentResponse ): ChatCompletionResponse & { provider: string } { const status = response.status; const isTokenLimit = @@ -430,6 +389,9 @@ export function transformPerplexityAgentResponse( if (!response.id || typeof response.created_at !== 'number') { throw new Error('Invalid Perplexity Agent response: missing identity'); } + if (!response.model || typeof response.model !== 'string') { + throw new Error('Invalid Perplexity Agent response: missing model'); + } if (!Array.isArray(response.output)) { throw new Error('Invalid Perplexity Agent response: missing output'); } @@ -441,16 +403,30 @@ export function transformPerplexityAgentResponse( ); } - const inputTokens = response.usage?.input_tokens ?? 0; - const outputTokens = response.usage?.output_tokens ?? 0; - const totalTokens = - response.usage?.total_tokens ?? inputTokens + outputTokens; + let usage: ChatCompletionResponse['usage']; + if (response.usage !== undefined) { + const { input_tokens, output_tokens, total_tokens } = response.usage; + if ( + ![input_tokens, output_tokens, total_tokens].every( + tokens => Number.isInteger(tokens) && (tokens as number) >= 0 + ) + ) { + throw new Error( + 'Invalid Perplexity Agent response: malformed usage' + ); + } + usage = { + prompt_tokens: input_tokens as number, + completion_tokens: output_tokens as number, + total_tokens: total_tokens as number + }; + } return { id: response.id, object: AGENT_OBJECT, created: response.created_at, - model: requestedModel, + model: response.model, provider: PERPLEXITY, choices: [ { @@ -466,24 +442,10 @@ export function transformPerplexityAgentResponse( finish_reason: isTokenLimit ? 'length' : 'stop' } ], - usage: { - prompt_tokens: inputTokens, - completion_tokens: outputTokens, - total_tokens: totalTokens - } + ...(usage && { usage }) }; } -function normalizeStreamEvent(event: AgentStreamEvent): AgentStreamEvent { - if (event.event && event.data) { - return { - ...event.data, - type: event.data.type || event.event - }; - } - return event; -} - function streamError(event: AgentStreamEvent): Error { const message = event.error?.message || @@ -523,19 +485,16 @@ function makeStreamChunk({ export function transformPerplexityAgentStream( response: Response, - requestedModel: string, abortController = new AbortController() ): ReadableStream { - const events = Stream.fromSSEResponse( - response, - abortController - ); + const events = iterateAgentSSE(response, abortController); const encoder = new TextEncoder(); return new ReadableStream({ async start(controller) { let responseId: string | undefined; let createdAt: number | undefined; + let executingModel: string | undefined; let lastSequence = -1; let terminalSeen = false; let roleEmitted = false; @@ -555,7 +514,7 @@ export function transformPerplexityAgentStream( 'Perplexity Agent stream event is malformed' ); } - const event = normalizeStreamEvent(rawEvent); + const event = rawEvent; if (typeof event.type !== 'string') { throw new Error( 'Perplexity Agent stream event is malformed' @@ -567,14 +526,20 @@ export function transformPerplexityAgentStream( ); } - if (typeof event.sequence_number === 'number') { - if (event.sequence_number <= lastSequence) { - throw new Error( - 'Perplexity Agent stream sequence is duplicate or out of order' - ); - } - lastSequence = event.sequence_number; + if ( + !Number.isInteger(event.sequence_number) || + (event.sequence_number as number) < 0 + ) { + throw new Error( + 'Perplexity Agent stream sequence is malformed' + ); } + if ((event.sequence_number as number) <= lastSequence) { + throw new Error( + 'Perplexity Agent stream sequence is duplicate or out of order' + ); + } + lastSequence = event.sequence_number as number; if ( event.type === 'response.created' || @@ -582,7 +547,13 @@ export function transformPerplexityAgentStream( ) { const id = event.response?.id; const created = event.response?.created_at; - if (!id || typeof created !== 'number') { + const model = event.response?.model; + if ( + !id || + typeof created !== 'number' || + !model || + typeof model !== 'string' + ) { throw new Error( 'Perplexity Agent stream identity event is malformed' ); @@ -597,13 +568,23 @@ export function transformPerplexityAgentStream( 'Perplexity Agent stream response identity changed' ); } + if (executingModel && executingModel !== model) { + throw new Error( + 'Perplexity Agent stream response identity changed' + ); + } responseId = id; createdAt = created; + executingModel = model; continue; } if (event.type === 'response.output_text.delta') { - if (!responseId || createdAt === undefined) { + if ( + !responseId || + createdAt === undefined || + !executingModel + ) { throw new Error( 'Perplexity Agent stream emitted text before identity' ); @@ -646,7 +627,7 @@ export function transformPerplexityAgentStream( makeStreamChunk({ id: responseId, created: createdAt, - model: requestedModel, + model: executingModel, delta: { role: 'assistant', content: '' @@ -663,7 +644,7 @@ export function transformPerplexityAgentStream( makeStreamChunk({ id: responseId, created: createdAt, - model: requestedModel, + model: executingModel, delta: { content: event.delta }, finishReason: null }) @@ -673,11 +654,7 @@ export function transformPerplexityAgentStream( continue; } - if ( - event.type === 'response.failed' || - event.type === 'response.incomplete' || - event.type === 'response.cancelled' - ) { + if (event.type === 'response.failed') { terminalSeen = true; throw streamError(event); } @@ -689,7 +666,8 @@ export function transformPerplexityAgentStream( !responseId || createdAt === undefined || event.response.id !== responseId || - event.response.created_at !== createdAt + event.response.created_at !== createdAt || + event.response.model !== executingModel ) { throw new Error( 'Perplexity Agent stream completion event is malformed' @@ -705,7 +683,12 @@ export function transformPerplexityAgentStream( 'Perplexity Agent stream ended before response.completed' ); } - if (!textSeen || !responseId || createdAt === undefined) { + if ( + !textSeen || + !responseId || + createdAt === undefined || + !executingModel + ) { throw new Error( 'Perplexity Agent stream completed without output text' ); @@ -716,7 +699,7 @@ export function transformPerplexityAgentStream( makeStreamChunk({ id: responseId, created: createdAt, - model: requestedModel, + model: executingModel, delta: {}, finishReason: 'stop' }) @@ -725,6 +708,7 @@ export function transformPerplexityAgentStream( controller.enqueue(encoder.encode('data: [DONE]\n\n')); controller.close(); } catch (error) { + abortController.abort(); controller.error( error instanceof Error ? error @@ -740,13 +724,44 @@ export function transformPerplexityAgentStream( async function getSafeAgentError(response: Response): Promise { try { - const body = (await response.json()) as { error?: AgentErrorInfo }; - return body.error?.message || `HTTP ${response.status}`; + const body = (await response.json()) as { error?: unknown }; + if ( + body.error && + typeof body.error === 'object' && + 'message' in body.error && + typeof body.error.message === 'string' + ) { + return body.error.message; + } + return `HTTP ${response.status}`; } catch { return `HTTP ${response.status}`; } } +function getAgentHttpError(status: number): { + code: + | 'BAD_REQUEST' + | 'UNAUTHORIZED' + | 'FORBIDDEN' + | 'NOT_FOUND' + | 'RATE_LIMITED' + | 'INTERNAL_SERVER_ERROR'; + status: StatusCode; +} { + if (status === 400 || status === 422) { + return { code: 'BAD_REQUEST', status: status as StatusCode }; + } + if (status === 401) return { code: 'UNAUTHORIZED', status: 401 }; + if (status === 403) return { code: 'FORBIDDEN', status: 403 }; + if (status === 404) return { code: 'NOT_FOUND', status: 404 }; + if (status === 429) return { code: 'RATE_LIMITED', status: 429 }; + if (status >= 500 && status <= 599) { + return { code: 'INTERNAL_SERVER_ERROR', status: status as StatusCode }; + } + return { code: 'BAD_REQUEST', status: status as StatusCode }; +} + export async function callPerplexityAgent( { pipe, @@ -776,26 +791,31 @@ export async function callPerplexityAgent( const url = getPerplexityAgentURL(baseURL); const abortController = new AbortController(); const fetcher = options.fetcher || fetch; - const response = await fetcher(url, { - method: 'POST', - headers: PerplexityAIApiConfig.headers(llmApiKey), - body: JSON.stringify(request), - signal: abortController.signal - }); + let response: Response; + try { + response = await fetcher(url, { + method: 'POST', + headers: PerplexityAIApiConfig.headers(llmApiKey), + body: JSON.stringify(request), + signal: abortController.signal + }); + } catch { + throw new ApiError({ + code: 'INTERNAL_SERVER_ERROR', + message: 'Unable to reach the Perplexity Agent API' + }); + } if (!response.ok) { + const mapped = getAgentHttpError(response.status); throw new ApiError({ - code: 'BAD_REQUEST', + ...mapped, message: await getSafeAgentError(response) }); } if (stream) { - return transformPerplexityAgentStream( - response, - pipe.model, - abortController - ); + return transformPerplexityAgentStream(response, abortController); } let body: PerplexityAgentResponse; @@ -804,5 +824,5 @@ export async function callPerplexityAgent( } catch { throw new Error('Invalid Perplexity Agent response: malformed JSON'); } - return transformPerplexityAgentResponse(body, pipe.model); + return transformPerplexityAgentResponse(body); } diff --git a/packages/baseai/src/dev/providers/perplexity/agentSSE.ts b/packages/baseai/src/dev/providers/perplexity/agentSSE.ts new file mode 100644 index 00000000..b7973dc7 --- /dev/null +++ b/packages/baseai/src/dev/providers/perplexity/agentSSE.ts @@ -0,0 +1,110 @@ +interface SSEMessage { + event: string | null; + data: string; +} + +function findEventBoundary( + buffer: string +): { index: number; length: number } | null { + const match = /\r\n\r\n|\n\n|\r\r/.exec(buffer); + return match ? { index: match.index, length: match[0].length } : null; +} + +function parseSSEMessage(block: string): SSEMessage | null { + let event: string | null = null; + const data: string[] = []; + + for (const line of block.split(/\r\n|\n|\r/)) { + if (!line || line.startsWith(':')) continue; + const separator = line.indexOf(':'); + const field = separator === -1 ? line : line.slice(0, separator); + const value = + separator === -1 ? '' : line.slice(separator + 1).replace(/^ /, ''); + + if (field === 'event') event = value; + if (field === 'data') data.push(value); + } + + return data.length ? { event, data: data.join('\n') } : null; +} + +function parseAgentEvent(message: SSEMessage): T { + let parsed: unknown; + try { + parsed = JSON.parse(message.data); + } catch { + throw new Error('Could not parse Perplexity Agent SSE message as JSON'); + } + + if ( + message.event && + parsed && + typeof parsed === 'object' && + !Array.isArray(parsed) && + !('type' in parsed) + ) { + return { ...parsed, type: message.event } as T; + } + return parsed as T; +} + +export async function* iterateAgentSSE( + response: Response, + abortController: AbortController +): AsyncGenerator { + if (!response.body) { + throw new Error('Perplexity Agent stream response has no body'); + } + + const reader = response.body.getReader(); + const decoder = new TextDecoder('utf-8'); + let buffer = ''; + let doneMarkerSeen = false; + const cancelReader = () => { + void reader.cancel().catch(() => undefined); + }; + abortController.signal.addEventListener('abort', cancelReader, { + once: true + }); + + const consumeBlock = (block: string): T | undefined => { + const message = parseSSEMessage(block); + if (!message) return; + if (doneMarkerSeen) { + throw new Error( + 'Perplexity Agent stream emitted data after the [DONE] marker' + ); + } + if (message.data.trim() === '[DONE]') { + doneMarkerSeen = true; + return; + } + return parseAgentEvent(message); + }; + + try { + while (!abortController.signal.aborted) { + const { done, value } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + + let boundary = findEventBoundary(buffer); + while (boundary) { + const block = buffer.slice(0, boundary.index); + buffer = buffer.slice(boundary.index + boundary.length); + const event = consumeBlock(block); + if (event !== undefined) yield event; + boundary = findEventBoundary(buffer); + } + } + + buffer += decoder.decode(); + if (buffer.trim()) { + const event = consumeBlock(buffer); + if (event !== undefined) yield event; + } + } finally { + abortController.signal.removeEventListener('abort', cancelReader); + reader.releaseLock(); + } +} diff --git a/packages/baseai/src/dev/providers/perplexity/agentStream.test.ts b/packages/baseai/src/dev/providers/perplexity/agentStream.test.ts index fed8d668..1808ba93 100644 --- a/packages/baseai/src/dev/providers/perplexity/agentStream.test.ts +++ b/packages/baseai/src/dev/providers/perplexity/agentStream.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it, vi } from 'vitest'; import { transformPerplexityAgentStream } from './agentResponse'; +import documentedAgentStream from './fixtures/documented-agent-stream.json'; const created = { type: 'response.created', @@ -7,7 +8,8 @@ const created = { response: { id: 'resp_stream_123', created_at: 1787928000, - status: 'in_progress' + status: 'in_progress', + model: 'openai/gpt-5.4-mini' } }; @@ -26,7 +28,8 @@ const completed = (sequenceNumber = 2) => ({ response: { id: 'resp_stream_123', created_at: 1787928000, - status: 'completed' + status: 'completed', + model: 'openai/gpt-5.4-mini' } }); @@ -76,10 +79,7 @@ describe('Perplexity Agent stream reducer', () => { async lineEnding => { const output = await readText( transformPerplexityAgentStream( - responseFromText( - sse([created, delta('Hello'), completed()], lineEnding) - ), - 'perplexity:fast' + responseFromText(sse(documentedAgentStream, lineEnding)) ) ); @@ -94,7 +94,7 @@ describe('Perplexity Agent stream reducer', () => { finish_reason: null }); expect(JSON.parse(frames[1]).choices[0].delta).toEqual({ - content: 'Hello' + content: 'A model card' }); expect(JSON.parse(frames[2]).choices[0]).toEqual({ delta: {}, @@ -102,6 +102,9 @@ describe('Perplexity Agent stream reducer', () => { finish_reason: 'stop' }); expect(frames[3]).toBe('[DONE]'); + for (const frame of frames.slice(0, 3)) { + expect(JSON.parse(frame).model).toBe('openai/gpt-5.4-mini'); + } expect(output.match(/finish_reason":"stop"/g)).toHaveLength(1); expect(output.match(/data: \[DONE\]/g)).toHaveLength(1); } @@ -118,9 +121,7 @@ describe('Perplexity Agent stream reducer', () => { source.slice(splitAt + 3) ]); - const output = await readText( - transformPerplexityAgentStream(response, 'perplexity:medium') - ); + const output = await readText(transformPerplexityAgentStream(response)); expect(output).toContain('café 🛰️'); expect(output).not.toContain('�'); }); @@ -136,7 +137,7 @@ describe('Perplexity Agent stream reducer', () => { ); await expect( - readText(transformPerplexityAgentStream(response, 'perplexity:low')) + readText(transformPerplexityAgentStream(response)) ).resolves.toContain('Hello'); }); @@ -156,12 +157,7 @@ describe('Perplexity Agent stream reducer', () => { 'data: [DONE]\n\n'; await expect( - readText( - transformPerplexityAgentStream( - responseFromText(source), - 'perplexity:high' - ) - ) + readText(transformPerplexityAgentStream(responseFromText(source))) ).resolves.toContain('Named event'); }); @@ -174,10 +170,10 @@ describe('Perplexity Agent stream reducer', () => { ); await expect( - readText( - transformPerplexityAgentStream(response, 'perplexity:fast') - ) - ).rejects.toThrow('Could not parse SSE message as JSON'); + readText(transformPerplexityAgentStream(response)) + ).rejects.toThrow( + 'Could not parse Perplexity Agent SSE message as JSON' + ); expect(consoleError).not.toHaveBeenCalled(); consoleError.mockRestore(); }); @@ -187,31 +183,21 @@ describe('Perplexity Agent stream reducer', () => { async data => { const response = responseFromText(`data: ${data}\n\n`); await expect( - readText( - transformPerplexityAgentStream(response, 'perplexity:fast') - ) + readText(transformPerplexityAgentStream(response)) ).rejects.toThrow('stream event is malformed'); } ); - it.each([ - { + it('fails on response.failed without success terminal output', async () => { + const terminal = { type: 'response.failed', sequence_number: 1, error: { message: 'upstream failed' } - }, - { - type: 'response.incomplete', - sequence_number: 1, - response: { error: { message: 'incomplete response' } } - } - ])('fails on $type without success terminal output', async terminal => { + }; const response = responseFromText(sse([created, terminal])); await expect( - readText( - transformPerplexityAgentStream(response, 'perplexity:fast') - ) - ).rejects.toThrow(); + readText(transformPerplexityAgentStream(response)) + ).rejects.toThrow('upstream failed'); }); it('fails on duplicate terminal events', async () => { @@ -219,9 +205,7 @@ describe('Perplexity Agent stream reducer', () => { sse([created, delta('Hello'), completed(), completed(3)]) ); await expect( - readText( - transformPerplexityAgentStream(response, 'perplexity:fast') - ) + readText(transformPerplexityAgentStream(response)) ).rejects.toThrow('after a terminal event'); }); @@ -230,12 +214,19 @@ describe('Perplexity Agent stream reducer', () => { sse([created, delta('Hello', 3), completed(2)]) ); await expect( - readText( - transformPerplexityAgentStream(response, 'perplexity:fast') - ) + readText(transformPerplexityAgentStream(response)) ).rejects.toThrow('duplicate or out of order'); }); + it('requires a sequence number on every event', async () => { + const response = responseFromText( + sse([{ ...created, sequence_number: undefined }]) + ); + await expect( + readText(transformPerplexityAgentStream(response)) + ).rejects.toThrow('sequence is malformed'); + }); + it('fails on out-of-order output indexes', async () => { const laterOutput = { ...delta('later', 1), @@ -250,9 +241,7 @@ describe('Perplexity Agent stream reducer', () => { sse([created, laterOutput, earlierOutput, completed(3)]) ); await expect( - readText( - transformPerplexityAgentStream(response, 'perplexity:fast') - ) + readText(transformPerplexityAgentStream(response)) ).rejects.toThrow('text indexes are out of order'); }); @@ -263,14 +252,30 @@ describe('Perplexity Agent stream reducer', () => { response: { id: 'resp_stream_123', created_at: 1787928001, - status: 'in_progress' + status: 'in_progress', + model: 'openai/gpt-5.4-mini' } }; const response = responseFromText(sse([created, changedIdentity])); await expect( - readText( - transformPerplexityAgentStream(response, 'perplexity:fast') - ) + readText(transformPerplexityAgentStream(response)) + ).rejects.toThrow('response identity changed'); + }); + + it('fails when the executing model changes', async () => { + const changedModel = { + type: 'response.in_progress', + sequence_number: 1, + response: { + id: 'resp_stream_123', + created_at: 1787928000, + status: 'in_progress', + model: 'anthropic/claude-sonnet-4-6' + } + }; + const response = responseFromText(sse([created, changedModel])); + await expect( + readText(transformPerplexityAgentStream(response)) ).rejects.toThrow('response identity changed'); }); @@ -279,27 +284,21 @@ describe('Perplexity Agent stream reducer', () => { sse([created, delta('partial')], '\n', false) ); await expect( - readText( - transformPerplexityAgentStream(prematureEOF, 'perplexity:fast') - ) + readText(transformPerplexityAgentStream(prematureEOF)) ).rejects.toThrow('before response.completed'); const prematureDone = responseFromText( sse([created, delta('partial')]) ); await expect( - readText( - transformPerplexityAgentStream(prematureDone, 'perplexity:fast') - ) + readText(transformPerplexityAgentStream(prematureDone)) ).rejects.toThrow('before response.completed'); }); it('fails when a completed stream contains no text', async () => { const response = responseFromText(sse([created, completed(1)])); await expect( - readText( - transformPerplexityAgentStream(response, 'perplexity:fast') - ) + readText(transformPerplexityAgentStream(response)) ).rejects.toThrow('without output text'); }); }); diff --git a/packages/baseai/src/dev/providers/perplexity/fixtures/documented-agent-response.json b/packages/baseai/src/dev/providers/perplexity/fixtures/documented-agent-response.json new file mode 100644 index 00000000..4c1fa407 --- /dev/null +++ b/packages/baseai/src/dev/providers/perplexity/fixtures/documented-agent-response.json @@ -0,0 +1,36 @@ +{ + "id": "resp_docs_model_card", + "object": "response", + "created_at": 1784292159, + "status": "completed", + "model": "openai/gpt-5.4-mini", + "output": [ + { + "type": "search_results", + "results": [ + { + "id": 4, + "url": "https://arxiv.org/abs/1810.03993" + } + ] + }, + { + "id": "msg_docs_model_card", + "content": [ + { + "text": "A model card documents intended use and limitations.[web:4]", + "type": "output_text", + "annotations": [] + } + ], + "role": "assistant", + "status": "completed", + "type": "message" + } + ], + "usage": { + "input_tokens": 120, + "output_tokens": 12, + "total_tokens": 132 + } +} diff --git a/packages/baseai/src/dev/providers/perplexity/fixtures/documented-agent-stream.json b/packages/baseai/src/dev/providers/perplexity/fixtures/documented-agent-stream.json new file mode 100644 index 00000000..0b453d1d --- /dev/null +++ b/packages/baseai/src/dev/providers/perplexity/fixtures/documented-agent-stream.json @@ -0,0 +1,60 @@ +[ + { + "type": "response.created", + "sequence_number": 0, + "response": { + "id": "resp_docs_stream", + "object": "response", + "created_at": 1784292159, + "status": "in_progress", + "model": "openai/gpt-5.4-mini", + "output": [] + } + }, + { + "type": "response.output_item.added", + "sequence_number": 1, + "output_index": 0, + "item": { + "id": "msg_docs_stream", + "content": [], + "role": "assistant", + "status": "in_progress", + "type": "message" + } + }, + { + "type": "response.output_text.delta", + "sequence_number": 2, + "item_id": "msg_docs_stream", + "output_index": 0, + "content_index": 0, + "delta": "A model card" + }, + { + "type": "response.completed", + "sequence_number": 3, + "response": { + "id": "resp_docs_stream", + "object": "response", + "created_at": 1784292159, + "status": "completed", + "model": "openai/gpt-5.4-mini", + "output": [ + { + "id": "msg_docs_stream", + "content": [ + { + "text": "A model card", + "type": "output_text", + "annotations": [] + } + ], + "role": "assistant", + "status": "completed", + "type": "message" + } + ] + } + } +] diff --git a/packages/baseai/src/dev/routes/v1/pipes/run.perplexity.test.ts b/packages/baseai/src/dev/routes/v1/pipes/run.perplexity.test.ts new file mode 100644 index 00000000..940fedbf --- /dev/null +++ b/packages/baseai/src/dev/routes/v1/pipes/run.perplexity.test.ts @@ -0,0 +1,276 @@ +import { handleError } from '@/dev/hono/errors'; +import documentedAgentResponse from '@/dev/providers/perplexity/fixtures/documented-agent-response.json'; +import documentedAgentStream from '@/dev/providers/perplexity/fixtures/documented-agent-stream.json'; +import { Hono, type Context } from 'hono'; +import { + afterEach, + beforeEach, + describe, + expect, + it, + type Mock, + vi +} from 'vitest'; +import type { Pipe } from 'types/pipe'; +import { registerV1PipesRun } from './run'; + +vi.mock('@/utils/logger-utils', () => ({ logger: vi.fn() })); + +const AGENT_ALIASES = ['fast', 'low', 'medium', 'high'] as const; +const LEGACY_MODELS = [ + 'llama-3.1-sonar-huge-128k-online', + 'llama-3.1-sonar-large-128k-online', + 'llama-3.1-sonar-small-128k-online', + 'llama-3.1-sonar-large-128k-chat', + 'llama-3.1-sonar-small-128k-chat' +] as const; + +function scaffoldPipe(model: string, overrides: Partial = {}): Pipe { + return { + name: 'perplexity-agent', + description: 'Perplexity Agent route integration', + status: 'private', + model: model as Pipe['model'], + stream: true, + json: false, + store: true, + moderate: true, + top_p: 1, + max_tokens: 1000, + temperature: 0.7, + presence_penalty: 1, + frequency_penalty: 1, + stop: [], + tool_choice: 'auto', + parallel_tool_calls: true, + messages: [ + { + role: 'system', + content: 'Answer concisely and cite factual claims.' + } + ], + variables: [], + memory: [], + tools: [], + ...overrides + }; +} + +function requestBody(pipe: Pipe, stream = false) { + return { + pipe, + stream, + messages: [{ role: 'user', content: 'What is a model card?' }], + llmApiKey: 'test-perplexity-key' + }; +} + +function encodeSSE(events: object[]): string { + return `${events + .map(event => `data: ${JSON.stringify(event)}\n\n`) + .join('')}data: [DONE]\n\n`; +} + +function legacyResponse(model: string) { + return { + id: 'legacy-response', + object: 'chat.completion', + created: 1784292159, + model, + choices: [ + { + index: 0, + message: { role: 'assistant', content: 'Legacy response' }, + finish_reason: 'stop' + } + ], + usage: { prompt_tokens: 5, completion_tokens: 2, total_tokens: 7 } + }; +} + +describe('POST /v1/pipes/run Perplexity dispatch', () => { + let app: Hono; + let fetchMock: Mock, ReturnType>; + + beforeEach(() => { + app = new Hono(); + app.onError((error, context) => + handleError(error, context as unknown as Context) + ); + registerV1PipesRun(app); + fetchMock = vi.fn(async (input, init) => { + const url = String(input); + const body = JSON.parse(String(init?.body || '{}')) as { + model?: string; + stream?: boolean; + }; + if (url.endsWith('/v1/agent')) { + if (body.stream) { + return new Response(encodeSSE(documentedAgentStream), { + status: 200, + headers: { 'content-type': 'text/event-stream' } + }); + } + return new Response(JSON.stringify(documentedAgentResponse), { + status: 200, + headers: { 'content-type': 'application/json' } + }); + } + return new Response( + JSON.stringify(legacyResponse(body.model || '')), + { + status: 200, + headers: { 'content-type': 'application/json' } + } + ); + }); + vi.stubGlobal('fetch', fetchMock); + }); + + afterEach(() => { + vi.unstubAllGlobals(); + }); + + async function runRoute(body: object): Promise { + return app.request('http://localhost/v1/pipes/run', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify(body) + }); + } + + it.each(AGENT_ALIASES)( + 'routes perplexity:%s through every runtime layer to /v1/agent', + async preset => { + const response = await runRoute( + requestBody(scaffoldPipe(`perplexity:${preset}`)) + ); + expect(response.status).toBe(200); + const result = await response.json(); + expect(result).toMatchObject({ + completion: + 'A model card documents intended use and limitations.[web:4]', + model: 'openai/gpt-5.4-mini' + }); + + expect(fetchMock).toHaveBeenCalledOnce(); + const [url, init] = fetchMock.mock.calls[0] as [ + string, + RequestInit + ]; + expect(url).toBe('https://api.perplexity.ai/v1/agent'); + expect(init.method).toBe('POST'); + expect(JSON.parse(String(init.body))).toEqual({ + preset, + input: [ + { + type: 'message', + role: 'system', + content: 'Answer concisely and cite factual claims.' + }, + { + type: 'message', + role: 'user', + content: 'What is a model card?' + } + ], + max_output_tokens: 1000, + stream: false, + temperature: 0.7, + top_p: 1 + }); + } + ); + + it.each(LEGACY_MODELS)( + 'keeps perplexity:%s on Chat Completions without rewriting it', + async model => { + const response = await runRoute( + requestBody(scaffoldPipe(`perplexity:${model}`)) + ); + expect(response.status).toBe(200); + const [url, init] = fetchMock.mock.calls[0] as [ + string, + RequestInit + ]; + expect(url).toBe('https://api.perplexity.ai/chat/completions'); + expect(JSON.parse(String(init.body)).model).toBe(model); + } + ); + + it('rejects request tools before provider fetch', async () => { + const response = await runRoute({ + ...requestBody(scaffoldPipe('perplexity:fast')), + tools: [ + { + type: 'function', + function: { name: 'lookup', parameters: { type: 'object' } } + } + ] + }); + expect(response.status).toBe(400); + expect(fetchMock).not.toHaveBeenCalled(); + }); + + it('streams one Agent response through the route', async () => { + const response = await runRoute( + requestBody(scaffoldPipe('perplexity:fast'), true) + ); + expect(response.status).toBe(200); + const text = await response.text(); + expect(text).toContain('A model card'); + expect(text.match(/data: \[DONE\]/g)).toHaveLength(1); + }); + + it('surfaces a provider stream failure after response headers', async () => { + fetchMock.mockResolvedValueOnce( + new Response( + encodeSSE([ + documentedAgentStream[0], + { + type: 'response.failed', + sequence_number: 1, + error: { message: 'stream failed' } + } + ]), + { + status: 200, + headers: { 'content-type': 'text/event-stream' } + } + ) + ); + const response = await runRoute( + requestBody(scaffoldPipe('perplexity:fast'), true) + ); + expect(response.status).toBe(200); + await expect(response.text()).rejects.toThrow('stream failed'); + }); + + it.each([ + [401, 'UNAUTHORIZED'], + [403, 'FORBIDDEN'], + [404, 'NOT_FOUND'], + [429, 'RATE_LIMITED'], + [500, 'INTERNAL_SERVER_ERROR'] + ] as const)( + 'preserves provider HTTP %i as %s at the route', + async (status, code) => { + fetchMock.mockResolvedValueOnce( + new Response( + JSON.stringify({ error: { message: 'Provider error' } }), + { + status + } + ) + ); + const response = await runRoute( + requestBody(scaffoldPipe('perplexity:fast')) + ); + expect(response.status).toBe(status); + await expect(response.json()).resolves.toMatchObject({ + success: false, + error: { status, code } + }); + } + ); +}); diff --git a/packages/baseai/src/dev/utils/stream/stream.ts b/packages/baseai/src/dev/utils/stream/stream.ts index 1086cbcc..dcdfb739 100644 --- a/packages/baseai/src/dev/utils/stream/stream.ts +++ b/packages/baseai/src/dev/utils/stream/stream.ts @@ -47,13 +47,16 @@ export class Stream implements AsyncIterable { try { data = JSON.parse(sse.data); - } catch { - throw new Error( - 'Could not parse SSE message as JSON' + } catch (e) { + console.error( + `Could not parse message into JSON:`, + sse.data ); + console.error(`From chunk:`, sse.raw); + throw e; } - if (data && data.error && !data.type) { + if (data && data.error) { throw new Error(data.error); } @@ -62,10 +65,13 @@ export class Stream implements AsyncIterable { let data; try { data = JSON.parse(sse.data); - } catch { - throw new Error( - 'Could not parse SSE message as JSON' + } catch (e) { + console.error( + `Could not parse message into JSON:`, + sse.data ); + console.error(`From chunk:`, sse.raw); + throw e; } // TODO: Is this where the error should be thrown? if (sse.event == 'error') { @@ -231,9 +237,6 @@ export async function* _iterSSEMessages( const sse = sseDecoder.decode(line); if (sse) yield sse; } - - const finalEvent = sseDecoder.flush(); - if (finalEvent) yield finalEvent; } /** @@ -281,7 +284,7 @@ function findDoubleNewlineIndex(buffer: Uint8Array): number { const newline = 0x0a; // \n const carriage = 0x0d; // \r - for (let i = 0; i < buffer.length - 1; i++) { + for (let i = 0; i < buffer.length - 2; i++) { if (buffer[i] === newline && buffer[i + 1] === newline) { // \n\n return i + 2; @@ -358,20 +361,6 @@ class SSEDecoder { return null; } - - flush(): ServerSentEvent | null { - if (!this.event && !this.data.length) return null; - - const sse: ServerSentEvent = { - event: this.event, - data: this.data.join('\n'), - raw: this.chunks - }; - this.event = null; - this.data = []; - this.chunks = []; - return sse; - } } /** @@ -442,10 +431,25 @@ class LineDecoder { if (bytes == null) return ''; if (typeof bytes === 'string') return bytes; + // Node: + if (typeof Buffer !== 'undefined') { + if (bytes instanceof Buffer) { + return bytes.toString(); + } + if (bytes instanceof Uint8Array) { + return Buffer.from(bytes).toString(); + } + + throw new Error( + `Unexpected: received non-Uint8Array (${bytes.constructor.name}) stream chunk in an environment with a global "Buffer" defined, which this library assumes to be Node. Please report this error.` + ); + } + + // Browser if (typeof TextDecoder !== 'undefined') { if (bytes instanceof Uint8Array || bytes instanceof ArrayBuffer) { this.textDecoder ??= new TextDecoder('utf8'); - return this.textDecoder.decode(bytes, { stream: true }); + return this.textDecoder.decode(bytes); } throw new Error( @@ -461,9 +465,6 @@ class LineDecoder { } flush(): string[] { - const remainingText = this.textDecoder?.decode() || ''; - if (remainingText) this.buffer.push(remainingText); - if (!this.buffer.length && !this.trailingCR) { return []; } From 557efc20ba58cb0dad24b46d3ec4dc10b0212ee9 Mon Sep 17 00:00:00 2001 From: andrewmadson-pplx Date: Fri, 28 Aug 2026 16:44:14 -0700 Subject: [PATCH 7/8] docs(perplexity): complete Agent API migration guide --- .changeset/strong-agents-replace-sonar.md | 4 +- .../docs/supported-models-and-providers.mdx | 75 +++++++++++++++++-- .../examples/pipe.perplexity-agent.stream.ts | 25 +++++++ examples/nodejs/package.json | 13 +++- examples/nodejs/readme.md | 53 +++++++++++++ examples/nodejs/tsconfig.json | 14 ++++ packages/baseai/CHANGELOG.md | 3 +- pnpm-lock.yaml | 14 +++- 8 files changed, 186 insertions(+), 15 deletions(-) create mode 100644 examples/nodejs/examples/pipe.perplexity-agent.stream.ts create mode 100644 examples/nodejs/tsconfig.json diff --git a/.changeset/strong-agents-replace-sonar.md b/.changeset/strong-agents-replace-sonar.md index 8be32e7e..52161ed1 100644 --- a/.changeset/strong-agents-replace-sonar.md +++ b/.changeset/strong-agents-replace-sonar.md @@ -6,4 +6,6 @@ Add Perplexity Agent API support with the `fast`, `low`, `medium`, and `high` presets for local BaseAI Pipes. The existing Perplexity Sonar identifiers remain on Chat Completions but are deprecated ahead of their scheduled September 27, -2026 shutdown; existing Pipes are not migrated automatically. +2026 shutdown; existing Pipes are not migrated automatically. Agent responses +preserve the executing model and available usage, provider HTTP errors retain +their BaseAI status categories, and tool-free Core streams return immediately. diff --git a/apps/baseai.dev/content/docs/docs/supported-models-and-providers.mdx b/apps/baseai.dev/content/docs/docs/supported-models-and-providers.mdx index f2f9084d..242a9ea0 100644 --- a/apps/baseai.dev/content/docs/docs/supported-models-and-providers.mdx +++ b/apps/baseai.dev/content/docs/docs/supported-models-and-providers.mdx @@ -170,13 +170,78 @@ Sonar Chat Completions is now [**Agent API**](https://docs.perplexity.ai/docs/ag The following legacy Perplexity Sonar identifiers remain routed through Chat Completions. Existing Pipes are not migrated automatically. -- `perplexity:llama-3.1-sonar-huge-128k-online` -- `perplexity:llama-3.1-sonar-large-128k-online` -- `perplexity:llama-3.1-sonar-small-128k-online` -- `perplexity:llama-3.1-sonar-large-128k-chat` -- `perplexity:llama-3.1-sonar-small-128k-chat` +
    +
  • perplexity:llama-3.1-sonar-huge-128k-online
    • +
  • perplexity:llama-3.1-sonar-large-128k-online
    • +
  • perplexity:llama-3.1-sonar-small-128k-online
    • +
  • perplexity:llama-3.1-sonar-large-128k-chat
    • +
  • perplexity:llama-3.1-sonar-small-128k-chat
    • +
 +#### Migrate a BaseAI Pipe + +Choose an Agent alias by workload from the table above; the historical BaseAI identifiers do not have an exact one-to-one mapping to today's Sonar product names. Existing Pipes are not changed automatically. + +For example, keep the rest of the Pipe configuration and replace its legacy model ID with the preset that fits the workload. + +Before: + +```ts +const researchPipe = (): PipeI => ({ + model: 'perplexity:llama-3.1-sonar-small-128k-online', + max_tokens: 1000, + temperature: 0.7, + top_p: 1, + // ...the remaining Pipe fields +}); +``` + +After: + +```ts +const researchPipe = (): PipeI => ({ + model: 'perplexity:fast', + max_tokens: 1000, + temperature: 0.7, + top_p: 1, + // ...the remaining Pipe fields +}); +``` + +BaseAI requires `max_tokens`, `temperature`, and `top_p`. Agent API receives them as request values (`max_tokens` becomes `max_output_tokens`), so they override the selected preset's tuned defaults. Review these values while migrating. + +Set both keys in the local caller's `.env` file: + +```bash +LANGBASE_API_KEY="your-langbase-key" +PERPLEXITY_API_KEY="your-perplexity-key" +``` + +`@baseai/core` reads `PERPLEXITY_API_KEY` from the local process and sends it as `llmApiKey` to the BaseAI server on `localhost:9000`; the local server uses it only for the Perplexity provider request. `LANGBASE_API_KEY` is read by the Pipe configuration. Keep both values server-side and out of source control. + +From the repository root, build and run the workspace-linked Node example: + +```bash +pnpm install +pnpm --filter example-nodejs check +pnpm --filter example-nodejs baseai dev +``` + +Then run either request in a second terminal: + +```bash +# Non-streaming answer plus citation metadata +pnpm --filter example-nodejs pipe.perplexity.agent + +# Streaming answer text through Pipe.run({ stream: true }) +pnpm --filter example-nodejs pipe.perplexity.agent.stream +``` + +This PR changes only the public BaseAI local runtime; Langbase-hosted execution and rollout are outside its scope. Custom tools and tool-result replay, streamed citation metadata, background mode, file output, explicit Agent models, `xhigh`, and `wide-research` are not supported in this integration. + +Read the [migration overview](https://docs.perplexity.ai/docs/agent-api/migrate-from-sonar/overview) for workload guidance and the [field-by-field migration guide](https://docs.perplexity.ai/docs/agent-api/migrate-from-sonar/how-to) for request, response, citation, and streaming changes. + | Legacy model | Provider | Owner | Context | Cost* | |----------------------------------------------------------------------------------------------------------|------------|-------|---------|--------------------------------------------| | llama-3.1-sonar-huge-128k-online
ID: | Perplexity | Meta | 127,072 | $5 prompt
$5 completion | diff --git a/examples/nodejs/examples/pipe.perplexity-agent.stream.ts b/examples/nodejs/examples/pipe.perplexity-agent.stream.ts new file mode 100644 index 00000000..ab4fe0a0 --- /dev/null +++ b/examples/nodejs/examples/pipe.perplexity-agent.stream.ts @@ -0,0 +1,25 @@ +import 'dotenv/config'; +import {getRunner, Pipe} from '@baseai/core'; +import perplexityAgentPipe from '../baseai/pipes/perplexity-agent'; + +const pipe = new Pipe(perplexityAgentPipe()); + +async function main() { + const {stream} = await pipe.run({ + messages: [ + { + role: 'user', + content: + 'What changed in the latest JavaScript language release?', + }, + ], + stream: true, + }); + + const runner = getRunner(stream); + runner.on('content', content => process.stdout.write(content)); + runner.on('end', () => process.stdout.write('\n')); + runner.on('error', error => console.error(error)); +} + +main(); diff --git a/examples/nodejs/package.json b/examples/nodejs/package.json index d2ac131a..6d70436e 100644 --- a/examples/nodejs/package.json +++ b/examples/nodejs/package.json @@ -7,22 +7,27 @@ "main": "index.js", "scripts": { "baseai": "baseai", + "check": "pnpm --filter baseai build && pnpm type-check && baseai --help", "pipe.run": "npx tsx ./examples/pipe.run.ts", "pipe.run.stream": "npx tsx ./examples/pipe.run.stream.ts", "pipe.run.stream.loop": "npx tsx ./examples/pipe.run.stream.loop.ts", "pipe.perplexity.agent": "npx tsx ./examples/pipe.perplexity-agent.ts", + "pipe.perplexity.agent.stream": "npx tsx ./examples/pipe.perplexity-agent.stream.ts", "pipe.generate.text": "npx tsx ./examples/pipe.generate.text.ts", - "pipe.stream.text": "npx tsx ./examples/pipe.stream.text.ts" + "pipe.stream.text": "npx tsx ./examples/pipe.stream.text.ts", + "type-check": "pnpm --filter @baseai/core build && tsc --noEmit" }, "keywords": [], "author": "Ahmad Awais (https://twitter.com/MrAhmadAwais)", "license": "UNLICENSED", "dependencies": { - "@baseai/core": "^0.9.43", + "@baseai/core": "workspace:*", "dotenv": "^16.4.5" }, "devDependencies": { - "baseai": "^0.9.44", - "tsx": "^4.19.0" + "@types/node": "^22.6.1", + "baseai": "workspace:*", + "tsx": "^4.19.0", + "typescript": "^5.6.2" } } diff --git a/examples/nodejs/readme.md b/examples/nodejs/readme.md index 54f74a16..dd0e89ef 100644 --- a/examples/nodejs/readme.md +++ b/examples/nodejs/readme.md @@ -4,6 +4,58 @@ BaseAI Node.js examples. Please read the [documentation](https://baseai.dev/docs) for more information. +## Perplexity Agent API migration example + +Sonar Chat Completions is now [**Agent API**](https://docs.perplexity.ai/docs/agent-api/quickstart). Migrate by September 27, 2026. View the [**Migration Guide**](https://docs.perplexity.ai/docs/agent-api/migrate-from-sonar/overview). + +The four opt-in local BaseAI Agent aliases are `perplexity:fast` for quick lookups, `perplexity:low` for everyday research, `perplexity:medium` for multi-hop research, and `perplexity:high` for deep research. Choose by workload; do not assume an exact mapping from these historical BaseAI IDs: + +- `perplexity:llama-3.1-sonar-huge-128k-online` +- `perplexity:llama-3.1-sonar-large-128k-online` +- `perplexity:llama-3.1-sonar-small-128k-online` +- `perplexity:llama-3.1-sonar-large-128k-chat` +- `perplexity:llama-3.1-sonar-small-128k-chat` + +Existing Pipes are not migrated automatically. Change only the model ID and review the required sampling fields: + +```diff +-model: 'perplexity:llama-3.1-sonar-small-128k-online', ++model: 'perplexity:fast', + max_tokens: 1000, + temperature: 0.7, + top_p: 1, +``` + +BaseAI sends `max_tokens` as Agent `max_output_tokens` and sends `temperature` and `top_p`; supplied values override preset defaults. + +Create `examples/nodejs/.env` with both server-side keys: + +```bash +LANGBASE_API_KEY="your-langbase-key" +PERPLEXITY_API_KEY="your-perplexity-key" +``` + +The local Core caller reads `PERPLEXITY_API_KEY` and passes it to the BaseAI server on `localhost:9000` for the provider request. `LANGBASE_API_KEY` is read by the Pipe configuration. Do not commit either value. + +From the repository root: + +```bash +pnpm install +pnpm --filter example-nodejs check +pnpm --filter example-nodejs baseai dev +``` + +In another terminal, run the non-streaming example to print completion text and typed `citationMetadata`, or the separate streaming example to consume the public Core stream: + +```bash +pnpm --filter example-nodejs pipe.perplexity.agent +pnpm --filter example-nodejs pipe.perplexity.agent.stream +``` + +This is local-runtime support only; it does not demonstrate Langbase-hosted execution. Custom tools and tool replay, streamed citation metadata, background mode, file output, explicit Agent models, `xhigh`, and `wide-research` are unsupported. See the [migration overview](https://docs.perplexity.ai/docs/agent-api/migrate-from-sonar/overview) and [field-by-field guide](https://docs.perplexity.ai/docs/agent-api/migrate-from-sonar/how-to). + +## Other examples + ```sh # Install the dependencies npm install @@ -23,6 +75,7 @@ npm run pipe.run npm run pipe.run.stream npm run pipe.run.stream.loop npm run pipe.perplexity.agent +npm run pipe.perplexity.agent.stream npm run pipe.generate.text npm run pipe.stream.text ``` diff --git a/examples/nodejs/tsconfig.json b/examples/nodejs/tsconfig.json new file mode 100644 index 00000000..d7ecca4c --- /dev/null +++ b/examples/nodejs/tsconfig.json @@ -0,0 +1,14 @@ +{ + "compilerOptions": { + "esModuleInterop": true, + "lib": ["ES2022", "DOM", "DOM.Iterable"], + "module": "ESNext", + "moduleResolution": "Bundler", + "noEmit": true, + "skipLibCheck": true, + "strict": true, + "target": "ES2022", + "types": ["node"] + }, + "include": ["baseai/**/*.ts", "examples/**/*.ts"] +} diff --git a/packages/baseai/CHANGELOG.md b/packages/baseai/CHANGELOG.md index 9ba8eefd..ab25a402 100644 --- a/packages/baseai/CHANGELOG.md +++ b/packages/baseai/CHANGELOG.md @@ -5,7 +5,8 @@ ### Minor Changes - Add local Perplexity Agent API support with the `fast`, `low`, `medium`, and - `high` presets, including text streaming and citation mapping. + `high` presets, including provider-local text streaming, citation mapping, + executing-model reporting, optional usage, and status-preserving errors. - Deprecate the five legacy `perplexity:llama-3.1-sonar-*` identifiers ahead of their scheduled September 27, 2026 shutdown. Existing Pipes are not migrated automatically; follow the [Perplexity Sonar migration diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index c323a229..ba4f904d 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -389,18 +389,24 @@ importers: examples/nodejs: dependencies: '@baseai/core': - specifier: ^0.9.43 - version: 0.9.43(react@19.2.4)(zod@3.23.8) + specifier: workspace:* + version: link:../../packages/core dotenv: specifier: ^16.4.5 version: 16.4.5 devDependencies: + '@types/node': + specifier: ^22.6.1 + version: 22.7.4 baseai: - specifier: ^0.9.44 - version: 0.9.44(@types/node@22.7.4)(typescript@5.6.2) + specifier: workspace:* + version: link:../../packages/baseai tsx: specifier: ^4.19.0 version: 4.19.1 + typescript: + specifier: ^5.6.2 + version: 5.6.2 examples/remix: dependencies: From 2fa4369ee20aa39feeb129e4fc9607f454ae2f2d Mon Sep 17 00:00:00 2001 From: andrewmadson-pplx Date: Fri, 28 Aug 2026 16:46:40 -0700 Subject: [PATCH 8/8] docs(perplexity): clarify OSS-only scope --- .../content/docs/docs/supported-models-and-providers.mdx | 2 +- examples/nodejs/readme.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/baseai.dev/content/docs/docs/supported-models-and-providers.mdx b/apps/baseai.dev/content/docs/docs/supported-models-and-providers.mdx index 242a9ea0..f9363805 100644 --- a/apps/baseai.dev/content/docs/docs/supported-models-and-providers.mdx +++ b/apps/baseai.dev/content/docs/docs/supported-models-and-providers.mdx @@ -162,7 +162,7 @@ Use a Perplexity Agent API preset for new local BaseAI integrations. Presets sel This first Agent integration supports system, user, and assistant text; non-streaming responses; streaming answer text; and preset search. Custom tools, tool-result replay, background runs, file outputs, and explicit Agent model selection are not supported. Citation metadata is included on non-streaming responses; streamed citation metadata is not exposed because the existing BaseAI stream chunk contract has no compatible citation field. -These Agent identifiers are supported by the public BaseAI local runtime. Langbase-hosted model availability and rollout are managed separately. +These Agent identifiers are supported only by the public BaseAI OSS local runtime. This integration does not change, validate, or propose fixes for Langbase-hosted or other proprietary systems. diff --git a/examples/nodejs/readme.md b/examples/nodejs/readme.md index dd0e89ef..82fa162a 100644 --- a/examples/nodejs/readme.md +++ b/examples/nodejs/readme.md @@ -52,7 +52,7 @@ pnpm --filter example-nodejs pipe.perplexity.agent pnpm --filter example-nodejs pipe.perplexity.agent.stream ``` -This is local-runtime support only; it does not demonstrate Langbase-hosted execution. Custom tools and tool replay, streamed citation metadata, background mode, file output, explicit Agent models, `xhigh`, and `wide-research` are unsupported. See the [migration overview](https://docs.perplexity.ai/docs/agent-api/migrate-from-sonar/overview) and [field-by-field guide](https://docs.perplexity.ai/docs/agent-api/migrate-from-sonar/how-to). +This is public BaseAI OSS local-runtime support only. It does not change, validate, or propose fixes for Langbase-hosted or other proprietary systems. Custom tools and tool replay, streamed citation metadata, background mode, file output, explicit Agent models, `xhigh`, and `wide-research` are unsupported. See the [migration overview](https://docs.perplexity.ai/docs/agent-api/migrate-from-sonar/overview) and [field-by-field guide](https://docs.perplexity.ai/docs/agent-api/migrate-from-sonar/how-to). ## Other examples