Real-Time AI Agent Event Stream with Kinesis: Build a Low-Latency LLM Chat Pipeline

Imagine your AI assistant responding to user messages in under 100 ms, no polling, no missed events. By streaming each utterance through Amazon Kinesis and a tiny Lambda consumer, you get a push‑based, scalable backbone for LLM calls. This pattern flips the classic request‑response model on its head.

In plain English: Instead of “client asks → server answers → client waits”, we push every user line into a fast conveyor belt (Kinesis) and let a worker pick it up instantly, reply, and push the answer back onto the belt.

---

Why Stream AI Agent Events with Kinesis?

When a chatbot receives a message, the traditional route is:

1. The front‑end makes an HTTP request.
2. The request hits an API Gateway → Lambda → LLM.
3. The response travels back the same path.

That round‑trip adds network hops, DNS lookups, and queue latency. If you replace the request‑response hop with an event stream, each utterance becomes a tiny packet that flows continuously, like cars on a highway that never stop at a toll booth.

Benefits that matter for a chat agent

Benefit	Why it helps the user	Why it helps the developer
Deterministic sub‑100 ms delivery	The user sees a reply almost instantly, keeping conversation natural.	You can reason about latency in tests because the stream guarantees ordering per shard.
Built‑in fan‑out	One message can be consumed by many workers (e.g., logging, analytics, moderation) without extra code.	No need for separate SQS queues or SNS topics.
No polling required	Workers don’t have to ask “any new data?” every second.	Reduces empty‑read cycles and saves compute credits.

Key takeaway: Kinesis turns “when will the next message arrive?” into “the next message arrives now”.

A simple analogy

Think of a mail slot (SQS) versus a conveyor belt (Kinesis). With a mail slot you must open the door, look inside, and take the letter—sometimes you open it to find nothing. With a conveyor belt, the letter slides right onto your desk the moment it’s placed on the belt, no extra checking needed.

---

Setting Up the Kinesis Data Stream

Before we can push chat utterances, we need a stream. A stream is a logical pipe that holds ordered records, split into shards (think of parallel lanes on a highway). Each shard can ingest up to 1 MiB/s of data (≈ 1000 records per second) and deliver up to 2 MiB/s to consumers.

Step 1: Create the stream with the AWS SDK for JavaScript v3

import { KinesisClient, CreateStreamCommand, PutRecordCommand } from "@aws-sdk/client-kinesis";

// 1️⃣ Create a Kinesis client that talks to the current region
const kinesis = new KinesisClient({});

// 2️⃣ Create a stream named "chat-input" with 2 shards (2 parallel lanes)
await kinesis.send(
  new CreateStreamCommand({
    StreamName: "chat-input",
    ShardCount: 2, // adjust based on expected traffic
  })
);
console.log("✅ Stream created");

// Helper: write a single chat utterance into the stream
export async function pushUserMessage(sessionId: string, userMessage: string) {
  // Payload we want to ship – JSON string → bytes
  const payload = Buffer.from(
    JSON.stringify({ sessionId, userMessage })
  );

  // PartitionKey decides which shard the record lands on.
  // Using sessionId spreads messages from the same conversation onto the same shard,
  // preserving order.
  await kinesis.send(
    new PutRecordCommand({
      StreamName: "chat-input",
      Data: payload,
      PartitionKey: sessionId,
    })
  );
}

What each line does

• KinesisClient – a thin wrapper around HTTP calls to the Kinesis service.
• CreateStreamCommand – asks AWS to allocate the pipe and its lanes.
• PutRecordCommand – drops one JSON packet onto the belt.
• PartitionKey – like a mailbox number; records with the same key go to the same lane, which guarantees ordering for a single conversation.

Tip: If you exceed the 1 MiB/s write limit, Kinesis will throttle writes and return a ProvisionedThroughputExceededException. Plan shard count accordingly.

Gotcha: shard limits at scale

Teams often start with a single shard, then see mysterious throttling once traffic grows. Remember: each shard is capped at 1 MiB/s write. If you expect 10 MiB/s, create at least 10 shards (or more, to leave headroom).

---

Lambda Consumer: Decode, Call Claude via Bedrock

