onErrorexport class TextGenerator extends WorkflowEntrypoint<CloudflareBindings> {
async run(
event: Readonly<WorkflowEvent<{ prompt: string }>>,
step: WorkflowStep,
) {
const { prompt } = event.payload;
return await step.do(
"generate-text",
{
retries: { limit: 3, delay: 1000, backoff: "exponential" },
onError: async (error, retry) => {
if (error instanceof AITimeoutError) {
// Complex model timed out, try with a faster one first
try {
return await generateText({
model: "gpt-3.5-turbo",
prompt,
timeout: 5000,
});
} catch {
// If faster model fails, retry with original complex model (retry is the original callback in step.do)
return retry();
}
}
throw error;
}
},
M
onError
DWORKFLOW_SERVICE Env variabe as? D
D✘ [ERROR] TypeError: The RPC receiver does not implement the method "createInstance"
S
yieldEvent is outside of a step - meaning that these calls can be repeated multiple times (not sure if that's you want or not)MM
FF
T
internal error before the workflow tries to execute any steps. Any ideas? 
T
TWorkflowEntrypoint and bindings are accessed via this.env rather than just env
D^[a-zA-Z0-9_][a-zA-Z0-9-_]*$ (https://developers.cloudflare.com/workflows/reference/limits)
RS
MMM
5e3dea7b807381cba36b9292bde14920
R1ac93aea0842f2d3ef10111b5bf7bb11M
Wnpx wrangler types supposed to work for workflows?services = [{ binding = "WORKFLOW_PROCESS_TRANSACTIONS", service = "money" }]wrangler 3.107.3MM
Ewrangler workflows instances list <YOUR_WORKFLOW_NAME> | grep "▶ Running" -B1 | grep -Eo '([a-f0-9\-]{36})' | xargs -I {} wrangler workflows instances terminate <YOUR_WORKFLOW_NAME> {} definitly helped me. You can also do xargs -I {} -P 10 sh -c to parallelizeR
R9 running in the screenshot above, no workflows had run in since Friday. Workflow metrics show 0 instances and 0 steps. Please, help!Mproducer worker where you send the events, and it then places them into the queue.main step, which contains all the rest of the steps as substep. That is - it iterates over the batch of received events one by one and all the rest is the same as before. MM
T
C
CRMS "services": [
{
"binding": "WORKFLOW_SERVICE",
"service": "wf-portcity-ai"
}
],WORKFLOW_SERVICE✘ [ERROR] TypeError: The RPC receiver does not implement the method "createInstance"yieldEventinternal errorWorkflowEntrypointthis.envenv^[a-zA-Z0-9_][a-zA-Z0-9-_]*$5e3dea7b807381cba36b9292bde149201ac93aea0842f2d3ef10111b5bf7bb11npx wrangler typesservices = [{ binding = "WORKFLOW_PROCESS_TRANSACTIONS", service = "money" }]wrangler 3.107.3interface Env {
SESSION_SECRET: string;
DB: D1Database;
WORKFLOW_PROCESS_TRANSACTIONS: Fetcher;
}wrangler workflows instances list <YOUR_WORKFLOW_NAME> | grep "▶ Running" -B1 | grep -Eo '([a-f0-9\-]{36})' | xargs -I {} wrangler workflows instances terminate <YOUR_WORKFLOW_NAME> {}xargs -I {} -P 10 sh -c9 runningproducermainexport class WORKFLOW_WITH_BATCHED_EVENTS extends WorkflowEntrypoint<Env, BatchPayload> {
async run(event: WorkflowEvent<BatchPayload>, step: WorkflowStep) {
// Can access bindings on `this.env`
// Can access params on `event.payload`
const ctx = this.ctx;
const events = event.payload.data;
// THIS IS THE MAIN STEP, WHICH CONTAINS THE ACTUAL SUB-STEPS, WHAT I MENTIONED BEFORE.
const process_events = await step.do("process_events",
async () => {
/// ITERATE OVER THE EVENTS ONE-BY-ONE
for(const key in events){
const e = events[key];
//SUBSTEPS - Process as usual.
const fbcapi = await step.do(`your_substep_name`,
async () => { ... }
)
//.... the rest of your wf logic
}
//...
})
}export class RAGWorkflow {
async run(event, step) {
const { text } = event.params;
const record = await step.do('create database record', async () => {
const query = 'INSERT INTO notes (text) VALUES (?) RETURNING *';
const { results } = await env.DB.prepare(query).bind(text).run();
const record = results[0];
if (!record) throw new Error('Failed to create note');
return record;
});
const embedding = await step.do('generate embedding', async () => {
const embeddings = await env.AI.run('@cf/baai/bge-base-en-v1.5', { text: text });
const values = embeddings.data[0];
if (!values) throw new Error('Failed to generate vector embedding');
return values;
});
await step.do('insert vector', async () => {
return env.VECTOR_INDEX.upsert([
{
id: record.id.toString(),
values: embedding,
},
]);
});
}
}interface CycleParams<
I extends Rpc.Serializable<I> = Rpc.Serializable<any>,
O extends Rpc.Serializable<O> = Rpc.Serializable<any>,
> {
name: string;
do: {
name: string;
callback: (data: I) => Promise<O>;
};
sleep?: {
name: string;
duration: WorkflowSleepDuration;
};
}
class Cycle<T> extends WorkflowEntrypoint<Env, CycleParams[]> {
async run(event: WorkflowEvent<CycleParams[]>, step: WorkflowStep) {
const { instanceId, payload, timestamp } = event;
const data: any[] = [null];
for (const [index, cycleStep] of payload.entries()) {
if (cycleStep.sleep)
await step.sleep(cycleStep.sleep.name, cycleStep.sleep.duration);
data[index + 1] = await step.do(cycleStep.do.name, () => {
if (data.length > index + 1) return data[index + 1];
return cycleStep.do.callback(data[index]);
});
}
return data[data.length - 1] as T;
}
}