From 5b74b420747ed8e6c99e39d36a079a8c84c577c7 Mon Sep 17 00:00:00 2001 From: Doug Perkes Date: Fri, 17 Jul 2026 09:21:57 -0500 Subject: [PATCH] New serverless pattern - sam-dotnet-responsestreaming-bedrock-multiturn --- .../.gitignore | 16 ++ .../README.md | 201 +++++++++++++++ .../events/continue-trip.json | 24 ++ .../events/new-trip.json | 24 ++ ...t-responsestreaming-bedrock-multiturn.json | 68 +++++ .../src/Models.cs | 37 +++ .../src/Program.cs | 240 ++++++++++++++++++ .../src/ResponseFormatter.cs | 55 ++++ .../src/SessionStore.cs | 228 +++++++++++++++++ .../src/ToolDefinitions.cs | 87 +++++++ .../src/TripPlannerAgent.cs | 216 ++++++++++++++++ .../src/TripPlannerMultiturn.csproj | 20 ++ .../template.yaml | 118 +++++++++ 13 files changed, 1334 insertions(+) create mode 100644 sam-dotnet-responsestreaming-bedrock-multiturn/.gitignore create mode 100644 sam-dotnet-responsestreaming-bedrock-multiturn/README.md create mode 100644 sam-dotnet-responsestreaming-bedrock-multiturn/events/continue-trip.json create mode 100644 sam-dotnet-responsestreaming-bedrock-multiturn/events/new-trip.json create mode 100644 sam-dotnet-responsestreaming-bedrock-multiturn/sam-dotnet-responsestreaming-bedrock-multiturn.json create mode 100644 sam-dotnet-responsestreaming-bedrock-multiturn/src/Models.cs create mode 100644 sam-dotnet-responsestreaming-bedrock-multiturn/src/Program.cs create mode 100644 sam-dotnet-responsestreaming-bedrock-multiturn/src/ResponseFormatter.cs create mode 100644 sam-dotnet-responsestreaming-bedrock-multiturn/src/SessionStore.cs create mode 100644 sam-dotnet-responsestreaming-bedrock-multiturn/src/ToolDefinitions.cs create mode 100644 sam-dotnet-responsestreaming-bedrock-multiturn/src/TripPlannerAgent.cs create mode 100644 sam-dotnet-responsestreaming-bedrock-multiturn/src/TripPlannerMultiturn.csproj create mode 100644 sam-dotnet-responsestreaming-bedrock-multiturn/template.yaml diff --git a/sam-dotnet-responsestreaming-bedrock-multiturn/.gitignore b/sam-dotnet-responsestreaming-bedrock-multiturn/.gitignore new file mode 100644 index 000000000..eb9697107 --- /dev/null +++ b/sam-dotnet-responsestreaming-bedrock-multiturn/.gitignore @@ -0,0 +1,16 @@ +## .NET +bin/ +obj/ +*.user +*.suo +*.userosscache +*.sln.docstates + +## SAM +.aws-sam/ +samconfig.toml + +## IDE +.idea/ +.vs/ +*.swp diff --git a/sam-dotnet-responsestreaming-bedrock-multiturn/README.md b/sam-dotnet-responsestreaming-bedrock-multiturn/README.md new file mode 100644 index 000000000..f73b19584 --- /dev/null +++ b/sam-dotnet-responsestreaming-bedrock-multiturn/README.md @@ -0,0 +1,201 @@ +# AWS Lambda Response Streaming with Multi-Turn Bedrock Agent — AI Trip Planner (.NET) + +This sample demonstrates a **multi-turn AI agent** that can ask the user clarifying questions before generating a trip itinerary. Conversation state is persisted in DynamoDB between requests, and the final itinerary is streamed token-by-token through API Gateway response streaming. + +## Architecture + +``` +┌──────────────┐ ┌──────────────────┐ ┌────────────────────┐ ┌─────────────┐ +│ Client │ │ API Gateway │ │ Lambda Function │ │ Amazon │ +│ │──POST─▶│ REST API │──────▶│ (Agent Loop) │◀─────▶│ Bedrock │ +│ │◀───────│ │◀──────│ │ │ (Claude) │ +└──────────────┘ └──────────────────┘ └────────┬───────────┘ └─────────────┘ + │ + ┌──────▼───────┐ + │ DynamoDB │ + │ (Sessions) │ + │ TTL: 1 hour │ + └──────────────┘ + +Flow: + 1. POST /plan { destination, days, interests } + → Agent asks a question → JSON response with sessionId + question + 2. POST /plan { sessionId, message: "answer" } + → Agent asks another question OR generates itinerary (streamed) + 3. Itinerary streams day-by-day via API Gateway response streaming +``` + +## What It Demonstrates + +- **Multi-turn agent with session persistence** — The agent can pause, ask the user a question, and resume when the user responds. Conversation state (Bedrock message history) is stored in DynamoDB between requests. +- **Tool use for control flow** — Two tools: `ask_user` (suspends the loop, returns a question) and `add_day_plan` (generates itinerary content). The model decides which to call. +- **DynamoDB TTL** — Sessions automatically expire after 1 hour, requiring no cleanup logic. +- **Conditional response format** — Returns JSON when asking questions; switches to streamed text when generating the itinerary. +- **API Gateway response streaming** — The final itinerary streams progressively using `ResponseTransferMode: STREAM`. + +## Project Structure + +``` +├── template.yaml # SAM template (API Gateway + Lambda + DynamoDB with TTL) +├── events/ +│ ├── new-trip.json # Start a new trip planning conversation +│ └── continue-trip.json # Continue with an answer to the agent's question +└── src/ + ├── Program.cs # Entry point — routes new vs continuation requests + ├── Models.cs # Request/response records + ├── TripPlannerAgent.cs # Agent loop with ask_user suspension support + ├── ToolDefinitions.cs # Tool schemas (add_day_plan + ask_user) + ├── SessionStore.cs # DynamoDB session persistence with TTL + ├── ResponseFormatter.cs # Streaming output formatting + └── TripPlannerMultiturn.csproj +``` + +--- + +## Prerequisites + +- [.NET 10 SDK](https://dotnet.microsoft.com/download/dotnet/10.0) +- [AWS SAM CLI](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/install-sam-cli.html) +- AWS account with credentials configured (`aws configure`) +- **Amazon Bedrock model access** — Enable access to Claude Sonnet in the [Bedrock console](https://console.aws.amazon.com/bedrock/home#/modelaccess) + +--- + +## Build & Deploy + +```bash +sam build +sam deploy --guided +``` + +--- + +## Testing + +### 1. Get the API endpoint + +```bash +API_URL=$(aws cloudformation describe-stacks \ + --stack-name sam-dotnet-responsestreaming-bedrock-multiturn \ + --query "Stacks[0].Outputs[?OutputKey=='ApiEndpoint'].OutputValue" \ + --output text) +``` + +### 2. Start a new trip (agent will ask a question) + +```bash +curl --no-buffer -s -X POST "$API_URL" \ + -H "Content-Type: application/json" \ + -d '{ + "destination": "Rome, Italy", + "days": 4, + "interests": "history, pasta, and architecture" + }' | python3 -m json.tool +``` + +Response (JSON — the agent is asking a question): +```json +{ + "SessionId": "a1b2c3d4e5f6...", + "Question": "What's your budget level — backpacker, mid-range, or luxury? And do you have any dietary restrictions I should consider for restaurant recommendations?", + "Status": "awaiting_input" +} +``` + +### 3. Answer the question (agent may ask another or start planning) + +```bash +SESSION_ID="" + +curl --no-buffer -s -X POST "$API_URL" \ + -H "Content-Type: application/json" \ + -d "{ + \"sessionId\": \"$SESSION_ID\", + \"message\": \"Mid-range budget. No dietary restrictions, but I love wine.\" + }" +``` + +If the agent is satisfied, it streams the itinerary: +``` +✈️ Trip Planner — Rome, Italy (4 days) + Interests: history, pasta, and architecture +════════════════════════════════════════════════════════════════ + +📅 Day 1: Ancient Rome & Trastevere + 🌅 Morning: Start at the Colosseum... + ☀️ Afternoon: Walk through the Roman Forum... + 🌙 Evening: Cross to Trastevere for dinner... + 💡 Tips: Book Colosseum tickets online... + +📅 Day 2: Vatican & Wine Tasting + ... +``` + +### 4. Skip questions (go straight to planning) + +```bash +curl --no-buffer -s -X POST "$API_URL" \ + -H "Content-Type: application/json" \ + -d '{ + "destination": "Tokyo, Japan", + "days": 3, + "interests": "food, anime, and technology. Skip questions and just plan." + }' +``` + +--- + +## How the Multi-Turn Pattern Works + +``` +Request 1 (new): + Client → Lambda → Bedrock (calls ask_user tool) + Lambda saves messages to DynamoDB, returns { sessionId, question } + +Request 2 (continue): + Client → Lambda (loads session from DynamoDB) + Lambda appends user answer → Bedrock (calls add_day_plan tools) + Lambda streams itinerary, deletes session +``` + +Key implementation details: + +1. **`ask_user` tool** — When the model calls this, the agent loop returns `AgentAskedQuestion` with the question text and the tool use ID. The Lambda saves the conversation and returns JSON. + +2. **Resuming** — On the next request, the Lambda loads the session, appends the user's answer as a `ToolResult` for the pending `ask_user` call, and resumes the agent loop. + +3. **DynamoDB TTL** — Each session is stored with an `ExpiresAt` attribute set to 1 hour from creation. DynamoDB automatically deletes expired items, so abandoned conversations don't accumulate. + +4. **Streaming only on final generation** — Questions are returned as buffered JSON (fast, small). The itinerary is streamed (large, progressive). The Lambda decides which response mode to use based on the agent's behavior. + +--- + +## DynamoDB Session Schema + +| Attribute | Type | Description | +|-----------|------|-------------| +| `SessionId` | String (PK) | Unique conversation identifier | +| `Messages` | String (JSON) | Serialized Bedrock message history | +| `Destination` | String | Trip destination | +| `Days` | Number | Trip duration | +| `Interests` | String | User interests | +| `PendingToolUseId` | String | ID of the ask_user tool call awaiting a response | +| `ExpiresAt` | Number (TTL) | Unix timestamp — DynamoDB auto-deletes after this time | + +--- + +## Cleanup + +```bash +sam delete +``` + +--- + +## References + +- [Building responsive APIs with API Gateway response streaming](https://aws.amazon.com/blogs/compute/building-responsive-apis-with-amazon-api-gateway-response-streaming/) +- [Bedrock Converse API with tool use](https://docs.aws.amazon.com/bedrock/latest/userguide/tool-use.html) +- [DynamoDB Time to Live (TTL)](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/TTL.html) +- [Lambda Response Streaming (.NET SDK PR)](https://github.com/aws/aws-lambda-dotnet/pull/2288) +- [AWS SAM Developer Guide](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/) diff --git a/sam-dotnet-responsestreaming-bedrock-multiturn/events/continue-trip.json b/sam-dotnet-responsestreaming-bedrock-multiturn/events/continue-trip.json new file mode 100644 index 000000000..9c3fa301e --- /dev/null +++ b/sam-dotnet-responsestreaming-bedrock-multiturn/events/continue-trip.json @@ -0,0 +1,24 @@ +{ + "resource": "/plan", + "path": "/plan", + "httpMethod": "POST", + "headers": { + "Content-Type": "application/json", + "Host": "abc123.execute-api.us-east-1.amazonaws.com" + }, + "queryStringParameters": null, + "pathParameters": null, + "stageVariables": null, + "requestContext": { + "resourceId": "abc123", + "resourcePath": "/plan", + "httpMethod": "POST", + "requestId": "test-request-id", + "accountId": "123456789012", + "stage": "prod", + "identity": { "sourceIp": "127.0.0.1", "userAgent": "curl/8.0" }, + "apiId": "abc123" + }, + "body": "{\"sessionId\": \"abc123def456\", \"message\": \"Medium budget, prefer a relaxed pace with long lunches\"}", + "isBase64Encoded": false +} diff --git a/sam-dotnet-responsestreaming-bedrock-multiturn/events/new-trip.json b/sam-dotnet-responsestreaming-bedrock-multiturn/events/new-trip.json new file mode 100644 index 000000000..e52d7434b --- /dev/null +++ b/sam-dotnet-responsestreaming-bedrock-multiturn/events/new-trip.json @@ -0,0 +1,24 @@ +{ + "resource": "/plan", + "path": "/plan", + "httpMethod": "POST", + "headers": { + "Content-Type": "application/json", + "Host": "abc123.execute-api.us-east-1.amazonaws.com" + }, + "queryStringParameters": null, + "pathParameters": null, + "stageVariables": null, + "requestContext": { + "resourceId": "abc123", + "resourcePath": "/plan", + "httpMethod": "POST", + "requestId": "test-request-id", + "accountId": "123456789012", + "stage": "prod", + "identity": { "sourceIp": "127.0.0.1", "userAgent": "curl/8.0" }, + "apiId": "abc123" + }, + "body": "{\"destination\": \"Rome, Italy\", \"days\": 4, \"interests\": \"history, pasta, and architecture\"}", + "isBase64Encoded": false +} diff --git a/sam-dotnet-responsestreaming-bedrock-multiturn/sam-dotnet-responsestreaming-bedrock-multiturn.json b/sam-dotnet-responsestreaming-bedrock-multiturn/sam-dotnet-responsestreaming-bedrock-multiturn.json new file mode 100644 index 000000000..fcaf7cde1 --- /dev/null +++ b/sam-dotnet-responsestreaming-bedrock-multiturn/sam-dotnet-responsestreaming-bedrock-multiturn.json @@ -0,0 +1,68 @@ +{ + "title": "Lambda Response Streaming with Multi-Turn Bedrock Agent (.NET)", + "description": "A multi-turn AI agent with DynamoDB session persistence that asks clarifying questions before streaming a trip itinerary through API Gateway response streaming.", + "language": ".NET", + "level": "300", + "framework": "SAM", + "introBox": { + "headline": "How it works", + "text": [ + "This pattern deploys a Lambda function that implements a multi-turn AI agent using Amazon Bedrock (Claude Sonnet) with tool use. The agent can pause and ask the user clarifying questions before generating output.", + "Conversation state (Bedrock message history) is persisted in DynamoDB between requests with TTL auto-expiry. When the agent has gathered enough information, it generates a full trip itinerary that streams token-by-token through API Gateway response streaming.", + "This pattern demonstrates session persistence, multi-turn agentic loops, and keepalive streaming to prevent API Gateway idle timeouts during long-running Bedrock calls." + ] + }, + "gitHub": { + "template": { + "repoURL": "https://github.com/aws-samples/serverless-patterns/tree/main/sam-dotnet-responsestreaming-bedrock-multiturn", + "templateURL": "serverless-patterns/sam-dotnet-responsestreaming-bedrock-multiturn", + "projectFolder": "sam-dotnet-responsestreaming-bedrock-multiturn", + "templateFile": "template.yaml" + } + }, + "resources": { + "bullets": [ + { + "text": "Lambda Response Streaming Documentation", + "link": "https://docs.aws.amazon.com/lambda/latest/dg/response-streaming.html" + }, + { + "text": "Amazon Bedrock Converse API", + "link": "https://docs.aws.amazon.com/bedrock/latest/userguide/conversation-inference-call.html" + }, + { + "text": "Amazon Bedrock Tool Use (Function Calling)", + "link": "https://docs.aws.amazon.com/bedrock/latest/userguide/tool-use.html" + }, + { + "text": "Amazon DynamoDB TTL", + "link": "https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/TTL.html" + } + ] + }, + "deploy": { + "text": [ + "sam build", + "sam deploy --guided" + ] + }, + "testing": { + "text": [ + "See the GitHub repo for detailed testing instructions." + ] + }, + "cleanup": { + "text": [ + "Delete the stack: sam delete." + ] + }, + "authors": [ + { + "name": "Doug Perkes", + "image": "", + "bio": "Senior Solutions Architect at AWS, focused on .NET and serverless.", + "linkedin": "dougperkes", + "twitter": "" + } + ] +} diff --git a/sam-dotnet-responsestreaming-bedrock-multiturn/src/Models.cs b/sam-dotnet-responsestreaming-bedrock-multiturn/src/Models.cs new file mode 100644 index 000000000..89ca4286c --- /dev/null +++ b/sam-dotnet-responsestreaming-bedrock-multiturn/src/Models.cs @@ -0,0 +1,37 @@ +namespace TripPlannerMultiturn; + +/// +/// Incoming request body from the client. +/// +public record TripPlannerRequest +{ + /// Session ID for continuing a conversation. Null for a new conversation. + public string? SessionId { get; init; } + + /// The user's message (initial request or answer to a question). + public string? Message { get; init; } + + /// Travel destination (used on first request). + public string? Destination { get; init; } + + /// Number of days (used on first request). + public int? Days { get; init; } + + /// Traveler interests (used on first request). + public string? Interests { get; init; } +} + +/// +/// A single day's plan extracted from the agent's tool call. +/// +public record DayPlan(int DayNumber, string Title, string Morning, string Afternoon, string Evening, string Tips); + +/// +/// The result sent back to the client when the agent asks a question. +/// +public record QuestionResponse(string SessionId, string Question, string Status = "awaiting_input"); + +/// +/// Trip parameters stored in the session. +/// +public record TripParameters(string Destination, int Days, string Interests); diff --git a/sam-dotnet-responsestreaming-bedrock-multiturn/src/Program.cs b/sam-dotnet-responsestreaming-bedrock-multiturn/src/Program.cs new file mode 100644 index 000000000..f76062f35 --- /dev/null +++ b/sam-dotnet-responsestreaming-bedrock-multiturn/src/Program.cs @@ -0,0 +1,240 @@ +using System.Net; +using System.Text.Json; +using Amazon.BedrockRuntime; +using Amazon.DynamoDBv2; +using Amazon.Lambda.APIGatewayEvents; +using Amazon.Lambda.Core; +using Amazon.Lambda.Core.ResponseStreaming; +using Amazon.Lambda.RuntimeSupport; +using Amazon.Lambda.Serialization.SystemTextJson; +using TripPlannerMultiturn; + +#pragma warning disable CA2252 // Opt in to preview features (response streaming) + +var bedrockClient = new AmazonBedrockRuntimeClient(); +var dynamoDbClient = new AmazonDynamoDBClient(); +var tableName = System.Environment.GetEnvironmentVariable("SESSION_TABLE_NAME") + ?? throw new InvalidOperationException("SESSION_TABLE_NAME not set"); + +var sessionStore = new SessionStore(dynamoDbClient, tableName); +var agent = new TripPlannerAgent(bedrockClient); + +var jsonOptions = new JsonSerializerOptions { PropertyNameCaseInsensitive = true }; + +var handler = async (APIGatewayProxyRequest request, ILambdaContext context) => +{ + var body = !string.IsNullOrEmpty(request.Body) + ? JsonSerializer.Deserialize(request.Body, jsonOptions) + : null; + + if (body is null) + { + await WriteJsonResponseAsync(400, new { message = "Request body is required." }); + return; + } + + if (body.SessionId is not null) + await HandleContinuationAsync(body, context); + else + await HandleNewConversationAsync(body, context); + + return; + + // ─── New conversation ─────────────────────────────────────────────────────── + + async Task HandleNewConversationAsync(TripPlannerRequest req, ILambdaContext ctx) + { + var trip = new TripParameters( + req.Destination ?? "Tokyo, Japan", + Math.Clamp(req.Days ?? 3, 1, 14), + req.Interests ?? "culture, food, and nature"); + + var messages = TripPlannerAgent.CreateInitialMessages(trip); + await RunAgentAndRespondAsync(null, trip, messages, ctx); + } + + // ─── Continue an existing conversation ────────────────────────────────────── + + async Task HandleContinuationAsync(TripPlannerRequest req, ILambdaContext ctx) + { + var session = await sessionStore.LoadSessionAsync(req.SessionId!); + if (session is null) + { + await WriteJsonResponseAsync(404, new { message = "Session not found or expired." }); + return; + } + + TripPlannerAgent.AppendUserAnswer(session.Messages, session.PendingToolUseId!, req.Message ?? ""); + await RunAgentAndRespondAsync(req.SessionId, session.TripParams, session.Messages, ctx, expectsItinerary: true); + } + + // ─── Run the agent once and respond based on outcome ──────────────────────── + + async Task RunAgentAndRespondAsync( + string? existingSessionId, TripParameters trip, + List messages, ILambdaContext ctx, + bool expectsItinerary = false) + { + Stream? responseStream = null; + StreamWriter? writer = null; + ResponseFormatter? formatter = null; + var streamStarted = false; + + // If we expect the itinerary, start streaming immediately to prevent + // CloudFront's 30-second idle timeout from killing the connection. + if (expectsItinerary) + { + (responseStream, writer, formatter) = CreateStreamingResponse(trip); + await formatter.WriteHeaderAsync(trip); + await writer.WriteLineAsync("⏳ Planning your trip..."); + await writer.FlushAsync(); + streamStarted = true; + } + + try + { + // Keepalive: send a dot every 5 seconds while waiting for Bedrock, + // to prevent CloudFront's 30-second idle timeout from killing the connection. + using var keepaliveCts = new CancellationTokenSource(); + Task? keepaliveTask = null; + var keepalivePrinted = false; + + if (streamStarted) + { + keepaliveTask = RunKeepaliveAsync(writer!, keepaliveCts.Token); + } + + var result = await agent.RunAsync( + trip, messages, + onDayPlan: async day => + { + // Stop keepalive before writing content + if (keepaliveTask is not null && !keepaliveCts.IsCancellationRequested) + { + await keepaliveCts.CancelAsync(); + await keepaliveTask; + keepalivePrinted = true; + } + + if (!streamStarted) + { + (responseStream, writer, formatter) = CreateStreamingResponse(trip); + await formatter.WriteHeaderAsync(trip); + streamStarted = true; + } + else if (keepalivePrinted) + { + // End the dots line before first day plan + await writer!.WriteLineAsync(); + await writer.WriteLineAsync(); + keepalivePrinted = false; + } + + await formatter!.WriteDayPlanAsync(day); + }, + onSummary: async text => + { + if (formatter is not null) + await formatter.WriteSummaryAsync(text); + }); + + // Ensure keepalive is stopped + if (!keepaliveCts.IsCancellationRequested) + await keepaliveCts.CancelAsync(); + if (keepaliveTask is not null) + await keepaliveTask; + + if (result is AgentAskedQuestion asked) + { + var sessionId = existingSessionId ?? Guid.NewGuid().ToString("N"); + await sessionStore.SaveSessionAsync(sessionId, asked.Messages, trip, asked.ToolUseId); + ctx.Logger.LogInformation($"Session {sessionId}: agent asked question"); + + if (streamStarted) + { + // Already streaming — write the question as text + await writer!.WriteLineAsync(); + await writer.WriteLineAsync($"❓ {asked.Question}"); + await writer.WriteLineAsync(); + await writer.WriteLineAsync($"Reply with: {{\"sessionId\": \"{sessionId}\", \"message\": \"your answer\"}}"); + await writer.FlushAsync(); + } + else + { + await WriteJsonResponseAsync(200, new QuestionResponse(sessionId, asked.Question)); + } + } + else if (streamStarted) + { + if (existingSessionId is not null) + await sessionStore.DeleteSessionAsync(existingSessionId); + await formatter!.WriteFooterAsync(); + } + } + catch (Exception ex) + { + ctx.Logger.LogError($"Error: {ex.Message}"); + if (formatter is not null) + await formatter.WriteErrorAsync(ex.Message); + else + await WriteJsonResponseAsync(500, new { message = $"Error: {ex.Message}" }); + } + finally + { + if (writer is not null) await writer.DisposeAsync(); + if (responseStream is not null) await responseStream.DisposeAsync(); + } + } + + // ─── Helpers ──────────────────────────────────────────────────────────────── + + (Stream responseStream, StreamWriter writer, ResponseFormatter formatter) CreateStreamingResponse(TripParameters trip) + { + var prelude = new HttpResponseStreamPrelude + { + StatusCode = HttpStatusCode.OK, + Headers = + { + { "Content-Type", "text/plain; charset=utf-8" }, + { "Cache-Control", "no-cache" }, + { "X-Content-Type-Options", "nosniff" } + } + }; + var rs = LambdaResponseStreamFactory.CreateHttpStream(prelude); + var w = new StreamWriter(rs) { AutoFlush = false }; + var f = new ResponseFormatter(w); + return (rs, w, f); + } + + async Task WriteJsonResponseAsync(int statusCode, object responseBody) + { + var prelude = new HttpResponseStreamPrelude + { + StatusCode = (HttpStatusCode)statusCode, + Headers = { { "Content-Type", "application/json" } } + }; + + using var rs = LambdaResponseStreamFactory.CreateHttpStream(prelude); + using var w = new StreamWriter(rs); + await w.WriteAsync(JsonSerializer.Serialize(responseBody, jsonOptions)); + await w.FlushAsync(); + } + + async Task RunKeepaliveAsync(StreamWriter w, CancellationToken ct) + { + try + { + while (!ct.IsCancellationRequested) + { + await Task.Delay(TimeSpan.FromSeconds(5), ct); + await w.WriteAsync("."); + await w.FlushAsync(); + } + } + catch (OperationCanceledException) { } + } +}; + +await LambdaBootstrapBuilder.Create(handler, new DefaultLambdaJsonSerializer()) + .Build() + .RunAsync(); diff --git a/sam-dotnet-responsestreaming-bedrock-multiturn/src/ResponseFormatter.cs b/sam-dotnet-responsestreaming-bedrock-multiturn/src/ResponseFormatter.cs new file mode 100644 index 000000000..e05fa03d9 --- /dev/null +++ b/sam-dotnet-responsestreaming-bedrock-multiturn/src/ResponseFormatter.cs @@ -0,0 +1,55 @@ +namespace TripPlannerMultiturn; + +/// +/// Formats and streams trip planner output to a StreamWriter. +/// +public class ResponseFormatter +{ + private readonly StreamWriter _writer; + + public ResponseFormatter(StreamWriter writer) + { + _writer = writer ?? throw new ArgumentNullException(nameof(writer)); + } + + public async Task WriteHeaderAsync(TripParameters trip) + { + await _writer.WriteLineAsync($"✈️ Trip Planner — {trip.Destination} ({trip.Days} days)"); + await _writer.WriteLineAsync($" Interests: {trip.Interests}"); + await _writer.WriteLineAsync(new string('═', 60)); + await _writer.WriteLineAsync(); + await _writer.FlushAsync(); + } + + public async Task WriteDayPlanAsync(DayPlan day) + { + await _writer.WriteLineAsync($"📅 Day {day.DayNumber}: {day.Title}"); + await _writer.WriteLineAsync($" 🌅 Morning: {day.Morning}"); + await _writer.WriteLineAsync($" ☀️ Afternoon: {day.Afternoon}"); + await _writer.WriteLineAsync($" 🌙 Evening: {day.Evening}"); + await _writer.WriteLineAsync($" 💡 Tips: {day.Tips}"); + await _writer.WriteLineAsync(); + await _writer.FlushAsync(); + } + + public async Task WriteSummaryAsync(string summary) + { + await _writer.WriteLineAsync(new string('═', 60)); + await _writer.WriteLineAsync(); + await _writer.WriteLineAsync(summary); + await _writer.FlushAsync(); + } + + public async Task WriteFooterAsync() + { + await _writer.WriteLineAsync(); + await _writer.WriteLineAsync("✅ Happy travels!"); + await _writer.FlushAsync(); + } + + public async Task WriteErrorAsync(string message) + { + await _writer.WriteLineAsync($"\n\n[Error: {message}]"); + await _writer.FlushAsync(); + } +} diff --git a/sam-dotnet-responsestreaming-bedrock-multiturn/src/SessionStore.cs b/sam-dotnet-responsestreaming-bedrock-multiturn/src/SessionStore.cs new file mode 100644 index 000000000..0a3c6645b --- /dev/null +++ b/sam-dotnet-responsestreaming-bedrock-multiturn/src/SessionStore.cs @@ -0,0 +1,228 @@ +using System.Text.Json; +using Amazon.DynamoDBv2; +using Amazon.DynamoDBv2.Model; +using Amazon.BedrockRuntime; +using Amazon.BedrockRuntime.Model; +using Amazon.Runtime.Documents; +using Document = Amazon.Runtime.Documents.Document; + +namespace TripPlannerMultiturn; + +/// +/// Persists conversation state (Bedrock messages + trip parameters) in DynamoDB. +/// Sessions expire automatically via DynamoDB TTL after 1 hour. +/// +public class SessionStore +{ + private static readonly TimeSpan SessionTtl = TimeSpan.FromHours(1); + + private readonly IAmazonDynamoDB _dynamoDb; + private readonly string _tableName; + + public SessionStore(IAmazonDynamoDB dynamoDb, string tableName) + { + _dynamoDb = dynamoDb; + _tableName = tableName; + } + + public async Task SaveSessionAsync( + string sessionId, + List messages, + TripParameters tripParams, + string? pendingToolUseId, + CancellationToken ct = default) + { + var expiresAt = DateTimeOffset.UtcNow.Add(SessionTtl).ToUnixTimeSeconds(); + var serializedMessages = SerializeMessages(messages); + + // DynamoDB items have a 400KB limit. Guard against oversized sessions. + if (serializedMessages.Length > 350_000) + throw new InvalidOperationException("Conversation history too large to store. Please start a new session."); + + var item = new Dictionary + { + ["SessionId"] = new() { S = sessionId }, + ["Messages"] = new() { S = serializedMessages }, + ["Destination"] = new() { S = tripParams.Destination }, + ["Days"] = new() { N = tripParams.Days.ToString() }, + ["Interests"] = new() { S = tripParams.Interests }, + ["ExpiresAt"] = new() { N = expiresAt.ToString() } + }; + + if (pendingToolUseId is not null) + item["PendingToolUseId"] = new() { S = pendingToolUseId }; + + await _dynamoDb.PutItemAsync(new PutItemRequest + { + TableName = _tableName, + Item = item + }, ct); + } + + public async Task LoadSessionAsync(string sessionId, CancellationToken ct = default) + { + var response = await _dynamoDb.GetItemAsync(new GetItemRequest + { + TableName = _tableName, + Key = new Dictionary + { + ["SessionId"] = new() { S = sessionId } + } + }, ct); + + if (!response.IsItemSet) + return null; + + var item = response.Item; + var messages = DeserializeMessages(item["Messages"].S); + var tripParams = new TripParameters( + item["Destination"].S, + int.TryParse(item["Days"].N, out var days) ? days : 3, + item["Interests"].S); + var pendingToolUseId = item.TryGetValue("PendingToolUseId", out var toolId) ? toolId.S : null; + + return new SessionData(messages, tripParams, pendingToolUseId); + } + + public async Task DeleteSessionAsync(string sessionId, CancellationToken ct = default) + { + await _dynamoDb.DeleteItemAsync(new DeleteItemRequest + { + TableName = _tableName, + Key = new Dictionary + { + ["SessionId"] = new() { S = sessionId } + } + }, ct); + } + + private static string SerializeMessages(List messages) + { + var simplified = messages.Select(m => new SerializedMessage + { + Role = m.Role.Value, + Content = m.Content.Select(c => + { + if (c.Text is not null) + return new SerializedContent { Type = "text", Text = c.Text }; + if (c.ToolUse is not null) + return new SerializedContent + { + Type = "tool_use", + ToolUseId = c.ToolUse.ToolUseId, + ToolName = c.ToolUse.Name, + ToolInput = SerializeDocument(c.ToolUse.Input) + }; + if (c.ToolResult is not null) + return new SerializedContent + { + Type = "tool_result", + ToolUseId = c.ToolResult.ToolUseId, + Text = c.ToolResult.Content.FirstOrDefault()?.Text + }; + return new SerializedContent { Type = "unknown" }; + }).ToList() + }).ToList(); + + return JsonSerializer.Serialize(simplified); + } + + private static List DeserializeMessages(string json) + { + var simplified = JsonSerializer.Deserialize>(json) ?? []; + return simplified.Select(m => + { + var message = new Message { Role = new ConversationRole(m.Role) }; + message.Content = m.Content.Select(c => c.Type switch + { + "text" => new ContentBlock { Text = c.Text }, + "tool_use" => new ContentBlock + { + ToolUse = new ToolUseBlock + { + ToolUseId = c.ToolUseId!, + Name = c.ToolName!, + Input = DeserializeDocument(c.ToolInput!) + } + }, + "tool_result" => new ContentBlock + { + ToolResult = new ToolResultBlock + { + ToolUseId = c.ToolUseId!, + Content = [new ToolResultContentBlock { Text = c.Text }] + } + }, + _ => new ContentBlock { Text = "[unknown content]" } + }).ToList(); + return message; + }).ToList(); + } + + private record SerializedMessage + { + public string Role { get; init; } = ""; + public List Content { get; init; } = []; + } + + private record SerializedContent + { + public string Type { get; init; } = ""; + public string? Text { get; init; } + public string? ToolUseId { get; init; } + public string? ToolName { get; init; } + public string? ToolInput { get; init; } + } + + /// + /// Serializes a Document to a JSON string by converting it to a dictionary structure. + /// + private static string SerializeDocument(Document doc) + { + var obj = DocumentToObject(doc); + return JsonSerializer.Serialize(obj); + } + + /// + /// Deserializes a JSON string back into a Document. + /// + private static Document DeserializeDocument(string json) + { + var element = JsonSerializer.Deserialize(json); + return JsonElementToDocument(element); + } + + private static Document JsonElementToDocument(JsonElement element) + { + return element.ValueKind switch + { + JsonValueKind.String => new Document(element.GetString()!), + JsonValueKind.Number when element.TryGetInt32(out var i) => new Document(i), + JsonValueKind.Number when element.TryGetInt64(out var l) => new Document(l), + JsonValueKind.Number => new Document(element.GetDouble()), + JsonValueKind.True => new Document(true), + JsonValueKind.False => new Document(false), + JsonValueKind.Null => new Document(), + JsonValueKind.Array => new Document(element.EnumerateArray().Select(JsonElementToDocument).ToList()), + JsonValueKind.Object => new Document(element.EnumerateObject() + .ToDictionary(p => p.Name, p => JsonElementToDocument(p.Value))), + _ => new Document() + }; + } + + private static object? DocumentToObject(Document doc) + { + if (doc.IsString()) return doc.AsString(); + if (doc.IsInt()) return doc.AsInt(); + if (doc.IsLong()) return doc.AsLong(); + if (doc.IsDouble()) return doc.AsDouble(); + if (doc.IsBool()) return doc.AsBool(); + if (doc.IsNull()) return null; + if (doc.IsList()) return doc.AsList().Select(DocumentToObject).ToList(); + if (doc.IsDictionary()) + return doc.AsDictionary().ToDictionary(kv => kv.Key, kv => DocumentToObject(kv.Value)); + return null; + } +} + +public record SessionData(List Messages, TripParameters TripParams, string? PendingToolUseId); diff --git a/sam-dotnet-responsestreaming-bedrock-multiturn/src/ToolDefinitions.cs b/sam-dotnet-responsestreaming-bedrock-multiturn/src/ToolDefinitions.cs new file mode 100644 index 000000000..1b62070a0 --- /dev/null +++ b/sam-dotnet-responsestreaming-bedrock-multiturn/src/ToolDefinitions.cs @@ -0,0 +1,87 @@ +using Amazon.BedrockRuntime.Model; +using Document = Amazon.Runtime.Documents.Document; + +namespace TripPlannerMultiturn; + +/// +/// Defines the tools available to the trip planner agent. +/// +public static class ToolDefinitions +{ + public const string AddDayPlanToolName = "add_day_plan"; + public const string AskUserToolName = "ask_user"; + + public static Tool AddDayPlanTool { get; } = new() + { + ToolSpec = new ToolSpecification + { + Name = AddDayPlanToolName, + Description = "Add a single day's plan to the trip itinerary. Call this once for each day of the trip.", + InputSchema = new ToolInputSchema + { + Json = Document.FromObject(new + { + type = "object", + properties = new + { + day_number = new { type = "integer", description = "The day number (1-indexed)" }, + title = new { type = "string", description = "A short title for the day" }, + morning = new { type = "string", description = "Morning activities (2-3 sentences)" }, + afternoon = new { type = "string", description = "Afternoon activities (2-3 sentences)" }, + evening = new { type = "string", description = "Evening activities and dinner (2-3 sentences)" }, + tips = new { type = "string", description = "Practical tips for this day" } + }, + required = new[] { "day_number", "title", "morning", "afternoon", "evening", "tips" } + }) + } + } + }; + + public static Tool AskUserTool { get; } = new() + { + ToolSpec = new ToolSpecification + { + Name = AskUserToolName, + Description = """ + Ask the user a clarifying question before proceeding with planning. + Use this when you need more information to create a better itinerary, + such as budget level, mobility constraints, dietary restrictions, + preferred pace, or specific must-see places. + You may ask at most 2 questions total before proceeding with planning. + """, + InputSchema = new ToolInputSchema + { + Json = Document.FromObject(new + { + type = "object", + properties = new + { + question = new { type = "string", description = "The question to ask the user" } + }, + required = new[] { "question" } + }) + } + } + }; + + public static readonly List AllTools = [AddDayPlanTool, AskUserTool]; + + public static DayPlan ParseDayPlan(ToolUseBlock toolUse) + { + var input = toolUse.Input.AsDictionary(); + return new DayPlan( + DayNumber: input["day_number"].AsInt(), + Title: input["title"].AsString(), + Morning: input["morning"].AsString(), + Afternoon: input["afternoon"].AsString(), + Evening: input["evening"].AsString(), + Tips: input["tips"].AsString() + ); + } + + public static string ParseQuestion(ToolUseBlock toolUse) + { + var input = toolUse.Input.AsDictionary(); + return input["question"].AsString(); + } +} diff --git a/sam-dotnet-responsestreaming-bedrock-multiturn/src/TripPlannerAgent.cs b/sam-dotnet-responsestreaming-bedrock-multiturn/src/TripPlannerAgent.cs new file mode 100644 index 000000000..e9c138478 --- /dev/null +++ b/sam-dotnet-responsestreaming-bedrock-multiturn/src/TripPlannerAgent.cs @@ -0,0 +1,216 @@ +using Amazon.BedrockRuntime; +using Amazon.BedrockRuntime.Model; + +namespace TripPlannerMultiturn; + +/// +/// Result of a single agent turn. Either the agent asks a question (and suspends), +/// or it completes the full itinerary. +/// +public abstract record AgentResult; +public record AgentAskedQuestion(string Question, string ToolUseId, List Messages) : AgentResult; +public record AgentCompleted : AgentResult; + +/// +/// Orchestrates the multi-turn Bedrock conversation. Supports suspending when +/// the agent calls ask_user and resuming when the user responds. +/// +public class TripPlannerAgent +{ + private const string ModelId = "us.anthropic.claude-sonnet-5"; + private const int MaxTokens = 4096; + private const int MaxTurns = 20; + + private readonly IAmazonBedrockRuntime _bedrockClient; + + public TripPlannerAgent(IAmazonBedrockRuntime bedrockClient) + { + _bedrockClient = bedrockClient; + } + + /// + /// Runs the agent loop from the given messages. Returns when the agent either + /// asks a question (AgentAskedQuestion) or finishes the itinerary (AgentCompleted). + /// + public async Task RunAsync( + TripParameters trip, + List messages, + Func onDayPlan, + Func onSummary) + { + var systemPrompt = BuildSystemPrompt(trip); + + for (var turn = 0; turn < MaxTurns; turn++) + { + var response = await CallModelAsync(systemPrompt, messages); + var assistantContent = response.Output.Message.Content; + + messages.Add(new Message + { + Role = ConversationRole.Assistant, + Content = assistantContent + }); + + var toolUseBlocks = assistantContent + .Where(c => c.ToolUse is not null) + .Select(c => c.ToolUse) + .ToList(); + + if (toolUseBlocks.Count > 0) + { + // Check if the agent wants to ask the user a question + var askUserTool = toolUseBlocks.FirstOrDefault(t => t.Name == ToolDefinitions.AskUserToolName); + if (askUserTool is not null) + { + var question = ToolDefinitions.ParseQuestion(askUserTool); + return new AgentAskedQuestion(question, askUserTool.ToolUseId, messages); + } + + // Process day plan tool calls (and return errors for unknown tools) + var toolResults = await ProcessToolCallsAsync(toolUseBlocks, onDayPlan); + messages.Add(new Message + { + Role = ConversationRole.User, + Content = toolResults + }); + } + else + { + // No tool calls — emit final summary + await EmitSummaryAsync(assistantContent, onSummary); + return new AgentCompleted(); + } + } + + return new AgentCompleted(); + } + + /// + /// Creates the initial messages list for a new conversation. + /// + public static List CreateInitialMessages(TripParameters trip) + { + return + [ + new Message + { + Role = ConversationRole.User, + Content = [new ContentBlock { Text = BuildUserMessage(trip) }] + } + ]; + } + + /// + /// Appends the user's answer to a pending ask_user tool call. + /// + public static void AppendUserAnswer(List messages, string toolUseId, string answer) + { + messages.Add(new Message + { + Role = ConversationRole.User, + Content = + [ + new ContentBlock + { + ToolResult = new ToolResultBlock + { + ToolUseId = toolUseId, + Content = [new ToolResultContentBlock { Text = answer }] + } + } + ] + }); + } + + private async Task CallModelAsync(string systemPrompt, List messages) + { + var request = new ConverseRequest + { + ModelId = ModelId, + System = [new SystemContentBlock { Text = systemPrompt }], + Messages = messages, + ToolConfig = new ToolConfiguration + { + Tools = ToolDefinitions.AllTools + }, + InferenceConfig = new InferenceConfiguration + { + MaxTokens = MaxTokens + } + }; + + return await _bedrockClient.ConverseAsync(request); + } + + private static async Task> ProcessToolCallsAsync( + List toolUseBlocks, Func onDayPlan) + { + var toolResults = new List(); + + foreach (var toolUse in toolUseBlocks) + { + if (toolUse.Name == ToolDefinitions.AddDayPlanToolName) + { + var dayPlan = ToolDefinitions.ParseDayPlan(toolUse); + await onDayPlan(dayPlan); + + toolResults.Add(new ContentBlock + { + ToolResult = new ToolResultBlock + { + ToolUseId = toolUse.ToolUseId, + Content = [new ToolResultContentBlock + { + Text = $"Day {dayPlan.DayNumber} added to itinerary successfully." + }] + } + }); + } + else + { + // Unknown tool — return an error result so the model can recover + toolResults.Add(new ContentBlock + { + ToolResult = new ToolResultBlock + { + ToolUseId = toolUse.ToolUseId, + Status = ToolResultStatus.Error, + Content = [new ToolResultContentBlock + { + Text = $"Unknown tool '{toolUse.Name}'. Available tools: add_day_plan, ask_user." + }] + } + }); + } + } + + return toolResults; + } + + private static async Task EmitSummaryAsync(List content, Func onSummary) + { + var textBlocks = content.Where(c => c.Text is not null).Select(c => c.Text!).ToList(); + foreach (var text in textBlocks) + { + await onSummary(text); + } + } + + private static string BuildSystemPrompt(TripParameters trip) => $""" + You are an expert travel planner. Create a detailed {trip.Days}-day itinerary for {trip.Destination}. + The traveler is interested in: {trip.Interests}. + + IMPORTANT RULES: + 1. Before planning, you SHOULD ask 1-2 clarifying questions using the ask_user tool to + understand the traveler better (e.g., budget, pace preference, dietary needs, mobility). + Ask only ONE question at a time. + 2. Once you have enough information, use the add_day_plan tool exactly {trip.Days} times. + 3. Plan the days in order. Make each day distinct with specific place names. + 4. After all days, provide a brief closing summary. + + If the user says to skip questions or just plan, proceed directly with add_day_plan. + """; + + private static string BuildUserMessage(TripParameters trip) => + $"Plan my {trip.Days}-day trip to {trip.Destination}. I'm interested in {trip.Interests}."; +} diff --git a/sam-dotnet-responsestreaming-bedrock-multiturn/src/TripPlannerMultiturn.csproj b/sam-dotnet-responsestreaming-bedrock-multiturn/src/TripPlannerMultiturn.csproj new file mode 100644 index 000000000..300d8c075 --- /dev/null +++ b/sam-dotnet-responsestreaming-bedrock-multiturn/src/TripPlannerMultiturn.csproj @@ -0,0 +1,20 @@ + + + net10.0 + enable + enable + Exe + TripPlannerMultiturn + Lambda + true + true + + + + + + + + + + diff --git a/sam-dotnet-responsestreaming-bedrock-multiturn/template.yaml b/sam-dotnet-responsestreaming-bedrock-multiturn/template.yaml new file mode 100644 index 000000000..6a8271861 --- /dev/null +++ b/sam-dotnet-responsestreaming-bedrock-multiturn/template.yaml @@ -0,0 +1,118 @@ +AWSTemplateFormatVersion: '2010-09-09' +Transform: AWS::Serverless-2016-10-31 +Description: > + Multi-turn AI Trip Planner agent using Amazon Bedrock (Claude Sonnet) with + Lambda Response Streaming through API Gateway. The agent can ask the user + clarifying questions before generating the itinerary, persisting conversation + state in DynamoDB between requests. + +Globals: + Function: + Timeout: 900 + MemorySize: 512 + Runtime: dotnet10 + Architectures: + - x86_64 + +Resources: + # ───────────────────────────────────────────────────────────────────── + # DynamoDB Table — conversation session store with TTL + # ───────────────────────────────────────────────────────────────────── + SessionTable: + Type: AWS::DynamoDB::Table + Properties: + BillingMode: PAY_PER_REQUEST + AttributeDefinitions: + - AttributeName: SessionId + AttributeType: S + KeySchema: + - AttributeName: SessionId + KeyType: HASH + TimeToLiveSpecification: + AttributeName: ExpiresAt + Enabled: true + DeletionPolicy: Delete + + # ───────────────────────────────────────────────────────────────────── + # REST API with response streaming + # ───────────────────────────────────────────────────────────────────── + TripApi: + Type: AWS::ApiGateway::RestApi + Properties: + Name: trip-planner-multiturn-api + Description: Multi-turn AI trip planner with response streaming + + PlanResource: + Type: AWS::ApiGateway::Resource + Properties: + RestApiId: !Ref TripApi + ParentId: !GetAtt TripApi.RootResourceId + PathPart: plan + + PlanMethod: + Type: AWS::ApiGateway::Method + Properties: + RestApiId: !Ref TripApi + ResourceId: !Ref PlanResource + HttpMethod: POST + AuthorizationType: NONE + Integration: + Type: AWS_PROXY + IntegrationHttpMethod: POST + ResponseTransferMode: STREAM + TimeoutInMillis: 900000 + Uri: !Sub arn:aws:apigateway:${AWS::Region}:lambda:path/2021-11-15/functions/${TripPlannerFunction.Arn}/response-streaming-invocations + + ApiDeployment: + Type: AWS::ApiGateway::Deployment + DependsOn: PlanMethod + Properties: + RestApiId: !Ref TripApi + + ApiStage: + Type: AWS::ApiGateway::Stage + Properties: + RestApiId: !Ref TripApi + DeploymentId: !Ref ApiDeployment + StageName: prod + + # ───────────────────────────────────────────────────────────────────── + # Lambda Function — multi-turn trip planner agent + # ───────────────────────────────────────────────────────────────────── + TripPlannerFunction: + Type: AWS::Serverless::Function + Properties: + Handler: TripPlannerMultiturn + CodeUri: ./src/ + Description: Multi-turn AI trip planner with session persistence and response streaming + Environment: + Variables: + SESSION_TABLE_NAME: !Ref SessionTable + Policies: + - DynamoDBCrudPolicy: + TableName: !Ref SessionTable + - Statement: + - Effect: Allow + Action: + - bedrock:InvokeModelWithResponseStream + - bedrock:InvokeModel + Resource: "*" + + TripPlannerFunctionPermission: + Type: AWS::Lambda::Permission + Properties: + FunctionName: !GetAtt TripPlannerFunction.Arn + Action: lambda:InvokeFunction + Principal: apigateway.amazonaws.com + SourceArn: !Sub arn:aws:execute-api:${AWS::Region}:${AWS::AccountId}:${TripApi}/* + +Outputs: + ApiEndpoint: + Description: API Gateway endpoint URL for the trip planner + Value: !Sub "https://${TripApi}.execute-api.${AWS::Region}.amazonaws.com/prod/plan" + SessionTableName: + Description: DynamoDB table for conversation sessions + Value: !Ref SessionTable + TripPlannerFunctionArn: + Description: Trip planner Lambda function ARN + Value: !GetAtt TripPlannerFunction.Arn