feat(consultation): let the jyotish agent drive skills and tools
This commit is contained in:
Jesse_Chen
@@ -1,8 +1,16 @@
|
||||
import type { AgentExecutionReceipt, PublicActivityPhase } from "./consultation-agent-events.ts";
|
||||
|
||||
export type AgentActivityView = Readonly<{
|
||||
phase: PublicActivityPhase;
|
||||
label: string;
|
||||
}>;
|
||||
|
||||
export type ChatMessage = {
|
||||
readonly role: "user" | "assistant";
|
||||
readonly text: string;
|
||||
readonly suggestions?: readonly string[];
|
||||
readonly techniqueTruth?: string;
|
||||
readonly agentExecutionReceipt?: AgentExecutionReceipt;
|
||||
readonly workflowReceipt?: {
|
||||
readonly route: string;
|
||||
readonly status: string;
|
||||
@@ -14,12 +22,14 @@ export type ChatMessage = {
|
||||
export type ChatMessageView = ChatMessage & {
|
||||
readonly renderKey: string;
|
||||
readonly state: "settled" | "streaming" | "thinking";
|
||||
readonly activity?: AgentActivityView;
|
||||
};
|
||||
|
||||
export function chatMessageViews(
|
||||
messages: readonly ChatMessage[],
|
||||
loading: boolean,
|
||||
streamingText: string,
|
||||
activity?: AgentActivityView,
|
||||
): readonly ChatMessageView[] {
|
||||
const settled = messages.map((message, index) => ({
|
||||
...message,
|
||||
@@ -35,6 +45,7 @@ export function chatMessageViews(
|
||||
text: streamingText,
|
||||
renderKey: `message-${messages.length}`,
|
||||
state: streamingText ? "streaming" : "thinking",
|
||||
activity,
|
||||
},
|
||||
];
|
||||
}
|
||||
|
||||
@@ -1,10 +1,12 @@
|
||||
import { z } from "zod";
|
||||
import { agentExecutionReceiptSchema, type AgentExecutionReceipt } from "./consultation-agent-events.ts";
|
||||
|
||||
const chatMessageSchema = z.object({
|
||||
role: z.enum(["user", "assistant"]),
|
||||
text: z.string().max(100_000),
|
||||
suggestions: z.array(z.string().max(200)).max(3).optional(),
|
||||
techniqueTruth: z.string().max(120).optional(),
|
||||
agentExecutionReceipt: agentExecutionReceiptSchema.optional(),
|
||||
workflowReceipt: z.object({
|
||||
route: z.string().max(120),
|
||||
status: z.string().max(120),
|
||||
@@ -40,6 +42,7 @@ export type ChatSessionWrite = Readonly<{
|
||||
text: string;
|
||||
suggestions?: readonly string[];
|
||||
techniqueTruth?: string;
|
||||
agentExecutionReceipt?: AgentExecutionReceipt;
|
||||
workflowReceipt?: Readonly<{
|
||||
route: string;
|
||||
status: string;
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
import { z } from "zod";
|
||||
|
||||
export const publicActivityPhaseSchema = z.enum([
|
||||
"loading-method",
|
||||
"chart-calculation",
|
||||
"evidence-validation",
|
||||
"answer-composition",
|
||||
]);
|
||||
export type PublicActivityPhase = z.infer<typeof publicActivityPhaseSchema>;
|
||||
|
||||
export const workflowReceiptSchema = z.object({
|
||||
route: z.string().max(120),
|
||||
status: z.string().max(120),
|
||||
preciseTiming: z.string().max(120),
|
||||
missingLayers: z.array(z.string().max(120)).max(30),
|
||||
}).strict();
|
||||
export type WorkflowReceipt = z.infer<typeof workflowReceiptSchema>;
|
||||
|
||||
const executionStepSchema = z.object({
|
||||
sequence: z.number().int().min(1).max(32),
|
||||
kind: z.enum(["skill", "tool", "validation"]),
|
||||
name: z.string().max(120),
|
||||
status: z.enum(["completed", "failed"]),
|
||||
durationMs: z.number().int().min(0).optional(),
|
||||
}).strict();
|
||||
|
||||
export const agentExecutionReceiptSchema = z.object({
|
||||
runId: z.string().min(1).max(120),
|
||||
runtime: z.literal("mastra-agentic"),
|
||||
skill: z.object({
|
||||
name: z.literal("jyotish-vedic-astrology"),
|
||||
loaded: z.boolean(),
|
||||
version: z.string().max(120).optional(),
|
||||
}).strict(),
|
||||
steps: z.array(executionStepSchema).max(32),
|
||||
workflow: workflowReceiptSchema,
|
||||
techniqueTruth: z.string().max(120).optional(),
|
||||
}).strict();
|
||||
export type AgentExecutionReceipt = z.infer<typeof agentExecutionReceiptSchema>;
|
||||
|
||||
const runStartedSchema = z.object({ type: z.literal("run.started"), runId: z.string(), requestId: z.string() }).strict();
|
||||
const skillStartedSchema = z.object({ type: z.literal("skill.started"), name: z.literal("jyotish-vedic-astrology") }).strict();
|
||||
const skillCompletedSchema = z.object({ type: z.literal("skill.completed"), name: z.literal("jyotish-vedic-astrology") }).strict();
|
||||
const toolStartedSchema = z.object({ type: z.literal("tool.started"), callId: z.string(), tool: z.literal("run-jyotish-consultation"), label: z.string() }).strict();
|
||||
const activitySchema = z.object({ type: z.literal("activity"), phase: publicActivityPhaseSchema, label: z.string().max(120) }).strict();
|
||||
const toolCompletedSchema = z.object({
|
||||
type: z.literal("tool.completed"), callId: z.string(), tool: z.literal("run-jyotish-consultation"),
|
||||
status: z.enum(["ready", "degraded", "blocked"]), durationMs: z.number().int().min(0),
|
||||
}).strict();
|
||||
const toolFailedSchema = z.object({
|
||||
type: z.literal("tool.failed"), callId: z.string(), tool: z.literal("run-jyotish-consultation"),
|
||||
code: z.enum(["calculation_failed", "timeout", "cancelled"]),
|
||||
}).strict();
|
||||
const answerDeltaSchema = z.object({ type: z.literal("answer.delta"), text: z.string() }).strict();
|
||||
const runCompletedSchema = z.object({ type: z.literal("run.completed"), receipt: agentExecutionReceiptSchema }).strict();
|
||||
const runFailedSchema = z.object({
|
||||
type: z.literal("run.failed"),
|
||||
code: z.enum(["runtime_contract_incomplete", "calculation_failed", "empty_answer", "cancelled"]),
|
||||
message: z.string().max(200),
|
||||
}).strict();
|
||||
|
||||
export const consultationAgentPublicEventSchema = z.discriminatedUnion("type", [
|
||||
runStartedSchema, skillStartedSchema, skillCompletedSchema, toolStartedSchema, activitySchema,
|
||||
toolCompletedSchema, toolFailedSchema, answerDeltaSchema, runCompletedSchema, runFailedSchema,
|
||||
]);
|
||||
export type ConsultationAgentPublicEvent = z.infer<typeof consultationAgentPublicEventSchema>;
|
||||
|
||||
export function createNdjsonParser(onEvent: (event: ConsultationAgentPublicEvent) => void) {
|
||||
let buffer = "";
|
||||
function consume(value: string, final: boolean) {
|
||||
buffer += value;
|
||||
const lines = buffer.split("\n");
|
||||
buffer = lines.pop() ?? "";
|
||||
for (const line of lines) {
|
||||
if (line.trim()) onEvent(consultationAgentPublicEventSchema.parse(JSON.parse(line)));
|
||||
}
|
||||
if (final && buffer.trim()) {
|
||||
onEvent(consultationAgentPublicEventSchema.parse(JSON.parse(buffer)));
|
||||
buffer = "";
|
||||
}
|
||||
}
|
||||
return Object.freeze({
|
||||
push: (value: string) => consume(value, false),
|
||||
finish: (value = "") => consume(value, true),
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,253 @@
|
||||
import type { ConsultationRuntimeState } from "../mastra/consultation-tools.ts";
|
||||
import {
|
||||
agentExecutionReceiptSchema,
|
||||
consultationAgentPublicEventSchema,
|
||||
publicActivityPhaseSchema,
|
||||
type AgentExecutionReceipt,
|
||||
type ConsultationAgentPublicEvent,
|
||||
} from "./consultation-agent-events.ts";
|
||||
import { createVisibleTextTransformer } from "./stream-text-response.ts";
|
||||
|
||||
type Chunk = { type?: string; payload?: Record<string, unknown>; data?: unknown };
|
||||
type ChunkStream = AsyncIterable<unknown> | ReadableStream<unknown>;
|
||||
|
||||
async function* readChunks(stream: ChunkStream): AsyncIterable<Chunk> {
|
||||
const values = Symbol.asyncIterator in stream
|
||||
? stream as AsyncIterable<unknown>
|
||||
: (async function* () {
|
||||
const reader = (stream as ReadableStream<unknown>).getReader();
|
||||
try {
|
||||
while (true) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) return;
|
||||
yield value;
|
||||
}
|
||||
} finally {
|
||||
reader.releaseLock();
|
||||
}
|
||||
})();
|
||||
for await (const value of values) {
|
||||
if (value && typeof value === "object") yield value as Chunk;
|
||||
}
|
||||
}
|
||||
type Status = "ready" | "degraded" | "blocked";
|
||||
|
||||
type EventOptions = {
|
||||
runId: string;
|
||||
requestId: string;
|
||||
toolStatus: () => Status;
|
||||
receipt: () => AgentExecutionReceipt;
|
||||
};
|
||||
|
||||
function activity(value: unknown): ConsultationAgentPublicEvent | null {
|
||||
if (!value || typeof value !== "object") return null;
|
||||
const data = value as { phase?: unknown; label?: unknown };
|
||||
const phase = publicActivityPhaseSchema.safeParse(data.phase);
|
||||
if (!phase.success || typeof data.label !== "string") return null;
|
||||
return { type: "activity", phase: phase.data, label: data.label.slice(0, 120) };
|
||||
}
|
||||
|
||||
function safeToolError(error: unknown) {
|
||||
if (error instanceof DOMException && error.name === "AbortError") return "cancelled" as const;
|
||||
if (error instanceof DOMException && error.name === "TimeoutError") return "timeout" as const;
|
||||
return "calculation_failed" as const;
|
||||
}
|
||||
|
||||
function mapChunk(
|
||||
chunk: Chunk,
|
||||
options: EventOptions,
|
||||
startedAt: Map<string, number>,
|
||||
jyotishSkillCallIds: Set<string>,
|
||||
): ConsultationAgentPublicEvent[] {
|
||||
const payload = chunk.payload ?? {};
|
||||
if (chunk.type === "data-jyotish-activity") {
|
||||
const event = activity(chunk.data);
|
||||
return event ? [event] : [];
|
||||
}
|
||||
if (chunk.type === "tool-call") {
|
||||
const toolName = payload.toolName;
|
||||
const callId = typeof payload.toolCallId === "string" ? payload.toolCallId : "tool";
|
||||
if (toolName === "skill" && (payload.args as { name?: unknown } | undefined)?.name === "jyotish-vedic-astrology") {
|
||||
jyotishSkillCallIds.add(callId);
|
||||
return [{ type: "skill.started", name: "jyotish-vedic-astrology" }];
|
||||
}
|
||||
if (toolName === "run-jyotish-consultation") {
|
||||
startedAt.set(callId, Date.now());
|
||||
return [{ type: "tool.started", callId, tool: "run-jyotish-consultation", label: "正在计算个人星盘" }];
|
||||
}
|
||||
}
|
||||
if (chunk.type === "tool-result") {
|
||||
const toolName = payload.toolName;
|
||||
const callId = typeof payload.toolCallId === "string" ? payload.toolCallId : "tool";
|
||||
if (toolName === "skill" && jyotishSkillCallIds.delete(callId)) {
|
||||
return [{ type: "skill.completed", name: "jyotish-vedic-astrology" }];
|
||||
}
|
||||
if (toolName === "run-jyotish-consultation") {
|
||||
return [{
|
||||
type: "tool.completed", callId, tool: "run-jyotish-consultation", status: options.toolStatus(),
|
||||
durationMs: Math.max(0, Date.now() - (startedAt.get(callId) ?? Date.now())),
|
||||
}];
|
||||
}
|
||||
}
|
||||
if (chunk.type === "tool-error") {
|
||||
const toolName = payload.toolName;
|
||||
if (toolName === "run-jyotish-consultation") {
|
||||
return [{
|
||||
type: "tool.failed",
|
||||
callId: typeof payload.toolCallId === "string" ? payload.toolCallId : "tool",
|
||||
tool: "run-jyotish-consultation",
|
||||
code: safeToolError(payload.error),
|
||||
}];
|
||||
}
|
||||
}
|
||||
return [];
|
||||
}
|
||||
|
||||
export async function collectAgentPublicEvents(stream: ChunkStream | Iterable<Chunk>, options: EventOptions) {
|
||||
const events: ConsultationAgentPublicEvent[] = [{ type: "run.started", runId: options.runId, requestId: options.requestId }];
|
||||
const startedAt = new Map<string, number>();
|
||||
const jyotishSkillCallIds = new Set<string>();
|
||||
for await (const chunk of stream instanceof ReadableStream || Symbol.asyncIterator in stream ? readChunks(stream as ChunkStream) : stream) {
|
||||
events.push(...mapChunk(chunk, options, startedAt, jyotishSkillCallIds));
|
||||
if (chunk.type === "text-delta" && typeof chunk.payload?.text === "string") {
|
||||
events.push({ type: "answer.delta", text: chunk.payload.text });
|
||||
}
|
||||
}
|
||||
events.push({ type: "run.completed", receipt: agentExecutionReceiptSchema.parse(options.receipt()) });
|
||||
return events.map((event) => consultationAgentPublicEventSchema.parse(event));
|
||||
}
|
||||
|
||||
type StreamAgentResponseOptions = EventOptions & {
|
||||
state: ConsultationRuntimeState;
|
||||
stream: ChunkStream;
|
||||
transformText?: (text: string) => string;
|
||||
requireTool: boolean;
|
||||
retry?: () => Promise<ChunkStream>;
|
||||
continueAfterDisconnect?: boolean;
|
||||
headers?: HeadersInit;
|
||||
onFirstActivity?: () => void | Promise<void>;
|
||||
onFirstOutput?: () => void | Promise<void>;
|
||||
onComplete?: (output: string, receipt: AgentExecutionReceipt) => void | Promise<void>;
|
||||
onError?: (error: unknown, emitted: boolean, output: string) => void | Promise<void>;
|
||||
onCancel?: (emitted: boolean) => void | Promise<void>;
|
||||
};
|
||||
|
||||
function contractReady(options: StreamAgentResponseOptions) {
|
||||
return options.state.jyotishSkillLoaded
|
||||
&& (!options.requireTool || (options.state.consultationToolCompleted && options.state.consultationToolCallCount === 1));
|
||||
}
|
||||
|
||||
export function streamAgentResponse(options: StreamAgentResponseOptions) {
|
||||
const encoder = new TextEncoder();
|
||||
let disconnected = false;
|
||||
let settled = false;
|
||||
let settling = false;
|
||||
let emitted = false;
|
||||
let firstActivity = false;
|
||||
let firstOutput = false;
|
||||
let fullOutput = "";
|
||||
const startedAt = new Map<string, number>();
|
||||
const jyotishSkillCallIds = new Set<string>();
|
||||
const send = (controller: ReadableStreamDefaultController<Uint8Array> | undefined, event: ConsultationAgentPublicEvent) => {
|
||||
if (!firstActivity && (event.type === "skill.started" || event.type === "tool.started" || event.type === "activity")) {
|
||||
firstActivity = true;
|
||||
void Promise.resolve(options.onFirstActivity?.()).catch(() => {});
|
||||
}
|
||||
if (!disconnected && controller) controller.enqueue(encoder.encode(`${JSON.stringify(consultationAgentPublicEventSchema.parse(event))}\n`));
|
||||
};
|
||||
|
||||
async function consumeAttempt(controller: ReadableStreamDefaultController<Uint8Array> | undefined, stream: ChunkStream) {
|
||||
const visible = createVisibleTextTransformer(options.transformText ?? ((value) => value));
|
||||
let held = "";
|
||||
let attemptOutput = "";
|
||||
let composingSent = false;
|
||||
const outputText = async (text: string) => {
|
||||
attemptOutput += text;
|
||||
held += text;
|
||||
if (!held || !contractReady(options)) return;
|
||||
if (!composingSent) {
|
||||
composingSent = true;
|
||||
send(controller, { type: "activity", phase: "answer-composition", label: "正在组织回答" });
|
||||
}
|
||||
if (!firstOutput && /\S/.test(held)) {
|
||||
firstOutput = true;
|
||||
await options.onFirstOutput?.();
|
||||
}
|
||||
send(controller, { type: "answer.delta", text: held });
|
||||
fullOutput += held;
|
||||
if (/\S/.test(held)) emitted = true;
|
||||
held = "";
|
||||
};
|
||||
for await (const chunk of readChunks(stream)) {
|
||||
for (const event of mapChunk(chunk, options, startedAt, jyotishSkillCallIds)) send(controller, event);
|
||||
if (chunk.type === "text-delta" && typeof chunk.payload?.text === "string") {
|
||||
await outputText(visible.push(chunk.payload.text));
|
||||
}
|
||||
}
|
||||
await outputText(visible.finish(""));
|
||||
return { held, attemptOutput };
|
||||
}
|
||||
|
||||
const body = new ReadableStream<Uint8Array>({
|
||||
start(controller) {
|
||||
void (async () => {
|
||||
send(controller, { type: "run.started", runId: options.runId, requestId: options.requestId });
|
||||
try {
|
||||
const first = await consumeAttempt(controller, options.stream);
|
||||
if (!contractReady(options) && options.retry) {
|
||||
if (options.state.steps.length < 32) {
|
||||
options.state.steps.push({ sequence: options.state.steps.length + 1, kind: "validation", name: "runtime-contract-retry", status: "completed" });
|
||||
}
|
||||
send(controller, { type: "activity", phase: "loading-method", label: "正在补齐方法与计算步骤" });
|
||||
await consumeAttempt(controller, await options.retry());
|
||||
}
|
||||
if (!contractReady(options)) throw new Error("runtime_contract_incomplete");
|
||||
if (!/\S/.test(fullOutput)) {
|
||||
if (/\S/.test(first.held) || /\S/.test(first.attemptOutput)) throw new Error("runtime_contract_incomplete");
|
||||
throw new Error("empty_answer");
|
||||
}
|
||||
settling = true;
|
||||
const receipt = agentExecutionReceiptSchema.parse(options.receipt());
|
||||
await options.onComplete?.(fullOutput, receipt);
|
||||
settled = true;
|
||||
settling = false;
|
||||
send(controller, { type: "run.completed", receipt });
|
||||
if (!disconnected) controller.close();
|
||||
} catch (error) {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
settling = false;
|
||||
try {
|
||||
await options.onError?.(error, emitted, fullOutput);
|
||||
} catch {}
|
||||
const code = error instanceof Error && error.message === "runtime_contract_incomplete"
|
||||
? "runtime_contract_incomplete" as const
|
||||
: error instanceof Error && error.message === "empty_answer"
|
||||
? "empty_answer" as const
|
||||
: "calculation_failed" as const;
|
||||
send(controller, { type: "run.failed", code, message: code === "runtime_contract_incomplete" ? "Agent 未完成必要的方法与计算步骤，本次不会扣点。" : "咨询暂时无法完成，本次不会扣点。" });
|
||||
if (!disconnected) controller.close();
|
||||
}
|
||||
})();
|
||||
},
|
||||
async cancel() {
|
||||
if (settled) return;
|
||||
if (settling || options.continueAfterDisconnect) {
|
||||
disconnected = true;
|
||||
return;
|
||||
}
|
||||
settled = true;
|
||||
await options.onCancel?.(emitted);
|
||||
},
|
||||
});
|
||||
return new Response(body, {
|
||||
headers: {
|
||||
"cache-control": "no-cache, no-transform",
|
||||
"content-type": "application/x-ndjson; charset=utf-8",
|
||||
"x-accel-buffering": "no",
|
||||
"x-ayanam-mode": "mastra-agentic",
|
||||
"x-ayanam-request-id": options.requestId,
|
||||
...options.headers,
|
||||
},
|
||||
});
|
||||
}
|
||||
@@ -31,7 +31,7 @@ function longestOpenerPrefixSuffix(value: string) {
|
||||
}
|
||||
|
||||
/** Sends only visible prose through the output guard and preserves metadata bytes. */
|
||||
function createVisibleTextTransformer(transform: (text: string) => string) {
|
||||
export function createVisibleTextTransformer(transform: (text: string) => string) {
|
||||
let rawBuffer = "";
|
||||
let visibleBuffer = "";
|
||||
let hiddenBuffer = "";
|
||||
|
||||
Reference in New Issue
Block a user