Now we need a worker that pulls records, calls an LLM, and pushes the reply back onto another stream (e.g., chat-output). In Node.js the Lambda handler runs whenever new records arrive.

The iterator problem

Kinesis uses a shard iterator – a cursor that points to the next record to read. An iterator is only valid for 5 minutes. If you reuse an old iterator, GetRecords returns an empty set even though data sits in the stream. The fix is to request a fresh iterator on every Lambda invocation.

Full Lambda code (TypeScript)

import {
  KinesisClient,
  GetShardIteratorCommand,
  GetRecordsCommand,
  PutRecordCommand,
} from "@aws-sdk/client-kinesis";
import {
  BedrockClient,
  InvokeModelCommand,
} from "@aws-sdk/client-bedrock-runtime"; // Bedrock SDK (v3)
import { Context, Handler } from "aws-lambda";

// Clients are created outside the handler so they are reused across invocations
const kinesis = new KinesisClient({});
const bedrock = new BedrockClient({});

// Names of the streams we use
const INPUT_STREAM = "chat-input";
const OUTPUT_STREAM = "chat-output";

// Helper: fetch a fresh iterator for a given shard
async function getIterator(shardId: string) {
  const resp = await kinesis.send(
    new GetShardIteratorCommand({
      StreamName: INPUT_STREAM,
      ShardId: shardId,
      ShardIteratorType: "LATEST", // start at newest record
    })
  );
  return resp.ShardIterator!;
}

// Lambda entry point
export const handler: Handler = async (event: any, _ctx: Context) => {
  // The event we receive is a Kinesis event payload generated by the service.
  // It already contains Records, but we will also demonstrate a manual pull
  // to illustrate iterator refresh logic.

  for (const record of event.Records) {
    // 1️⃣ Decode the base64 payload that Kinesis gives us
    const payload = Buffer.from(record.kinesis.data, "base64").toString("utf-8");
    const { sessionId, userMessage } = JSON.parse(payload);
    console.log(`👂 Received message from ${sessionId}: ${userMessage}`);

    // 2️⃣ Call Claude (or any Bedrock model) via invokeModel
    const invokeResp = await bedrock.send(
      new InvokeModelCommand({
        ModelId: "anthropic.claude-v2", // example model identifier
        ContentType: "application/json",
        Accept: "application/json",
        Body: JSON.stringify({
          prompt: userMessage,
          max_tokens_to_sample: 256,
        }),
      })
    );

    // 3️⃣ Parse the model’s answer
    const modelOutput = JSON.parse(Buffer.from(invokeResp.Body!).toString("utf-8"));
    const assistantReply = modelOutput.completion?.trim() ?? "I’m sorry, I didn’t understand.";

    console.log(`🤖 Claude replied: ${assistantReply}`);

    // 4️⃣ Push the reply onto the output stream for downstream consumers
    await kinesis.send(
      new PutRecordCommand({
        StreamName: OUTPUT_STREAM,
        PartitionKey: sessionId, // keep same ordering for the conversation
        Data: Buffer.from(
          JSON.stringify({ sessionId, assistantReply })
        ),
      })
    );
  }

  // --- Manual iterator refresh example (useful if you run a while‑loop consumer) ---
  // Assume we know the shard ID (e.g., from environment variable)
  const shardId = process.env.SHARD_ID!;
  const iterator = await getIterator(shardId);
  const recordsResp = await kinesis.send(
    new GetRecordsCommand({ ShardIterator: iterator, Limit: 10 })
  );

  // If the iterator were stale, `recordsResp.Records` would be empty even though data exists.
  // Refreshing every call avoids that silent failure.
  // -------------------------------------------------------------------------------

  return { statusCode: 200 };
};

Explanation of key lines

• record.kinesis.data – Kinesis delivers each record as a base64 string; we decode it.
• InvokeModelCommand – Bedrock’s API to run a Large Language Model (LLM). The request body follows the model’s JSON schema.
• PutRecordCommand on OUTPUT_STREAM – pushes the assistant’s reply back onto a second stream so other services (analytics, UI push, logging) can consume it without touching the Lambda again.
• The iterator refresh block shows the pattern you must adopt if you ever write a long‑running consumer (e.g., a container that reads continuously).

