用 Kinesis Data Streams 流式传输 Claude Agent 事件

Real‑time LLM interactions feel like they belong on websockets, but a Kinesis Data Stream gives you durable, auto‑scaling fan‑out with millisecond latency. In this guide you’ll see how to pipe Claude’s token stream through Kinesis and a Node.js 22 Lambda, turning a flaky WebSocket demo into an enterprise‑grade pipeline. Why Kinesis Beats WebSockets for LLM Streams When you ask Claude for a response, the model sends back tokens (tiny pieces of text) as they are generated. A WebSocket connection can push each token straight to the browser, but it has three hidden costs: Ephemeral connection – if the client disconnects, the in‑flight tokens are lost. Single consumer – only the WebSocket server receives the data; you can’t add a logger, analytics, or a backup consumer without duplicating work. Scaling limits – a single server must handle every concurrent chat, which becomes fragile under load. A Kinesis Data Stream is a managed pipe that breaks the data into shards (independent write/read lanes). Each shard can feed many consumers at the same time (this is called enhanced fan‑out). Think of a water main (the stream) that splits into many smaller hoses (the consumers). Even if one hose is turned off, the water still flows through the main pipe. In plain English – Kinesis lets you treat each token as a durable record that any number of services can read, while still keeping latency low enough for a chat‑like experience. Quick code sketch: putting a token into Kinesis // npm install @aws-sdk/client-kinesis import { KinesisClient, PutRecordCommand } from "@aws-sdk/client-kinesis"; const client = new KinesisClient({ region: "us-east-1" }); async function putToken(streamName, partitionKey, token) { // The token is a short string, we store it as UTF‑8 bytes. const command = new PutRecordCommand({ StreamName: streamName, PartitionKey: partitionKey, // decides which shard receives the record Data: Buffer.from(token), // binary payload }); await client.send(command); } Enter fullscreen mode Exit fullscreen mode Key takeaway – With a few lines you can write a token to a durable, auto‑scaled pipe that any Lambda can later read. Setting Up a Kinesis Data Stream for Claude Tokens What you need to know first Shard – a unit of capacity; each shard can ingest up to 1 MiB / second (≈ 1000 records) and serve up to 2 MiB / second to consumers. Enhanced fan‑out – a feature that gives each consumer its own read throughput, avoiding the 2 MiB shared limit. It only works when the consumer is registered by ARN (Amazon Resource Name). Forgetting this registration falls back to the shared 2 MiB pool and adds ~200 ms latency. Iterator – a pointer that tells Kinesis where to start reading. It expires after 5 minutes, so a consumer must refresh it regularly. Step‑by‑step creation with the SDK import { KinesisClient, CreateStreamCommand, RegisterStreamConsumerCommand, } from "@aws-sdk/client-kinesis"; const kinesis = new KinesisClient({ region: "us-east-1" }); const STREAM_NAME = "claude-token-stream"; const CONSUMER_NAME = "claude-lambda-consumer"; async function createStream() { // Create a stream with 2 shards (adjust based on expected throughput) await kinesis.send( new CreateStreamCommand({ StreamName: STREAM_NAME, ShardCount: 2 }) ); console.log(`Stream ${STREAM_NAME} created`); } async function registerConsumer() { // Registering by ARN enables enhanced fan‑out await kinesis.send( new RegisterStreamConsumerCommand({ StreamARN: `arn:aws:kinesis:us-east-1:123456789012:stream/${STREAM_NAME}`, ConsumerName: CONSUMER_NAME, }) ); console.log(`Consumer ${CONSUMER_NAME} registered for enhanced fan‑out`); } // Run once (or guard with existence checks) await createStream(); await registerConsumer(); Enter fullscreen mode Exit fullscreen mode Tip – After creating the stream, give it a minute to become ACTIVE before you register the consumer; otherwise the registration request will fail. Gotchas to keep in mind Issue Why it hurts How to avoid Shard write limit (1 MiB/s) Burst of tokens can exceed capacity and cause throttling Start with 2‑4 shards; monitor WriteProvisionedThroughputExceeded metrics Iterator expiration (5 min) A stale iterator makes GetRecords return ExpiredIteratorException Refresh iterator on every loop iteration or catch the error and request a new one Empty GetRecords response Kinesis is eventually consistent; records may appear a few milliseconds later Loop with a short sleep (e.g., 200 ms) when no records are returned Lambda Consumer with Node.js 22: Decoding and Forwarding Why a Lambda is a good fit A Lambda is a serverless function that runs only when invoked. By attaching it to the Kinesis stream as an event source, AWS automatically polls the stream, passes batches of records to your code, and handles scaling for you. Node.js 22 brings the native fetch API, so we can forward each token to a browser using Server‑Sent Events (SSE) without adding extra libraries. Minimal Lambda handler // npm install @aws-sdk/client-lambda @aws-sdk/client-kinesis import { KinesisClient, GetRecordsCommand, GetShardIteratorCommand } from "@aws-sdk/client-kinesis"; const kinesis = new KinesisClient({ region: "us-east-1" }); const STREAM_NAME = process.env.STREAM_NAME; const CONSUMER_ARN = process.env.CONSUMER_ARN; // ARN of the enhanced fan‑out consumer // Exponential backoff helper function wait(ms) { return new Promise(res => setTimeout(res, ms)); } export const handler = async (event, context) => { // 1️⃣ Get a shard iterator for each shard (enhanced fan‑out) const shards = await kinesis.send( new DescribeStreamCommand({ StreamName: STREAM_NAME }) ).then(resp => resp.StreamDescription.Shards); for (const shard of shards) { const iteratorCmd = new GetShardIteratorCommand({ StreamName: STREAM_NAME, ShardId: shard.ShardId, ShardIteratorType: "LATEST", // start at newest token ConsumerARN: CONSUMER_ARN, // enables enhanced fan‑out }); let iterator = await kinesis.send(iteratorCmd).then(r => r.ShardIterator); // 2️⃣ Pull records in a loop (Lambda will be terminated after timeout) while (true) { try { const recordsCmd = new GetRecordsCommand({ ShardIterator: iterator, Limit: 100 }); const { Records, NextShardIterator } = await kinesis.send(recordsCmd); iterator = NextShardIterator; if (Records.length === 0) { // No data right now, pause briefly to avoid tight looping await wait(200); continue; } // 3️⃣ Forward each token via SSE to the client for (const rec of Records) { const token = Buffer.from(rec.Data).toString("utf-8"); await forwardTokenSSE(token); } } catch (err) { if (err.name === "ProvisionedThroughputExceededException") { // 4️⃣ Throttling – back off exponentially for (let attempt = 1; attempt { const body = JSON.parse(event.body); const { prompt, userId } = body; // simple payload from the browser // Call Claude with streaming enabled const response = await fetch(CLAUDE_API, { method: "POST", headers: { "Content-Type": "application/json", "x-api-key": process.env.CLAUDE_API_KEY, }, body: JSON.stringify({ model: "claude-3-5-sonnet-20241001", prompt, max_tokens_to_sample: 1024, stream: true, }), }); // Read the raw byte stream; each line is a JSON fragment with a token const reader = response.body.getReader(); const decoder = new TextDecoder(); while (true) { const { done, value } = await reader.read(); if (done) break; // Split on newline because Claude sends each token on its own line const chunk = decoder.decode(value); const lines = chunk.split("\n").filter(l => l.trim()); for (const line of lines) { const payload = JSON.parse(line); const token = payload.completion; // token string // Write token to Kinesis – partition by userId to keep order per chat await kinesis.send( new PutRecordCommand({ StreamName: STREAM_NAME, PartitionKey: userId, Data: Buffer.from(token), }) ); } } // Respond quickly; the real‑time work continues in the consumer Lambda return { statusCode: 200, body: JSON.stringify({ message: "Chat started" }), }; }; Enter fullscreen mode Exit fullscreen mode In plain English – This Lambda acts like a mailroom clerk: it receives Claude’s letters (tokens), stamps each with the user’s ID (the partition key), and drops them into the Kinesis mailbox for the delivery Lambda to pick up. Common pitfalls for this proxy Missing stream:true – Claude will wait until the whole answer is ready, defeating the real‑time goal. Incorrect partition key – If you use a constant key, all chats share a single shard, quickly hitting the 1 MiB/s write ceiling. Forgot to set CONTENT_TYPE: text/event-stream on the SSE endpoint – browsers will buffer the response, appearing as if nothing is happening. Handling Errors and Observability Why observability matters A streaming pipeline has many moving parts: Claude API latency, Kinesis throttling, Lambda timeouts, and network hiccups. Without visibility you’ll see occasional “missing tokens” but not know why. Simple CloudWatch logging and metrics import { CloudWatchClient, PutMetricDataCommand } from "@aws-sdk/client-cloudwatch"; const cw = new CloudWatchClient({ region: "us-east-1" }); const METRIC_NAMESPACE = "ClaudeKinesisPipeline"; async function recordMetric(name, value = 1) { await cw.send( new PutMetricDataCommand({ Namespace: METRIC_NAMESPACE, MetricData: [ { MetricName: name, Value: value, Unit: "Count", }, ], }) ); } Enter fullscreen mode Exit fullscreen mode Insert recordMetric("TokensProcessed") after each successful token forward, and recordMetric("ThrottleBackoff") inside the backoff catch block. CloudWatch dashboards can then show spikes. Dealing with iterator expiration if (err.name === "ExpiredIteratorException") { // Log for visibility console.warn("Shard iterator expired, requesting a fresh one"); // Re‑fetch the iterator (same code as initial request) const fresh = await kinesis.send(iteratorCmd); iterator = fresh.ShardIterator; } Enter fullscreen mode Exit fullscreen mode Dead‑letter queue (DLQ) for irrecoverable records If a record repeatedly triggers ProvisionedThroughputExceededException after several retries, move it to an SQS queue for later analysis. This prevents the Lambda from getting stuck on a single bad shard. Key tip – Always set a maximum retry count (e.g., 5) before sending to DLQ; otherwise exponential backoff can keep the function running until it times out, inflating costs. The Takeaway You now have a complete, production‑grade pattern for streaming Claude’s token output through Kinesis and a Node.js 22 Lambda. Kinesis provides durability and multi‑consumer fan‑out, eliminating the fragility of a single WebSocket server. Enhanced fan‑out only works when you register the consumer by ARN; forgetting this leaves you with higher latency and extra read costs. Shards are the unit of capacity; start with multiple shards and monitor throughput to avoid throttling. Lambda can read directly from Kinesis, decode token bytes, and forward them with SSE using the built‑in fetch API. Handle ProvisionedThroughputExceededException with exponential backoff and refresh stale shard iterators to keep the pipeline alive. Add lightweight CloudWatch metrics and a DLQ to gain visibility and prevent silent data loss. With these pieces in place, you can replace a brittle WebSocket demo with a scalable, server‑less architecture that works the same way for a handful of users or for thousands of concurrent chats. Happy streaming! Transparency notice This article was written with the help of an AI system — Groq (GPT OSS 120B). Published: 2026-10-23 · Primary focus: Kinesis All code blocks are intended to be correct and runnable, but please verify them against the official docs for the tools mentioned before using in production. Find an error? Drop a comment — corrections are always welcome.