Durable AI エージェントを構築する

import { AgentWorkflow } from "agents/workflows";
import type { AgentWorkflowEvent, AgentWorkflowStep } from "agents/workflows";
import Anthropic from "@anthropic-ai/sdk";
import {
	tools,
	searchReposTool,
	getRepoTool,
	type SearchReposInput,
	type GetRepoInput,
} from "./tools";
import type { ResearchAgent } from "./agent";

type Params = { task: string };

export class ResearchWorkflow extends AgentWorkflow<ResearchAgent, Params> {
	async run(event: AgentWorkflowEvent<Params>, step: AgentWorkflowStep) {
		const client = new Anthropic({ apiKey: this.env.ANTHROPIC_API_KEY });

		const messages: Anthropic.MessageParam[] = [
			{ role: "user", content: event.payload.task },
		];

		const toolDefinitions = tools.map(({ run, ...rest }) => rest);

		// Durable agent loop - each turn is checkpointed
		for (let turn = 0; turn < 10; turn++) {
			// Report progress to Agent and connected clients
			await this.reportProgress({
				step: `llm-turn-${turn}`,
				status: "running",
				percent: turn / 10,
				message: `Processing turn ${turn + 1}...`,
			});

			const response = (await step.do(
				`llm-turn-${turn}`,
				{ retries: { limit: 3, delay: "10 seconds", backoff: "exponential" } },
				async () => {
					const msg = await client.messages.create({
						model: "claude-sonnet-4-5-20250929",
						max_tokens: 4096,
						tools: toolDefinitions,
						messages,
					});
					// Serialize for Workflow state
					return JSON.parse(JSON.stringify(msg));
				},
			)) as Anthropic.Message;

			if (!response || !response.content) continue;

			messages.push({ role: "assistant", content: response.content });

			if (response.stop_reason === "end_turn") {
				const textBlock = response.content.find(
					(b): b is Anthropic.TextBlock => b.type === "text",
				);
				const result = {
					status: "complete",
					turns: turn + 1,
					result: textBlock?.text ?? null,
				};

				// Report completion (durable)
				await step.reportComplete(result);
				return result;
			}

			const toolResults: Anthropic.ToolResultBlockParam[] = [];

			for (const block of response.content) {
				if (block.type !== "tool_use") continue;

				// Broadcast tool execution to clients
				this.broadcastToClients({
					type: "tool_call",
					tool: block.name,
					turn,
				});

				const result = await step.do(
					`tool-${turn}-${block.id}`,
					{ retries: { limit: 2, delay: "5 seconds" } },
					async () => {
						switch (block.name) {
							case "search_repos":
								return searchReposTool.run(block.input as SearchReposInput);
							case "get_repo":
								return getRepoTool.run(block.input as GetRepoInput);
							default:
								return `Unknown tool: ${block.name}`;
						}
					},
				);

				toolResults.push({
					type: "tool_result",
					tool_use_id: block.id,
					content: result,
				});
			}

			messages.push({ role: "user", content: toolResults });
		}

		return { status: "max_turns_reached", turns: 10 };
	}
}

import { Agent } from "agents";

type State = {
	currentWorkflow?: string;
	status?: string;
};

export class ResearchAgent extends Agent<Env, State> {
	initialState: State = {};

	// Start a research task - called via HTTP or WebSocket
	async startResearch(task: string) {
		const instanceId = await this.runWorkflow("RESEARCH_WORKFLOW", { task });
		this.setState({
			...this.state,
			currentWorkflow: instanceId,
			status: "running",
		});
		return { instanceId };
	}

	// Get status of a workflow
	async getResearchStatus(instanceId: string) {
		return this.getWorkflow(instanceId);
	}

	// Called when workflow reports progress
	async onWorkflowProgress(
		workflowName: string,
		instanceId: string,
		progress: unknown,
	) {
		// Broadcast to all connected WebSocket clients
		this.broadcast(JSON.stringify({ type: "progress", instanceId, progress }));
	}

	// Called when workflow completes
	async onWorkflowComplete(
		workflowName: string,
		instanceId: string,
		result?: unknown,
	) {
		this.setState({ ...this.state, status: "complete" });
		this.broadcast(JSON.stringify({ type: "complete", instanceId, result }));
	}

	// Called when workflow errors
	async onWorkflowError(
		workflowName: string,
		instanceId: string,
		error: string,
	) {
		this.setState({ ...this.state, status: "error" });
		this.broadcast(JSON.stringify({ type: "error", instanceId, error }));
	}
}

import { getAgentByName, routeAgentRequest } from "agents";

export { ResearchAgent } from "./agent";
export { ResearchWorkflow } from "./workflow";

export default {
	async fetch(request: Request, env: Env): Promise<Response> {
		const url = new URL(request.url);

		// Route WebSocket connections to /agents/research-agent/{name}
		const agentResponse = await routeAgentRequest(request, env);
		if (agentResponse) return agentResponse;

		// HTTP API for starting research tasks
		if (request.method === "POST" && url.pathname === "/research") {
			const { task, agentId } = await request.json<{
				task: string;
				agentId?: string;
			}>();

			// Get agent instance by name (creates if doesn't exist)
			const agent = await getAgentByName(
				env.ResearchAgent,
				agentId ?? "default",
			);

			// Start the research workflow via RPC
			const result = await agent.startResearch(task);
			return Response.json(result);
		}

		// Check workflow status
		if (url.pathname === "/status") {
			const instanceId = url.searchParams.get("instanceId");
			const agentId = url.searchParams.get("agentId") ?? "default";

			if (!instanceId) {
				return Response.json({ error: "instanceId required" }, { status: 400 });
			}

			const agent = await getAgentByName(env.ResearchAgent, agentId);
			const status = await agent.getResearchStatus(instanceId);

			return Response.json(status);
		}

		return new Response("POST /research with { task } to start", {
			status: 400,
		});
	},
} satisfies ExportedHandler<Env>;