Helpful tip: When you enable Enhanced Fan‑Out (next section), Lambda receives records via an event source mapping and you don’t need to manage iterators yourself. However, if you ever write a custom consumer, always request a new iterator on each loop.

Lambda‑specific gotchas

• require(esm) in Node 22 – If you depend on the esm package, Lambda layers compiled for Node 18 will silently fail. Stick to native ESM (import …) or upgrade your layer.
• SnapStart + VPC – SnapStart speeds up cold starts by serializing the execution environment, but when a Lambda sits inside a VPC the time spent attaching ENIs dominates. You won’t see savings unless the function is outside a VPC.
• Response streaming – If you later need to stream the assistant’s reply directly to a client (instead of writing to a stream), set Content-Type: text/event-stream and return the body as a readable stream; otherwise Lambda buffers the whole response.
• Provisioned Concurrency – Guarantees a warm container but charges per‑hour even when idle. For a low‑traffic chatbot, on‑demand scaling is usually cheaper.

---

Enhanced Fan‑Out for Sub‑millisecond Latency

Standard Kinesis consumption works like a single bus that all consumers share; the bus capacity (2 MiB/s per shard) is split among them. Enhanced fan‑out gives each consumer its own 2 MiB/s pipe, removing contention and cutting per‑record latency to < 1 ms on the AWS side.

How it works

When you register a consumer with RegisterStreamConsumer, Kinesis creates a dedicated shard‑level endpoint. The consumer then reads via SubscribeToShard, which pushes records to the consumer as soon as they land on the shard. No polling, no empty reads.

Registering a Lambda as an enhanced fan‑out consumer

import {
  KinesisClient,
  RegisterStreamConsumerCommand,
  DescribeStreamConsumerCommand,
} from "@aws-sdk/client-kinesis";

const kinesis = new KinesisClient({});

async function ensureConsumer() {
  const consumerName = "chat-lambda-consumer";

  // 1️⃣ Try to describe an existing consumer (idempotent)
  const describe = await kinesis.send(
    new DescribeStreamConsumerCommand({
      StreamARN: "arn:aws:kinesis:us-east-1:123456789012:stream/chat-input",
      ConsumerName: consumerName,
    })
  ).catch(() => null);

  if (describe?.ConsumerDescription?.ConsumerARN) {
    console.log("✅ Consumer already exists");
    return describe.ConsumerDescription.ConsumerARN;
  }

  // 2️⃣ Register a new consumer – this is the step that enables enhanced fan‑out
  const register = await kinesis.send(
    new RegisterStreamConsumerCommand({
      StreamARN: "arn:aws:kinesis:us-east-1:123456789012:stream/chat-input",
      ConsumerName: consumerName,
    })
  );

  console.log("🛎️ Registered enhanced fan‑out consumer");
  return register.Consumer?.ConsumerARN!;
}

// Call during deployment (CDK, SAM, or manual script)
ensureConsumer().then((arn) => console.log(`Consumer ARN: ${arn}`));

Key points

• RegisterStreamConsumerCommand creates a dedicated delivery channel for the Lambda.
• The consumer name must be unique per stream; re‑running the script is safe because we first try to describe it.
• Once the consumer exists, you attach it to the Lambda via an event source mapping (in the console or via IaC). The mapping will automatically use the fan‑out endpoint.

In plain English: Enhanced fan‑out gives each worker its own private lane on the highway, so no traffic jam ever slows you down.

Performance impact

• Standard mode: Lambda polls every ~5 seconds, incurring ~50–100 ms of added latency.
• Enhanced fan‑out: Records arrive as soon as they’re written; typical end‑to‑end latency (write → Lambda start → LLM call) can be under 100 ms if the LLM itself is fast.

Remember the shard limits – each enhanced consumer still respects the 2 MiB/s read per shard, but now the limit is per consumer instead of shared. If you plan many parallel agents, allocate more shards accordingly.

---

End‑to‑End Flow Diagram and Deployment

Below is a textual flow that you can paste into a markdown diagram tool (e.g., Mermaid) if you like visualizing it later.

sequenceDiagram
    participant UI as Front‑end (Web / Mobile)
    participant K1 as Kinesis (chat-input)
    participant L as Lambda (consumer)
    participant B as Bedrock (Claude)
    participant K2 as Kinesis (chat-output)
    participant downstream as Downstream services

    UI->>K1: PutRecord({sessionId, userMessage})
    K1->>L: Push (Enhanced fan‑out)
    L->>B: invokeModel(userMessage)
    B->>L: Model response
    L->>K2: PutRecord({sessionId, assistantReply})
    K2->>downstream: Subscribe (analytics, UI push, etc.)

Deploying with the AWS CDK (TypeScript)

Only a few lines are needed because the heavy lifting (stream creation, consumer registration) is done by the CDK constructs.

import * as cdk from "aws-cdk-lib";
import { Construct } from "constructs";
import * as kinesis from "aws-cdk-lib/aws-kinesis";
import * as lambda from "aws-cdk-lib/aws-lambda";
import * as lambdaEventSources from "aws-cdk-lib/aws-lambda-event-sources";

export class ChatPipelineStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    // 1️⃣ Input stream (enhanced fan‑out enabled by default)
    const input = new kinesis.Stream(this, "ChatInput", {
      streamName: "chat-input",
      shardCount: 2, // adjust based on traffic
    });

    // 2️⃣ Output stream for replies
    const output = new kinesis.Stream(this, "ChatOutput", {
      streamName: "chat-output",
      shardCount: 1,
    });

    // 3️⃣ Lambda that consumes the input stream
    const consumerFn = new lambda.Function(this, "ChatConsumer", {
      runtime: lambda.Runtime.NODEJS_22_X,
      handler: "index.handler",
      code: lambda.Code.fromAsset("lambda"), // folder with the TypeScript compiled code
      memorySize: 256,
      timeout: cdk.Duration.seconds(10),
    });

    // 4️⃣ Attach the Lambda as an enhanced fan‑out consumer
    consumerFn.addEventSource(
      new lambdaEventSources.KinesisEventSource(input, {
        startingPosition: lambda.StartingPosition.LATEST,
        // Setting `batchSize` to 1 ensures we process each utterance immediately.
        batchSize: 1,
        // `bisectBatchOnFunctionError` = false because we want atomic processing.
        bisectBatchOnFunctionError: false,
        // Enhanced fan‑out is true by default; we make it explicit.
        maxBatchingWindow: cdk.Duration.millis(0),
        enabled: true,
      })
    );

    // 5️⃣ Grant the Lambda permissions to read from input and write to output
    input.grantRead(consumerFn);
    output.grantWrite(consumerFn);
  }
}

Why each piece matters

• shardCount – directly maps to the 1 MiB/s write ceiling; more shards = more throughput.
• batchSize: 1 – forces the Lambda to receive one record at a time, reducing queuing delay.
• maxBatchingWindow: 0 ms – disables the small time window that would otherwise wait to fill a batch, keeping latency low.
• Permissions (grantRead, grantWrite) are required for the Lambda to call GetRecords and PutRecord without manual IAM policy crafting.

Pro tip: If you ever need to add more shards after launch, use the UpdateShardCount API. Remember to adjust your downstream consumers to handle the new shard IDs.

---

The Takeaway

What you now have in your toolbox

• Streaming chat events with Kinesis removes polling latency and guarantees order per conversation.
• Enhanced fan‑out gives each Lambda its own private delivery lane, shrinking end‑to‑end latency to sub‑100 ms in most cases.
• A shard iterator expires after 5 minutes; always request a fresh iterator when you write a custom consumer.
• The Lambda consumer decodes the JSON payload, calls Claude (or any Bedrock model) via invokeModel, and writes the response back to a second stream.
• Common gotchas: shard throughput limits, iterator expiration, Node 22 require(esm) issues, SnapStart + VPC mismatch, and provisioned concurrency cost surprises.
• Deploying with CDK (or SAM) makes the whole pipeline repeatable and lets you version‑control the architecture.

Armed with this pattern you can replace a traditional request‑response chatbot with a push‑based, horizontally scalable pipeline that feels instantaneous to your users. Happy streaming!

---

Transparency notice

This article was written with the help of an AI system — Groq (GPT OSS 120B).

Published: 2026-09-07 · 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.