Skip to content

Instantly share code, notes, and snippets.

@CertainLach

CertainLach/rpc.ts

Created September 18, 2025 15:22
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save CertainLach/2fd6affebb87f0ebbe3705cd3ea411f3 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save CertainLach/2fd6affebb87f0ebbe3705cd3ea411f3 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
rpc.ts
import { audio, image } from '@interpreter/media-conversions';
import { parseableZod, z } from '@interpreter/util';
import { ToolMap, ToolSetSet } from '@interpreter/llm-model/tool';
import { toolCall, toolResult } from '@interpreter/llm-model/toolModel';
import { cacheDelimiter, injectDelimiter } from './cache.ts';
import type { LlmModel } from '@interpreter/llm-model/vendor';
import { Tracer } from '@interpreter/tracing';
import dedent from 'dedent';
import { JsonEnforcerKind } from '@interpreter/free-jsonschema';
import { fail } from '@std/assert';
export const functionCallTracer = new Tracer('function-calling');
export const llmRequestTracer = new Tracer('llm-request');
export {
CacheDelimiter,
cacheDelimiter,
InjectDelimiter,
injectDelimiter,
} from './cache.ts';
/**
* Plaintext passed to LLM
*
* Note that it may include markdown code blocks, and other formatting,
* that is treated by model as plausible output tokens.
*
* It does not, however, includes reasoning tokens, those are `ThinkingPart`
*/
export class TextPart {
// Dummy field to make type branded
#private: unknown;
/**
* @param ws
* section_start if this text part starts a new section, and needs newlines before
* section_end if this text part ends a section, and needs newlines after
* section_start_end if this text part should be isolated, and needs newlines before and after.
*/
constructor(
public text: string,
public ws?:
| 'section_start'
| 'section_end'
| 'section_start_end',
) {
this.#private;
}
toJSON() {
if (this.ws === undefined) {
return this.text;
}
return {
text: this.text,
ws: this.ws,
};
}
}
export const textPart = parseableZod(
TextPart,
z.union([
z.string(),
z.object({
text: z.string(),
ws: z.enum([
'section_start',
'section_end',
'section_start_end',
]).optional(),
}).strict(),
]),
(v) => {
if (typeof v === 'string') return new TextPart(v);
return new TextPart(v.text, v.ws);
},
);
/**
* Same as `TextPart`, but is intended to be used for partially untrusted
* input, various anti-promt-injection techniques might be applied to such input
*/
export class InjectedPart {
#private: unknown;
/**
* @param injected Type of injected content visible to the llm, e.g "provided_input"
*/
constructor(public injected: string, public text: string) {
this.#private;
}
/**
* Trivial injected part processing - wrap it in tags, works great for llms with good self-moderation.
*/
uncheckedToText() {
return new TextPart(
`<${this.injected}>\n${this.text}\n</${this.injected}>`,
'section_start_end',
);
}
}
export const injectedPart = parseableZod(
InjectedPart,
z.object({
injected: z.string(),
text: z.string(),
}).strict(),
(v) => new InjectedPart(v.injected, v.text),
);
const redactedKind = z.union([
z.literal('anthropicPartial').describe(
"partial response, for which we don't know the signature yet",
),
z.literal('anthropicRedacted').describe(
"thinking is redacted by anthropic, can be passed back to the api, but can't be read",
),
z.literal('anthropicNormal').describe(
'thinking was not redacted, signature is valid for anthropic',
),
]);
export class ThinkingPart {
#private: unknown;
constructor(
public thinking: string,
public redacted: z.infer<typeof redactedKind>,
public proprietarySignature: string | undefined,
) {
this.#private;
}
get text(): string | undefined {
if (
this.redacted === 'anthropicNormal' ||
this.redacted === 'anthropicPartial'
) {
return this.text;
}
return undefined;
}
}
export const thinkingPart = parseableZod(
ThinkingPart,
z.object({
thinking: z.string(),
redacted: redactedKind,
proprietarySignature: z.string().optional(),
}),
(v) => {
return new ThinkingPart(v.thinking, v.redacted, v.proprietarySignature);
},
);
// TODO: Content quality: text may be summarized, images can be зашакалены, voice can have reduced bitrate, etc.
// Right now this processing should be handled on the user side, which is suboptimal for our openai image url caching.
const plainMessagePart = z.union([
textPart,
image,
audio,
]);
export type PlainMessagePart = z.infer<typeof plainMessagePart>;
const assistantMessagePart = z.union([
plainMessagePart,
thinkingPart,
toolCall,
]);
export type AssistantMessagePart = z.infer<typeof assistantMessagePart>;
const assistantMessageContent = z.union([
z.string(),
assistantMessagePart.array(),
]).transform((v) => {
if (typeof v === 'string') return [new TextPart(v)];
return v;
});
const userMessagePart = z.union([
plainMessagePart,
injectedPart,
toolResult,
cacheDelimiter,
]);
export type UserMessagePart = z.infer<typeof userMessagePart>;
const userMessageContent = z.union([
z.string(),
userMessagePart.array(),
]).transform((v) => {
if (typeof v === 'string') return [new TextPart(v)];
return v;
});
const systemMessagePart = z.union([
plainMessagePart,
cacheDelimiter,
injectDelimiter,
]);
export type SystemMessagePart = z.infer<typeof systemMessagePart>;
const systemMessageContent = z.union([
z.string(),
systemMessagePart.array(),
]).transform((v) => {
if (typeof v === 'string') return [new TextPart(v)];
return v;
});
const messagePart = z.union([
plainMessagePart,
toolCall,
toolResult,
cacheDelimiter,
thinkingPart,
injectedPart,
]);
export type MessagePart = z.infer<typeof messagePart>;
export const assistantMessage = z.object({
role: z.literal('assistant'),
name: z.string().optional(),
content: assistantMessageContent,
}).strict();
export type AssistantMessage = z.infer<typeof assistantMessage>;
const userMessage = z.object({
role: z.literal('user'),
name: z.string().optional(),
content: userMessageContent,
}).strict();
export type UserMessage = z.infer<typeof userMessage>;
function roleM<R extends string, P>(
role: R,
_parts: TemplateStringsArray,
extra: P[],
) {
const parts = _parts.map((p, i) => {
const firstPart = i === 0;
const lastPart = i === _parts.length - 1;
let dedented = dedent.withOptions({ trimWhitespace: false })(p);
const startsSection = dedented.startsWith('\n\n');
const endsSection = dedented.endsWith('\n\n');
if (startsSection || firstPart) {
dedented = dedented.trimStart();
}
if (endsSection || lastPart) {
dedented = dedented.trimEnd();
}
return new TextPart(
dedented,
startsSection && endsSection
? 'section_start_end'
: startsSection
? 'section_start'
: endsSection
? 'section_end'
: undefined,
);
});
const content: P[] = [parts[0] as P];
extra.forEach((extra, i) => {
content.push(extra, content[i + 1]);
});
return {
role,
content,
};
}
export function userM(
parts: TemplateStringsArray,
...extra: UserMessagePart[]
): UserMessage & { withName(name: string): UserMessage } {
const msg = roleM(
'user',
parts,
extra,
);
return {
...msg,
withName: (name: string) => ({
...msg,
name,
}),
};
}
export function assistantM(
parts: TemplateStringsArray,
...extra: AssistantMessagePart[]
): AssistantMessage {
return roleM('assistant', parts, extra);
}
export function systemM(
parts: TemplateStringsArray,
...extra: SystemMessagePart[]
): SystemMessagePart[] {
return roleM('system', parts, extra).content;
}
const normalMessage = z.discriminatedUnion('role', [
assistantMessage,
userMessage,
]);
const message = z.union([
normalMessage,
cacheDelimiter,
injectDelimiter,
]);
export type Message = z.infer<typeof message>;
const request = z.object({
system: systemMessageContent,
messages: message.array(),
predictedOutput: systemMessageContent.optional(),
jsonOutput: z.object({
schema: z.unknown(),
guidance: z.string(),
}).strict().optional(),
}).strict();
export type Request = z.infer<typeof request>;
// TODO: Dedicated zod type for that, to expose in api
export type UserRequest = Omit<Request, 'jsonOutput'>;
export type RequestOptions = {
tss?: ToolSetSet;
signal?: AbortSignal;
tools?: ToolMap;
};
export class LlmError extends Error {
constructor(
message: string,
public code: string,
public options?: ErrorOptions,
) {
super(`${message} ${code}`, options);
}
toJSON() {
return this.code;
}
}
export class LlmServerError extends LlmError {}
export class LlmRateLimitError extends LlmServerError {
constructor(message: string) {
super(message, 'llm_rate_limit');
}
}
export class LlmBatchCancelledError extends LlmServerError {
constructor() {
super('batch cancelled', 'llm_batch_cancelled');
}
}
export class LlmBatchExpiredError extends LlmServerError {
constructor() {
super('batch expired', 'llm_batch_expired');
}
}
export class LlmMaxTokensError extends LlmServerError {
constructor(message: string) {
super(message, 'llm_max_tokens');
}
}
export class LlmQualityError extends LlmError {
}
export const llmError = z.union([
z.literal('llm_rate_limit'),
z.literal('llm_batch_cancelled'),
z.literal('llm_batch_expired'),
z.literal('llm_max_tokens'),
z.string(),
]).transform((v) => {
switch (v) {
case 'llm_rate_limit':
return new LlmRateLimitError('cached');
case 'llm_batch_cancelled':
return new LlmBatchCancelledError();
case 'llm_batch_expired_error':
return new LlmBatchExpiredError();
case 'llm_max_tokens':
return new LlmMaxTokensError('cached');
default:
return new LlmError('cached', v);
}
});
export type LlmRpcResponseFormat =
| 'json_object'
| 'json_schema'
| 'proprietary_guided_json'
// | 'proprietary_forced_tool' - anthropic is handled in backend-specific way
// | 'proprietary_commander_script_emulation' - commander is handled in backend-specific way
| 'text';
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment