Category: Uncategorized

This blog post is based on work co-developed with Flo Health.
Healthcare science is rapidly advancing. Maintaining accurate and up-to-date medical content directly impacts people’s lives, health decisions, and well-being. When someone searches for health information, they are often at their most vulnerable, making accuracy not just important, but potentially life-saving.
Flo Health creates thousands of medical articles every year, providing millions of users worldwide with medically credible information on women’s health. Verifying the accuracy and relevance of this vast content library is a significant challenge. Medical knowledge evolves continuously, and manual review of each article is not only time-consuming but also prone to human error. This is why the team at Flo Health, the company behind the leading women’s health app Flo, is using generative AI to facilitate medical content accuracy at scale. Through a partnership with AWS Generative AI Innovation Center, Flo Health is developing an innovative approach, further called, “Medical Automated Content Review and Revision Optimization Solution” (MACROS) to verify and maintain the accuracy of its extensive health information library. This AI-powered solution is capable of:

Efficiently processing large volumes of medical content based on credible scientific sources.
Identifying potential inaccuracies or outdated information based on credible scientific resources.
Proposing updates based on the latest medical research and guidelines, as well as incorporating user feedback.

The system powered by Amazon Bedrock enables Flo Health to conduct medical content reviews and revision assessments at scale, ensuring up-to-date accuracy and supporting more informed healthcare decision-making. This system performs detailed content analysis, providing comprehensive insights on medical standards and guidelines adherence for Flo’s medical experts to review. It is also designed for seamless integration with Flo’s existing tech infrastructure, facilitating automatic updates where appropriate.
This two-part series explores Flo Health’s journey with generative AI for medical content verification. Part 1 examines our proof of concept (PoC), including the initial solution, capabilities, and early results. Part 2 covers focusing on scaling challenges and real-world implementation. Each article stands alone while collectively showing how AI transforms medical content management at scale.
Proof of Concept goals and success criteria
Before diving into the technical solution, we established clear objectives for our PoC medical content review system:
Key Objectives:

Validate the feasibility of using generative AI for medical content verification
Determine accuracy levels compared to manual review
Assess processing time and cost improvements

Success Metrics:

Accuracy: Content piece recall of 90%
Efficiency: Reduce detection time from hours to minutes per guideline
Cost Reduction: Reduce expert review workload
Quality: Maintain Flo’s editorial standards and medical accuracy
Speed: 10x faster than manual review process

To verify the solution meets Flo Health’s high standards for medical content, Flo Health’s medical experts and content teams were working closely with AWS technical specialists through regular review sessions, providing critical feedback and medical expertise to continuously enhance the AI model’s performance and accuracy. The result is MACROS, our custom-built solution for AI-assisted medical content verification.
Solution overview
In this section, we outline how the MACROS solution uses Amazon Bedrock and other AWS services to automate medical content review and revisions.

Figure 1. Medical Automated Content Review and Revision Optimization Solution Overview
As shown in Figure 1, the developed solution supports two major processes:

Content Review and Revision: Enables the medical standards and style adherence of existing medical articles at scale given the pre-specified custom rules and guidelines and proposes a revision that conforms to the new medical standards as well as Flo’s style and tone guidelines.
Rule Optimization: MACROS accelerates the process of extracting the new (medical) guidelines from the (medical) research, pre-processing them into the format needed for content review, as well as optimizing their quality.

Both steps can be conducted through the user interface (UI) as well as the direct API call. The UI support enables medical experts to directly see the content review statistics, interact with changes, and do manual adjustments. The API call support is intended for the integration into pipeline for periodic assessment.
Architecture
Figure 2 depicts the architecture of MACROS. It consists of two major parts: backend and frontend.

Figure 2. MACROS architecture

In the following, the flow of major app components is presented:
1. Users begin by gathering and preparing content that must meet medical standards and rules.
2. In the second step, the data is provided as PDF, TXT files or text through the Streamlit UI that is hosted in Amazon Elastic Container Service (ECS). The authentication for file upload happens through Amazon API Gateway
3. Alternatively, custom Flo Health JSON files can be directly uploaded to the Amazon Simple Storage Service (S3) bucket of the solution stack.
4. The ECS hosted frontend has AWS IAM permissions to orchestrate tasks using AWS Step Functions.
5. Further, the ECS container has access to the S3 for listing, downloading and uploading files either via pre-signed URL or boto3.
6. Optionally, if the input file is uploaded via the UI, the solution invokes AWS Step Functions service that starts the pre-processing functionality within hosted by an AWS Lambda function. This Lambda has access to Amazon Textract for extracting text from PDF files. The files are stored in S3 and also returned to the UI.
7-9. Hosted on AWS Lambda, Rule Optimizer, Content Review and Revision functions are orchestrated via AWS Step Function. They have access to Amazon Bedrock for generative AI capabilities to perform rule extraction from unstructured data, content review and revision, respectively. Furthermore, they have access to S3 via boto3 SDK to store the results.
10. The Compute Stats AWS Lambda function has access to S3 and can read and combine the results of individual revision and review runs.
11. The solution leverages Amazon CloudWatch for system monitoring and log management. For production deployments dealing with critical medical content, the monitoring capabilities could be extended with custom metrics and alarms to provide more granular insights into system performance and content processing patterns.
Future enhancements
While our current architecture utilizes AWS Step Functions for workflow orchestration, we’re exploring the potential of Amazon Bedrock Flows for future iterations. Bedrock Flows offers promising capabilities for streamlining AI-driven workflows, potentially simplifying our architecture and enhancing integration with other Bedrock services. This alternative could provide more seamless management of our AI processes, especially as we scale and evolve our solution.
Content review and revision
At the core of MACROS lies its Content Review and Revision functionality with Amazon Bedrock foundation models. The Content Review and Revision block consists of five major components: 1) The optional Filtering stage 2) Chunking 3) Review 4) Revision and 5) Post-processing, depicted in Figure 3.

Figure 3. Content review and revision pipeline

Here’s how MACROS processes the uploaded medical content:

Filtering (Optional): The journey begins with an optional filtering step. This smart feature checks whether the set of rules is relevant for the article, potentially saving time and resources on unnecessary processing.
Chunking: The source text is then split into paragraphs. This crucial step facilitates good quality assessment and helps prevent unintended revisions to unrelated text. Chunking can be conducted using heuristics, such as punctuation or regular expression-based splits, as well as using large language models (LLM) to identify semantically complete chunks of text.
Review: Each paragraph or section undergoes a thorough review against the relevant rules and guidelines.
Revision: Only the paragraphs flagged as non-adherent move forward to the revision stage, streamlining the process and maintaining the integrity of adherent content. The AI suggests updates to bring non-adherent paragraphs in line with the latest guidelines and Flo’s style requirements.
Post-processing: Finally, the revised paragraphs are seamlessly integrated back into the original text, resulting in an updated, adherent document.

The Filtering step can be conducted using an additional LLM via Amazon Bedrock call that assesses each section separately with the following prompt structure:

Figure 4. Simplified LLM-based filtering step

Further, non-LLM approaches can be feasible to support the Filtering step:

Encoding the rules and the articles into dense embedding vectors and calculating similarity between them. By setting the similarity threshold we can identify which rule set is considered to be relevant for the input document.
Similarly, the direct keyword-level overlap between the document and the rule can be identified using BLEU or ROUGE metrics.

Content review, as already mentioned, is conducted on a text section basis against group of rules and leads to response in XML format, such as:

<xml>
<section_text> Section text without any changes </section_text>
<adherence> 0 <adherence>
<rule_name> Text of the non-adherent rule </rule_name>
<reason> Reason why the section is non-adherent to the rule </reason>
<rule_name> Text of the non-adherent rule </rule_name>
<reason> Reason why the section is non-adherent to the rule </reason>
<section_text> Section text without any changes </section_text>
<adherence> 1 <adherence>
<section_text> Section text without any changes </section_text>
<adherence> 1 <adherence>
</xml>

Here, 1 indicates adherence and 0 – non-adherence of the text to the specified rules. Using XML format helps to achieve reliable parsing of the output.
This Review step iterates over the sections in the text to make sure that the LLM pays attention to each section separately, which led to more robust results in our experimentation. To facilitate higher non-adherent section detection accuracy, the user can also use the Multi-call mode, where instead of one Amazon Bedrock call assessing adherence of the article against all rules, we have one independent call per rule.
The Revision step receives the output of the Review (non-adherent sections and the reasons for non-adherence), as well as the instruction to create the revision in a similar tone. It then suggests revisions of the non-adherent sentences in a style similar to the original text. Finally, the Post-processing step combines the original text with new revisions, making sure that no other sections are changed.
Different steps of the flow require different levels of LLM model complexity. While simpler tasks like chunking can be done efficiently with a relatively small model like Claude Haiku models family, more complex reasoning tasks like content review and revision require larger models like Claude Sonnet or Opus models family to facilitate accurate analysis and high-quality content generation. This tiered approach to model selection optimizes both performance and cost-efficiency of the solution.
Operating modes
The Content Review and Revision feature operates in two UI modes: Detailed Document Processing and Multi Document Processing, each catering to different scales of content management. The Detailed Document Processing mode offers a granular approach to content assessment and is depicted in Figure 5. Users can upload documents in various formats (PDF, TXT, JSON or paste text directly) and specify the guidelines against which the content should be evaluated.

Figure 5. Detailed Document Processing example
Users can choose from predefined rule sets, here, Vitamin D, Breast Health, and Premenstrual Syndrome and Dysphoric Disorder (PMS and PMDD), or input custom guidelines. These custom guidelines can include rules such as “The title of the article must be medically accurate” as well as adherent and non-adherent to the rule examples of content.
The rulesets make sure that the assessment aligns with specific medical standards and Flo’s unique style guide. The interface allows for on-the-fly adjustments, making it ideal for thorough, individual document reviews. For larger-scale operations, the Multi Document Processing mode should be used. This mode is designed to handle numerous custom JSON files simultaneously, mimicking how Flo would integrate MACROS into their content management system.
Extracting rules and guidelines from unstructured data
Actionable and well-prepared guidelines are not always immediately available. Sometimes they are given in unstructured files or need to be found. Using the Rule Optimizer feature, we can extract and refine actionable guidelines from multiple complex documents.
Rule Optimizer processes raw PDF documents to extract text, which is then chunked into meaningful sections based on document headers. This segmented content is processed through Amazon Bedrock using specialized system prompts, with two distinct modes: Style/tonality and Medical mode.
Style/tonality mode focuses on extracting the guidelines on how the text should be written, its style, what formats and words can or cannot be used.
Rule Optimizer assigns a priority for each rule: high, medium, and low. The priority level indicates the rule’s importance, guiding the order of content review and focusing attention on critical areas first. Rule Optimizer includes a manual editing interface where users can refine rule text, adjust classifications, and manage priorities. Therefore, if users need to update a given rule, the changes are stored for future use in Amazon S3.
The Medical mode is designed to process medical documents and is adapted to a more scientific language. It allows grouping of extracted rules into three classes:

Medical condition guidelines
Treatment specific guidelines
Changes to advice and trends in health

Figure 6. Simplified medical rule optimization prompt

Figure 6 provides an example of a medical rule optimization prompt, consisting of three main components: role setting – medical AI expert, description of what makes a good rule, and finally the expected output. We identify the sufficiently good quality for a rule if it is:

Clear, unambiguous, and actionable
Relevant, consistent, and concise (max two sentences)
Written in active voice
Avoids unnecessary jargon

Implementation considerations and challenges
During our PoC development, we identified several crucial considerations that would benefit others implementing similar solutions:

Data preparation: This emerged as a fundamental challenge. We learned the importance of standardizing input formats for both medical content and guidelines while maintaining consistent document structures. Creating diverse test sets across different medical topics proved essential for comprehensive validation.
Cost management: Monitoring and optimizing cost quickly became a key priority. We implemented token usage tracking and optimized prompt design and batch processing to balance performance and efficiency.
Regulatory and ethical compliance: Given the sensitive nature of medical content, strict regulatory and ethical safeguards were critical. We established robust documentation practices for AI decisions, implemented strict version control for medical guidelines and continuous human medical expert oversight for the AI-generated suggestions. Regional healthcare regulations were carefully considered throughout implementation.
Integration and scaling: We recommend starting with a standalone testing environment while planning for future content management system (CMS) integration through well-designed API endpoints. Building with modularity in mind proved valuable for future enhancements. Throughout the process, we faced common challenges such as maintaining context in long medical articles, balancing processing speed with accuracy, and facilitating consistent tone across AI-suggested revisions.
Model optimization: The diverse model selection capability of Amazon Bedrock proved particularly valuable. Through its platform, we can choose optimal models for specific tasks, achieve cost efficiency without sacrificing accuracy, and smoothly upgrade to newer models – all while maintaining our existing architecture.

Preliminary Results
Our Proof of Concept delivered strong results across the critical success metrics, demonstrating the potential of AI-assisted medical content review. The solution exceeded target processing speed improvements while maintaining 80% accuracy and over 90% recall in identifying content requiring updates. Most notably, the AI-powered system applied medical guidelines more consistently than manual reviews and significantly reduced the time burden on medical experts.
Key Takeaways
During implementation, we uncovered several insights critical for optimizing AI performance in medical content analysis. Content chunking was essential for accurate assessment across long documents, and expert validation of parsing rules helped medical experts to maintain clinical precision.Most importantly, the project confirmed that human-AI collaboration – not full automation – is key to successful implementation. Regular expert feedback and clear performance metrics guided system refinements and incremental improvements. While the system significantly streamlines the review process, it works best as an augmentation tool, with medical experts remaining essential for final validation, creating a more efficient hybrid approach to medical content management.
Conclusion and next steps
This first part of our series demonstrates how generative AI can make the medical content review process faster, more efficient, and scalable while maintaining high accuracy. Stay tuned for Part 2 of this series, where we cover the production journey, deep diving into challenges and scaling strategies.Are you ready to move your AI initiatives into production?

Learn more about the AWS Generative AI Innovation Center and contact your AWS Account Manager to be connected to our expert guidance and support.
Visit the Amazon Bedrock documentation to learn more about available foundation models and their capabilities
Join our AWS Builder community to connect with others on a similar AI journey.

About the authors
Liza (Elizaveta) Zinovyeva, Ph.D., is an Applied Scientist at AWS Generative AI Innovation Center and is based in Berlin. She helps customers across different industries to integrate Generative AI into their existing applications and workflows. She is passionate about AI/ML, finance and software security topics. In her spare time, she enjoys spending time with her family, sports, learning new technologies, and table quizzes.
Callum Macpherson is a Data Scientist at the AWS Generative AI Innovation Center, where cutting-edge AI meets real-world business transformation. Callum partners directly with AWS customers to design, build, and scale generative AI solutions that unlock new opportunities, accelerate innovation, and deliver measurable impact across industries.
Arefeh Ghahvechi is a Senior AI Strategist at the AWS GenAI Innovation Center, specializing in helping customers realize rapid value from generative AI technologies by bridging innovation and implementation. She identifies high-impact AI opportunities while building the organizational capabilities needed for scaled adoption across enterprises and national initiatives.
Nuno Castro is a Sr. Applied Science Manager. He’s has 19 years experience in the field in industries such as finance, manufacturing, and travel, leading ML teams for 11 years.
Dmitrii Ryzhov is a Senior Account Manager at Amazon Web Services (AWS), helping digital-native companies unlock business potential through AI, generative AI, and cloud technologies. He works closely with customers to identify high-impact business initiatives and accelerate execution by orchestrating strategic AWS support, including access to the right expertise, resources, and innovation programs.
Nikita Kozodoi, PhD, is a Senior Applied Scientist at the AWS Generative AI Innovation Center working on the frontier of AI research and business. Nikita builds and deploys generative AI and ML solutions that solve real-world problems and drive business impact for AWS customers across industries.
Aiham Taleb, PhD, is a Senior Applied Scientist at the Generative AI Innovation Center, working directly with AWS enterprise customers to leverage Gen AI across several high-impact use cases. Aiham has a PhD in unsupervised representation learning, and has industry experience that spans across various machine learning applications, including computer vision, natural language processing, and medical imaging.

Organizations handle vast amounts of sensitive customer information through various communication channels. Protecting Personally Identifiable Information (PII), such as social security numbers (SSNs), driver’s license numbers, and phone numbers has become increasingly critical for maintaining compliance with data privacy regulations and building customer trust. However, manually reviewing and redacting PII is time-consuming, error-prone, and scales poorly as data volumes grow.
Organizations face challenges when dealing with PII scattered across different content types – from texts to images. Traditional approaches often require separate tools and workflows for handling text and image content, leading to inconsistent redaction practices and potential security gaps. This fragmented approach not only increases operational overhead but also raises the risk of accidental PII exposure.
This post shows an automated PII detection and redaction solution using Amazon Bedrock Data Automation and Amazon Bedrock Guardrails through a use case of processing text and image content in high volumes of incoming emails and attachments. The solution features a complete email processing workflow with a React-based user interface for authorized personnel to more securely manage and review redacted email communications and attachments. We walk through the step-by-step solution implementation procedures used to deploy this solution. Finally, we discuss the solution benefits, including operational efficiency, scalability, security and compliance, and adaptability.
Solution overview
The solution provides an automated system for protecting sensitive information in business communications through three main capabilities:

Automated PII detection and redaction for both email content and attachments using Amazon Bedrock Data Automation and Guardrails, making sure that sensitive data is consistently protected across different content types.
More secure data management workflows where processed communications are encrypted and stored with appropriate access controls, while maintaining a complete audit trail of operations.
Web-based interface options for authorized agents to efficiently manage redacted communications, supported by features like automated email categorization and customizable folder management.

This unified approach helps organizations maintain compliance with data privacy requirements while streamlining their communication workflows.
The following diagram outlines the solution architecture. 
The diagram illustrates the backend PII detection and redaction workflow and the frontend application user interface orchestrated by AWS Lambda and Amazon EventBridge. The process follows these steps:

The workflow starts with the user sending an email to the incoming email server hosted on Amazon Simple Email Service (Amazon SES). This is an optional step.
Alternatively, users can upload the emails and attachments directly into an Amazon Simple Storage Service (S3) landing bucket.
An S3 event notification triggers the initial processing AWS Lambda function that generates a unique case ID and creates a tracking record in Amazon DynamoDB.
Lambda orchestrates the PII detection and redaction workflow by extracting email body and attachments from the email and saving it in a raw email bucket followed by invoking Amazon Bedrock Data Automation and Guardrails for detecting and redacting PII.
Amazon Bedrock Data Automation processes attachments to extract text from the files.
Amazon Bedrock Guardrails detects and redacts the PII from both email body and text from attachments and stores the redacted content in another S3 bucket.
DynamoDB tables are updated with email messages, folders metadata, and email filtering rules.
An Amazon EventBridge Scheduler is used to run the Rules Engine Lambda on a schedule which processes new emails that have yet to be categorized into folders based on enabled email filtering rules criteria.
The Rules Engine Lambda also communicates with DynamoDB to access the messages table and the rules table.
Users can access the optional application user interface through Amazon API Gateway, which manages user API requests and routes requests to render the user interface through S3 static hosting. Users may choose to enable authentication for the user interface based on their security requirements. Alternatively, users can check the status of their email processing in the DynamoDB table and S3 bucket with PII redacted content.
A Portal API Lambda fetches the case details based on user requests.
The static assets served by API Gateway are stored in a private S3 bucket.
Optionally, users may enable Amazon CloudWatch and AWS CloudTrail to provide visibility into the PII detection and redaction process, while using Amazon Simple Notification Service to deliver real-time alerts for any failures, facilitating immediate attention to issues.

In the following sections, we walk through the procedures for implementing this solution.
Walkthrough
The solution implementation involves infrastructure and optional portal setup.
Prerequisites
Before beginning the implementation, make sure to have the following components installed and configured.

An AWS account
Git
Python 3.7 or higher
Node v18 or higher
NPM v9.8 or higher
AWS CDK v2.166 or higher
Terminal/CLI such as macOS Terminal, PowerShell or Windows Terminal, or the Linux command line. AWS CloudShell can also be used when all code is located within an AWS account

Infrastructure setup and deployment process
Verify that an existing virtual private cloud VPC that contains three private subnets with no internet access is created in your AWS account. All AWS CloudFormation stacks need to be deployed within the same AWS account.
CloudFormation stacks
The solution contains three stacks (two required, one optional) that deploys in your AWS account:

S3Stack – Provisions the core infrastructure including S3 buckets for raw and redacted email storage with automatic lifecycle policies, a DynamoDB table for email metadata tracking with time-to-live (TTL) and global secondary indexes, and VPC security groups for more secure Lambda function access. It also creates Amazon Identity and Access Management (IAM) roles with comprehensive permissions for S3, DynamoDB, and Bedrock services, forming a more secure foundation for the entire PII detection and redaction workflow.
ConsumerStack – Provisions the core processing infrastructure including Amazon Bedrock Data Automation projects for document text extraction and Bedrock Guardrails configured to anonymize comprehensive PII entities, along with Lambda functions for email and attachment processing with Amazon Simple Notification Service (SNS) topics for success/failure notifications. It also creates Amazon Simple Email Service (SES) receipt rules for incoming email handling when a domain is configured and S3 event notifications to trigger the email processing workflow automatically.
PortalStack (optional) – This is only needed when users want to use a web-based user interface for managing emails. It provisions the optional web interface including a regional API Gateway, DynamoDB tables for redacted message storage, and S3 buckets for static web assets.

Amazon SES (optional)
Move directly to the Solution Deployment section that follows if Amazon SES is not being used.
The following Amazon SES Setup is optional. The code may be tested without this setup as well. Steps to test the application with or without Amazon SES is covered in the Testing section.
Set up Amazon SES with prod access and verify the domain/email identities for which the solution is to work. We also need to add the MX records in the DNS provider maintaining the domain. Please refer to the following links:

Request SES Production Access
Setting up Amazon SES email receiving

Create credentials for SMTP and save it in AWS Secrets Manager secret with name SmtpCredentials. An IAM user is created for this process.
If any other name is being used for the secret, update the context.json line secret_name with the name of the secret created.
The key for the username in the secret should be smtp_username and the key for password should be smtp_password when storing the same in AWS Secrets Manager.

Obtaining Amazon SES SMTP credentials

Solution deployment
Run the following commands from within a terminal/CLI environment.

Clone the repository

git clone https://github.com/aws-samples/sample-bda-redaction.git

The infra/cdk.json file tells the CDK Toolkit how to execute your app

cd sample-bda-redaction/infra/

Optional: Create and activate a new Python virtual environment (make sure to use python 3.12 as lambda is in CDK is configured for same. If using some other python version update CDK code to reflect the same in lambda runtime)

python3 -m venv .venv
. .venv/bin/activate

Upgrade pip

pip install –upgrade pip

Install Python packages

pip install -r requirements.txt

Create context.json file

cp context.json.example context.json

Update the context.json file with the correct configuration options for the environment.

Property Name
Default
Description
When to Create

vpc_id
“”
VPC ID where resources are deployed
VPC needs to be created prior to execution

raw_bucket
“”
S3 bucket storing raw messages and attachments
Created during CDK deployment

redacted_bucket_name
“”
S3 bucket storing redacted messages and attachments
Created during CDK deployment

inventory_table_name
“”
DynamoDB table name storing redacted message details
Created during CDK deployment

resource_name_prefix
“”
Prefix used when naming resources during the stack creation
During stack creation

retention
90
Number of days for retention of the messages in the redacted and raw S3 buckets
During stack creation

The following properties are only required when the portal is being provisioned.

Property Name
Default
Description

environment
development
The type of environment where resources are provisioned. Values are development or production

Use cases that require the usage of Amazon SES to manage redacted email messages need to set the following configuration variables. Otherwise, these are optional.

Property Name
Description
Comment

domain
The verified domain or email name that is used for Amazon SES
This can be left blank if not setting up Amazon SES

auto_reply_from_email
Email address of the “from” field of the email message. Also used as the email address where emails are forwarded from the Portal application
This can be left blank if not setting up the Portal

secret_name
AWS Secrets Manager secret containing SMTP credentials for forward email functionality from the portal

Deploy Infrastructure by running the following commands from the root of the infra directory.

Bootstrap the AWS account to use AWS CDK

cdk bootstrap

Users can now synthesize the CloudFormation template for this code. Additional environment variables before the cdk synth suppresses the warnings. The deployment process should take approximately 10 min for a first-time deployment to complete.

JSII_DEPRECATED=quiet
JSII_SILENCE_WARNING_UNTESTED_NODE_VERSION=quiet
cdk synth –no-notices

Replace <<resource_name_prefix>> with its chosen value and then run:

JSII_DEPRECATED=quiet
JSII_SILENCE_WARNING_UNTESTED_NODE_VERSION=quiet
cdk deploy <<resource_name_prefix>>-S3Stack <<resource_name_prefix>>-ConsumerStack –no-notices

Testing

Testing the application with Amazon SES

Before starting the test, make sure the Amazon SES Email Receiving rule set that was created by the <<resource_name_prefix>>-ConsumerStack stack is active. We can check by executing the below command and make sure name in the output is <<resource_name_prefix>>-rule-setaws ses describe-active-receipt-rule-set. If the name does not match or the output is blank, execute the following to activate the same:

# Replace <<resource_name_prefix>> with resource_name_prefix used in context.json

aws ses set-active-receipt-rule-set –rule-set-name <<resource_name_prefix>>-rule-set

Once we have the correct rule set active, we can test the application using Amazon SES by sending an email to the verified email or domain in Amazon SES, which automatically triggers the redaction pipeline. Progress can be tracked in the DynamoDB table <<inventory_table_name>>. The inventory table name can be found on the resources tab in the AWS CloudFormation Console for the <<resource_name_prefix>>-S3Stack stack and Logical ID EmailInventoryTable. A unique <<case_id>> is generated and used in the DynamoDB inventory table for each email being processed. Once redaction is complete, the redacted email body can be found in <<redacted_bucket_name>>/redacted/<<today_date>>/<<case_id>>/email_body/ and redacted attachments in <<redacted_bucket_name>>/redacted/<<today_date>>/<<case_id>>/attachments/.

Testing the application without Amazon SES

As described earlier, the solution is used to redact any PII data in the email body and attachments. Therefore, to test the application, we need to provide an email file which needs to be redacted. We can do that without Amazon SES by directly uploading an email file to the raw S3 bucket. The raw bucket name can be found on the output tab in the AWS CloudFormation Console for <<resource_name_prefix>>-S3Stack stack and Export Name RawBucket. This triggers the workflow of redacting the email body and attachments by S3 event notification triggering the Lambda. For your convenience, a sample email is available in the infra/pii_redaction/sample_email directory of the repository. Below are the steps to test the application without Amazon SES using the same email file.

# Replace <<raw_bucket>> with raw bucket name created during deployment

aws s3 cp pii_redaction/sample_email/ccvod0ot9mu6s67t0ce81f8m2fp5d2722a7hq8o1 s3://<<raw_bucket>>/domain_emails/

The above triggers the redaction of the email process. You can track the progress in the DynamoDB table <<inventory_table_name>>. A unique <<case_id>> is generated and used in the DynamoDB inventory table for each email being processed. The inventory table name can be found on the resources tab in the AWS CloudFormation Console for <<resource_name_prefix>>-S3Stack stack and Logical ID EmailInventoryTable. Once redaction is complete, the redacted email body can be found in <<redacted_bucket_name>>/redacted/<<today_date>>/<<case_id>>/email_body/ and redacted attachments in <<redacted_bucket_name>>/redacted/<<today_date>>/<<case_id>>/attachments/.
Portal setup
The installation of the portal is completely optional. This section can be skipped; check the console of the AWS account where the solution is deployed to view the resources created. The portal serves as a web interface to manage the PII-redacted emails processed by the backend AWS infrastructure, allowing users to view sanitized email content. The Portal can be used to:

List messages: View processed emails with redacted content
Message details: View individual email content and attachments

Portal Prerequisites: This portal requires the installation of the following software tools:

TypeScript
Node v18 or higher
NPM v9.8 or higher

Infrastructure Deployment

Synthesize the CloudFormation template for this code by going to the directory root of the solution. Now run the following command:

cd sample-bda-redaction/infra/

Optional: Create and activate a new Python virtual environment (if the virtual environment has not been created previously):

python3 -m venv .venv. .venv/bin/activatepip install -r requirements.txt

Users can now synthesize the CloudFormation template for this code.

JSII_DEPRECATED=quiet
JSII_SILENCE_WARNING_UNTESTED_NODE_VERSION=quiet
cdk synth –no-notices

Deploy the React-based portal. Replace <<resource_name_prefix>> with its chosen value:

JSII_DEPRECATED=quiet
JSII_SILENCE_WARNING_UNTESTED_NODE_VERSION=quiet
cdk deploy <<resource_name_prefix>>-PortalStack –no-notices

The first-time deployment should take approximately 10 minutes to complete.
Environment Variables

Create a new environment file by going to the root of the app directory and update the following variables in the .env file (by copying the .env.example file to .env) using the following command to create the .env file using a terminal/CLI environment.

cp .env.example .env

The file can be created using your preferred text editor as well.

Environment Variable Name
Default
Description
Required

VITE_APIGW
“”
URL of the API Gateway invokes URL (including protocol) without the path (remove /portal from the value). This value can be found in the output of the PortalStack after deploying through AWS CDK. It can also be found under the Outputs tab of the PortalStack CloudFormation stack under the export name of PiiPortalApiGatewayInvokeUrl
Yes

VITE_BASE
/portal
It specifies the path used to request the static files needed to render the portal
Yes

VITE_API_PATH
/api
It specifies the path needed to send requests to the API Gateway
Yes

Portal deployment
Run the following commands from within a terminal/CLI environment.

Before running any of the following commands, go to the root of the app directory to build this application for production by running the following commands:

Install NPM packages

npm install

Build the files

npm run build

After the build succeeds, transfer all of the files within the dist/ directory into the Amazon S3 bucket that is designated for these assets (specified in the PortalStack provisioned via CDK).

Example: aws s3 sync dist/ s3://<<name-of-s3-bucket>> –delete<<name-of-s3-bucket>> is the S3 bucket that has been created in the <<resource-name-prefix>>-PortalStack CloudFormation stack with the Logical ID of PrivateWebHostingAssets. This value can be obtained from the Resources tab of the CloudFormation stack in the AWS Console. This value is also output during the cdk deploy process when the PortalStack has been successfully completed.

Accessing the portal
Use the API Gateway invoke URL from the API Gateway that has been created during the cdk deploy process to access the portal from a web browser. This URL can be found by following these steps:

Visit the AWS Console
Go to API Gateway and find the API Gateway that has been created during the cdk deploy process. The name of the API Gateway can be found in the Resources section of the <<resource-name-prefix>>-PortalStack CloudFormation stack.
Click on the Stages link in the left-hand menu.
Make sure that the portal stage is selected
Find the Invoke URL and copy that value
Enter that value in the address bar of your web browser

The portal’s user interface is now visible within the web browser. If any emails have been processed, they are listed on the home page of the portal.
Access control (optional)
For production deployment, we recommend these approaches to controlling and managing access to the Portal.
Clean up
To avoid incurring future charges, follow these steps to remove the resources created by this solution:

Delete the contents of the S3 buckets created by the solution:

Raw email bucket
Redacted email bucket
Portal static assets bucket (if portal was deployed)

Delete or disable the Amazon SES rule step created by the solution using below cli command:

#to disable the rule set use below command
aws ses set-active-receipt-rule-set

#to delete the rule set use below command
# Replace <<resource_name_prefix>> with resource_name_prefix used in context.json
aws ses delete-receipt-rule-set –rule-set-name <resource_name_prefix>>-rule-set

Remove the CloudFormation stacks in the following order:

cdk destroy <<resource_name_prefix>>-PortalStack (if deployed)
cdk destroy <<resource_name_prefix>>-ConsumerStack
cdk destroy <<resource_name_prefix>>-S3Stack

CDK Destroy does not remove the access log Amazon S3 bucket created as part of the deployment. Users can get access to the log bucket name in the output tab of stack <<resource_name_prefix>>-S3Stack with export name AccessLogsBucket. Execute the below steps to delete the access log bucket:

To delete the contents of the access log bucket, follow the instructions on deleting S3 bucket
Access to the log bucket is version-enabled and deleting the content of the bucket in the above step does not delete versioned objects in the bucket. That needs to be removed separately using below aws cli commands:

#to remove versioned objects use below aws cli command
aws s3api delete-objects –bucket ${accesslogbucket} –delete “$(aws s3api list-object-versions –bucket ${accesslogbucket} –query='{Objects: Versions[].{Key:Key,VersionId:VersionId}}’)”

#once versioned objects are removed we need to remove the delete markers of the versioned objects using below aws cli command
aws s3api delete-objects –bucket ${accesslogbucket} –delete “$(aws s3api list-object-versions –bucket ${accesslogbucket} –query='{Objects: DeleteMarkers[].{Key:Key,VersionId:VersionId}}’)”

Delete the access log Amazon S3 bucket using below aws cli command:

#delete the access log bucket itself using below aws cli command
aws s3api delete-bucket –bucket ${accesslogbucket}

If Amazon SES is configured:

Remove the verified domain/email identities
Delete the MX records from your DNS provider
Delete the SMTP credentials from AWS Secrets Manager

Delete any CloudWatch Log groups created by the Lambda functions

The VPC and its associated resources as prerequisites for this solution may not be deleted if they may be used by other applications.
Conclusion
In this post, we demonstrated how to automate the detection and redaction of PII across both text and image content using Amazon Bedrock Data Automation and Amazon Bedrock Guardrails. By centralizing and streamlining the redaction process, organizations can strengthen alignment with data privacy requirements, enhance security practices, and minimize operational overhead.
However, it is equally important to make sure that your solution is built with Amazon Bedrock Data Automation’s document processing constraints in mind. Amazon Bedrock Data Automation supports PDF, JPEG, and PNG file formats with a maximum console-processing size of 200 MB (500 MB via API), and single documents may not exceed 20 pages unless document splitting is enabled.
By using Amazon Bedrock Data Automation and Amazon Bedrock Guardrails centralized redaction capabilities, organizations can boost data privacy compliance management, cut operational overhead, and maintain stringent security across diverse workloads. This solution’s extensibility further enables integration with other AWS services, fine-tuning detection logic for more advanced PII patterns, and broadening support for additional file types or languages in the future, thereby evolving into a more robust, enterprise-scale data protection framework.
We encourage exploration of the provided GitHub repository to deploy this solution within your organization. In addition to delivering operational efficiency, scalability, security, and adaptability, the solution also provides a unified interface and robust audit trail that simplifies data governance. By refining detection rules, users can integrate additional file formats where possible and use Amazon Bedrock Data Automation and Amazon Bedrock Guardrails modular framework.
We invite you to implement this PII detection and redaction solution in the following GitHub repo to build a more secure, compliance-aligned, and highly adaptable data protection solution on Amazon Bedrock that addresses evolving business and regulatory requirements.

About the Authors
Himanshu Dixit is a Delivery Consultant at AWS Professional Services specializing in databases and analytics, bringing over 18 years of experience in technology. He is passionate for artificial intelligence, machine learning, and generative AI, leveraging these cutting-edge technologies to create innovative solutions that address real-world challenges faced by customers. Outside of work, he enjoys playing badminton, tennis, cricket, table tennis and spending time with her two daughters.
David Zhang is an Engagement Manager at AWS Professional Services, where he leads enterprise-scale AI/ML, cloud transformation initiatives for Fortune 100 customers in telecom, finance, media, and entertainment. Outside of work, he enjoys experimenting with new recipes in his kitchen, playing tenor saxophone, and capturing life’s moments through his camera.
Richard Session is a Lead User Interface Developer for AWS ProServe, bringing over 15 years of experience as a full-stack developer across marketing/advertising, enterprise technology, automotive, and ecommerce industries. With a passion for creating intuitive and engaging user experiences, he uses his extensive background to craft exceptional interfaces for AWS’s enterprise customers. When he’s not designing innovative user experiences, Richard can be found pursuing his love for coffee, spinning tracks as a DJ, or exploring new destinations around the globe.
Viyoma Sachdeva is a Principal Industry Specialist in AWS. She is specialized in AWS DevOps, containerization and IoT helping Customer’s accelerate their journey to AWS Cloud.

In this tutorial, we demonstrate how to build a unified Apache Beam pipeline that works seamlessly in both batch and stream-like modes using the DirectRunner. We generate synthetic, event-time–aware data and apply fixed windowing with triggers and allowed lateness to demonstrate how Apache Beam consistently handles both on-time and late events. By switching only the input source, we keep the core aggregation logic identical, which helps us clearly understand how Beam’s event-time model, windows, and panes behave without relying on external streaming infrastructure. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browser!pip -q install -U “grpcio>=1.71.2” “grpcio-status>=1.71.2”
!pip -q install -U apache-beam crcmod

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions
from apache_beam.transforms.window import FixedWindows
from apache_beam.transforms.trigger import AfterWatermark, AfterProcessingTime, AccumulationMode
from apache_beam.testing.test_stream import TestStream
import json
from datetime import datetime, timezone

We install the required dependencies and ensure version compatibility so that Apache Beam. We import the core Beam APIs along with windowing, triggers, and TestStream utilities needed later in the pipeline. We also bring in standard Python modules for time handling and JSON formatting. Check out the FULL CODES here.

Copy CodeCopiedUse a different BrowserMODE = “stream”
WINDOW_SIZE_SECS = 60
ALLOWED_LATENESS_SECS = 120

def make_event(user_id, event_type, amount, event_time_epoch_s):
return {“user_id”: user_id, “event_type”: event_type, “amount”: float(amount), “event_time”: int(event_time_epoch_s)}

base = datetime.now(timezone.utc).replace(microsecond=0)
t0 = int(base.timestamp())

BATCH_EVENTS = [
make_event(“u1”, “purchase”, 20, t0 + 5),
make_event(“u1”, “purchase”, 15, t0 + 20),
make_event(“u2”, “purchase”, 8, t0 + 35),
make_event(“u1”, “refund”, -5, t0 + 62),
make_event(“u2”, “purchase”, 12, t0 + 70),
make_event(“u3”, “purchase”, 9, t0 + 75),
make_event(“u2”, “purchase”, 3, t0 + 50),
]

We define the global configuration that controls window size, lateness, and execution mode. We create synthetic events with explicit event-time timestamps so that windowing behavior is deterministic and easy to reason about. We prepare a small dataset that intentionally includes out-of-order and late events to observe Beam’s event-time semantics. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browserdef format_joined_record(kv):
user_id, d = kv
return {
“user_id”: user_id,
“count”: int(d[“count”][0]) if d[“count”] else 0,
“sum_amount”: float(d[“sum_amount”][0]) if d[“sum_amount”] else 0.0,
}

class WindowedUserAgg(beam.PTransform):
def expand(self, pcoll):
stamped = pcoll | beam.Map(lambda e: beam.window.TimestampedValue(e, e[“event_time”]))
windowed = stamped | beam.WindowInto(
FixedWindows(WINDOW_SIZE_SECS),
allowed_lateness=ALLOWED_LATENESS_SECS,
trigger=AfterWatermark(
early=AfterProcessingTime(10),
late=AfterProcessingTime(10),
),
accumulation_mode=AccumulationMode.ACCUMULATING,
)
keyed = windowed | beam.Map(lambda e: (e[“user_id”], e[“amount”]))
counts = keyed | beam.combiners.Count.PerKey()
sums = keyed | beam.CombinePerKey(sum)
return (
{“count”: counts, “sum_amount”: sums}
| beam.CoGroupByKey()
| beam.Map(format_joined_record)
)

We build a reusable Beam PTransform that encapsulates all windowed aggregation logic. We apply fixed windows, triggers, and accumulation rules, then group events by user and compute counts and sums. We keep this transform independent of the data source, so the same logic applies to both batch and streaming inputs. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browserclass AddWindowInfo(beam.DoFn):
def process(self, element, window=beam.DoFn.WindowParam, pane_info=beam.DoFn.PaneInfoParam):
ws = float(window.start)
we = float(window.end)
yield {
**element,
“window_start_utc”: datetime.fromtimestamp(ws, tz=timezone.utc).strftime(“%H:%M:%S”),
“window_end_utc”: datetime.fromtimestamp(we, tz=timezone.utc).strftime(“%H:%M:%S”),
“pane_timing”: str(pane_info.timing),
“pane_is_first”: pane_info.is_first,
“pane_is_last”: pane_info.is_last,
}

def build_test_stream():
return (
TestStream()
.advance_watermark_to(t0)
.add_elements([
beam.window.TimestampedValue(make_event(“u1”, “purchase”, 20, t0 + 5), t0 + 5),
beam.window.TimestampedValue(make_event(“u1”, “purchase”, 15, t0 + 20), t0 + 20),
beam.window.TimestampedValue(make_event(“u2”, “purchase”, 8, t0 + 35), t0 + 35),
])
.advance_processing_time(5)
.advance_watermark_to(t0 + 61)
.add_elements([
beam.window.TimestampedValue(make_event(“u1”, “refund”, -5, t0 + 62), t0 + 62),
beam.window.TimestampedValue(make_event(“u2”, “purchase”, 12, t0 + 70), t0 + 70),
beam.window.TimestampedValue(make_event(“u3”, “purchase”, 9, t0 + 75), t0 + 75),
])
.advance_processing_time(5)
.add_elements([
beam.window.TimestampedValue(make_event(“u2”, “purchase”, 3, t0 + 50), t0 + 50),
])
.advance_watermark_to(t0 + 121)
.advance_watermark_to_infinity()
)

We enrich each aggregated record with window and pane metadata so we can clearly see when and why results are emitted. We convert Beam’s internal timestamps into human-readable UTC times for clarity. We also define a TestStream that simulates real streaming behavior using watermarks, processing-time advances, and late data. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browserdef run_batch():
with beam.Pipeline(options=PipelineOptions([])) as p:
(
p
| beam.Create(BATCH_EVENTS)
| WindowedUserAgg()
| beam.ParDo(AddWindowInfo())
| beam.Map(json.dumps)
| beam.Map(print)
)

def run_stream():
opts = PipelineOptions([])
opts.view_as(StandardOptions).streaming = True
with beam.Pipeline(options=opts) as p:
(
p
| build_test_stream()
| WindowedUserAgg()
| beam.ParDo(AddWindowInfo())
| beam.Map(json.dumps)
| beam.Map(print)
)

run_stream() if MODE == “stream” else run_batch()

We wire everything together into executable batch and stream-like pipelines. We toggle between modes by changing a single flag while reusing the same aggregation transform. We run the pipeline and print the windowed results directly, making the execution flow and outputs easy to inspect.

In conclusion, we demonstrated that the same Beam pipeline can process both bounded batch data and unbounded, stream-like data while preserving identical windowing and aggregation semantics. We observed how watermarks, triggers, and accumulation modes influence when results are emitted and how late data updates previously computed windows. Also, we focused on the conceptual foundations of Beam’s unified model, providing a solid base for later scaling the same design to real streaming runners and production environments.

Check out the FULL CODES here. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.

Check out our latest release of ai2025.dev, a 2025-focused analytics platform that turns model launches, benchmarks, and ecosystem activity into a structured dataset you can filter, compare, and export
The post A Coding Implementation to Build a Unified Apache Beam Pipeline Demonstrating Batch and Stream Processing with Event-Time Windowing Using DirectRunner appeared first on MarkTechPost.

Technology Innovation Institute (TII), Abu Dhabi, has released Falcon-H1R-7B, a 7B parameter reasoning specialized model that matches or exceeds many 14B to 47B reasoning models in math, code and general benchmarks, while staying compact and efficient. It builds on Falcon H1 7B Base and is available on Hugging Face under the Falcon-H1R collection.

Falcon-H1R-7B is interesting because it combines 3 design choices in 1 system, a hybrid Transformer along with Mamba2 backbone, a very long context that reaches 256k tokens in standard vLLM deployments, and a training recipe that mixes supervised long form reasoning with reinforcement learning using GRPO.

Hybrid Transformer plus Mamba2 architecture with long context

Falcon-H1R-7B is a causal decoder only model with a hybrid architecture that combines Transformer layers and Mamba2 state space components. The Transformer blocks provide standard attention based reasoning, while the Mamba2 blocks give linear time sequence modeling and better memory scaling as context length grows. This design targets the 3 axes of reasoning efficiency that the team describes, speed, token efficiency and accuracy.

The model runs with a default –max-model-len of 262144 when served through vLLM, which corresponds to a practical 256k token context window. This allows very long chain of thought traces, multi step tool use logs and large multi document prompts in a single pass. The hybrid backbone helps control memory use at these sequence lengths and improves throughput compared with a pure Transformer 7B baseline on the same hardware.

Training recipe for reasoning tasks

Falcon H1R 7B uses a 2 stage training pipeline:

In the first stage, the team runs cold start supervised fine tuning on top of Falcon-H1-7B Base. The SFT (supervised fine tuning) data mixes step by step long form reasoning traces in 3 main domains, mathematics, coding and science, plus non reasoning domains such as chat, tool calling and safety. Difficulty aware filtering upweights harder problems and downweights trivial ones. Targets can reach up to 48k tokens, so the model sees long derivations and full solution paths during training.

In the second stage, the SFT checkpoint is refined with GRPO, which is a group relative policy optimization method for reinforcement learning. Rewards are given when the generated reasoning chain is verifiably correct. For math problems, the system uses symbolic checks on the final answer. For code, it executes the generated program against unit tests. This RL stage pushes the model to keep useful intermediate steps while staying within a token budget.

The result is a 7B model that is tuned specifically for chain of thought reasoning, rather than general chat.

Benchmarks in math, coding and general reasoning

The Falcon-H1R-7B benchmark scores are grouped across math, code and agentic tasks, and general reasoning tasks.

In the math group, Falcon-H1R-7B reaches an aggregate score of 73.96%, ahead of Apriel-1.5-15B at 69.32% and larger models like Qwen3-32B and Nemotron-H-47B. On individual benchmarks:

AIME 24, 88.1%, higher than Apriel-1.5-15B at 86.2%

AIME 25, 83.1%, higher than Apriel-1.5-15B at 80%

HMMT 25, 64.9%, above all listed baselines

AMO Bench, 36.3%, compared with 23.3% for DeepSeek-R1-0528 Qwen3-8B

For code and agentic workloads, the model reaches 33.95% as a group score. On LiveCodeBench v6, Falcon-H1R-7B scores 68.6%, which is higher than Qwen3-32B and other baselines. It also scores 28.3% on the SciCode sub problem benchmark and 4.9% on Terminal Bench Hard, where it ranks second behind Apriel 1.5-15B but ahead of several 8B and 32B systems.

https://huggingface.co/blog/tiiuae/falcon-h1r-7b

On general reasoning, Falcon-H1R-7B achieves 49.48% as a group score. It records 61.3% on GPQA D, close to other 8B models, 72.1% on MMLU Pro, which is higher than all other 8B models in the above table, 11.1% on HLE and 53.4% on IFBench, where it is second only to Apriel 1.5 15B.

The key takeaway is that a 7B model can sit in the same performance band as many 14B to 47B reasoning models, if the architecture and training pipeline are tuned for reasoning tasks.

Inference throughput and test time scaling

The team also benchmarked Falcon-H1R-7B on throughput and test time scaling under realistic batch settings.

For a 512 token input and 32k token output, Falcon-H1R-7B reaches about 1,000 tokens per second per GPU at batch size 32 and about 1,500 tokens per second per GPU at batch size 64, nearly double the throughput of Qwen3-8B in the same configuration. For an 8k input and 16k output, Falcon-H1R-7B reaches around 1,800 tokens per second per GPU, while Qwen3-8B stays below 900. The hybrid Transformer along with Mamba architecture is a key factor in this scaling behavior, because it reduces the quadratic cost of attention for long sequences.

Falcon-H1R-7B is also designed for test time scaling using Deep Think with confidence, known as DeepConf. The idea is to run many chains of thought in parallel, then use the model’s own next token confidence scores to filter noisy traces and keep only high quality candidates.

On AIME 24 and AIME 25, Falcon-H1R-7B reaches 96.7% accuracy with fewer than 100 million generated tokens, which puts it on a favorable Pareto frontier of accuracy versus token cost compared with other 8B, 14B and 32B reasoning models. On the parser verifiable subset of AMO Bench, it reaches 35.9% accuracy with 217 million tokens, again ahead of the comparison models at similar or larger scale.

Key Takeaways

Falcon-H1R-7B is a 7B parameter reasoning model that uses a hybrid Transformer along with Mamba2 architecture and supports a 256k token context for long chain of thought prompts.

The model is trained in 2 stages, supervised fine tuning on long reasoning traces in math, code and science up to 48k tokens, followed by GRPO based reinforcement learning with verifiable rewards for math and code.

Falcon-H1R-7B achieves strong math performance, including about 88.1% on AIME 24, 83.1% on AIME 25 and a 73.96% aggregate math score, which is competitive with or better than larger 14B to 47B models.

On coding and agentic tasks, Falcon-H1R-7B obtains 33.95% as a group score and 68.6% on LiveCodeBench v6, and it is also competitive on general reasoning benchmarks such as MMLU Pro and GPQA D.

The hybrid design improves throughput, reaching around 1,000 to 1,800 tokens per second per GPU in the reported settings, and the model supports test time scaling through Deep Think with confidence to improve accuracy using multiple reasoning samples under a controlled token budget.

Check out the Technical details and MODEL WEIGHTS here. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.

Check out our latest release of ai2025.dev, a 2025-focused analytics platform that turns model launches, benchmarks, and ecosystem activity into a structured dataset you can filter, compare, and export
The post TII Abu-Dhabi Released Falcon H1R-7B: A New Reasoning Model Outperforming Others in Math and Coding with only 7B Params with 256k Context Window appeared first on MarkTechPost.

In this tutorial, we build a genuinely advanced Agentic AI system using LangGraph and OpenAI models by going beyond simple planner, executor loops. We implement adaptive deliberation, where the agent dynamically decides between fast and deep reasoning; a Zettelkasten-style agentic memory graph that stores atomic knowledge and automatically links related experiences; and a governed tool-use mechanism that enforces constraints during execution. By combining structured state management, memory-aware retrieval, reflexive learning, and controlled tool invocation, we demonstrate how modern agentic systems can reason, act, learn, and evolve rather than respond in a single pass. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browser!pip -q install -U langgraph langchain-openai langchain-core pydantic numpy networkx requests

import os, getpass, json, time, operator
from typing import List, Dict, Any, Optional, Literal
from typing_extensions import TypedDict, Annotated
import numpy as np
import networkx as nx
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.messages import SystemMessage, HumanMessage, ToolMessage, AnyMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

We set up the execution environment by installing all required libraries and importing the core modules. We bring together LangGraph for orchestration, LangChain for model and tool abstractions, and supporting libraries for memory graphs and numerical operations. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browserif not os.environ.get(“OPENAI_API_KEY”):
os.environ[“OPENAI_API_KEY”] = getpass.getpass(“Enter OPENAI_API_KEY: “)

MODEL = os.environ.get(“OPENAI_MODEL”, “gpt-4o-mini”)
EMB_MODEL = os.environ.get(“OPENAI_EMBED_MODEL”, “text-embedding-3-small”)

llm_fast = ChatOpenAI(model=MODEL, temperature=0)
llm_deep = ChatOpenAI(model=MODEL, temperature=0)
llm_reflect = ChatOpenAI(model=MODEL, temperature=0)
emb = OpenAIEmbeddings(model=EMB_MODEL)

We securely load the OpenAI API key at runtime and initialize the language models used for fast, deep, and reflective reasoning. We also configure the embedding model that powers semantic similarity in memory. This separation allows us to flexibly switch reasoning depth while maintaining a shared representation space for memory. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browserclass Note(BaseModel):
note_id: str
title: str
content: str
tags: List[str] = Field(default_factory=list)
created_at_unix: float
context: Dict[str, Any] = Field(default_factory=dict)

class MemoryGraph:
def __init__(self):
self.g = nx.Graph()
self.note_vectors = {}

def _cos(self, a, b):
return float(np.dot(a, b) / ((np.linalg.norm(a) + 1e-9) * (np.linalg.norm(b) + 1e-9)))

def add_note(self, note, vec):
self.g.add_node(note.note_id, **note.model_dump())
self.note_vectors[note.note_id] = vec

def topk_related(self, vec, k=5):
scored = [(nid, self._cos(vec, v)) for nid, v in self.note_vectors.items()]
scored.sort(key=lambda x: x[1], reverse=True)
return [{“note_id”: n, “score”: s, “title”: self.g.nodes[n][“title”]} for n, s in scored[:k]]

def link_note(self, a, b, w, r):
if a != b:
self.g.add_edge(a, b, weight=w, reason=r)

def evolve_links(self, nid, vec):
for r in self.topk_related(vec, 8):
if r[“score”] >= 0.78:
self.link_note(nid, r[“note_id”], r[“score”], “evolve”)

MEM = MemoryGraph()

We construct an agentic memory graph inspired by the Zettelkasten method, where each interaction is stored as an atomic note. We embed each note and connect it to semantically related notes using similarity scores. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browser@tool
def web_get(url: str) -> str:
import urllib.request
with urllib.request.urlopen(url, timeout=15) as r:
return r.read(25000).decode(“utf-8″, errors=”ignore”)

@tool
def memory_search(query: str, k: int = 5) -> str:
qv = np.array(emb.embed_query(query))
hits = MEM.topk_related(qv, k)
return json.dumps(hits, ensure_ascii=False)

@tool
def memory_neighbors(note_id: str) -> str:
if note_id not in MEM.g:
return “[]”
return json.dumps([
{“note_id”: n, “weight”: MEM.g[note_id][n][“weight”]}
for n in MEM.g.neighbors(note_id)
])

TOOLS = [web_get, memory_search, memory_neighbors]
TOOLS_BY_NAME = {t.name: t for t in TOOLS}

We define the external tools the agent can invoke, including web access and memory-based retrieval. We integrate these tools in a structured way so the agent can query past experiences or fetch new information when necessary. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browserclass DeliberationDecision(BaseModel):
mode: Literal[“fast”, “deep”]
reason: str
suggested_steps: List[str]

class RunSpec(BaseModel):
goal: str
constraints: List[str]
deliverable_format: str
must_use_memory: bool
max_tool_calls: int

class Reflection(BaseModel):
note_title: str
note_tags: List[str]
new_rules: List[str]
what_worked: List[str]
what_failed: List[str]

class AgentState(TypedDict, total=False):
run_spec: Dict[str, Any]
messages: Annotated[List[AnyMessage], operator.add]
decision: Dict[str, Any]
final: str
budget_calls_remaining: int
tool_calls_used: int
max_tool_calls: int
last_note_id: str

DECIDER_SYS = “Decide fast vs deep.”
AGENT_FAST = “Operate fast.”
AGENT_DEEP = “Operate deep.”
REFLECT_SYS = “Reflect and store learnings.”

We formalize the agent’s internal representations using structured schemas for deliberation, execution goals, reflection, and global state. We also define the system prompts that guide behavior in fast and deep modes. This ensures the agent’s reasoning and decisions remain consistent, interpretable, and controllable. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browserdef deliberate(st):
spec = RunSpec.model_validate(st[“run_spec”])
d = llm_fast.with_structured_output(DeliberationDecision).invoke([
SystemMessage(content=DECIDER_SYS),
HumanMessage(content=json.dumps(spec.model_dump()))
])
return {“decision”: d.model_dump(), “budget_calls_remaining”: st[“budget_calls_remaining”] – 1}

def agent(st):
spec = RunSpec.model_validate(st[“run_spec”])
d = DeliberationDecision.model_validate(st[“decision”])
llm = llm_deep if d.mode == “deep” else llm_fast
sys = AGENT_DEEP if d.mode == “deep” else AGENT_FAST
out = llm.bind_tools(TOOLS).invoke([
SystemMessage(content=sys),
*st.get(“messages”, []),
HumanMessage(content=json.dumps(spec.model_dump()))
])
return {“messages”: [out], “budget_calls_remaining”: st[“budget_calls_remaining”] – 1}

def route(st):
return “tools” if st[“messages”][-1].tool_calls else “finalize”

def tools_node(st):
msgs = []
used = st.get(“tool_calls_used”, 0)
for c in st[“messages”][-1].tool_calls:
obs = TOOLS_BY_NAME[c[“name”]].invoke(c[“args”])
msgs.append(ToolMessage(content=str(obs), tool_call_id=c[“id”]))
used += 1
return {“messages”: msgs, “tool_calls_used”: used}

def finalize(st):
out = llm_deep.invoke(st[“messages”] + [HumanMessage(content=”Return final output”)])
return {“final”: out.content}

def reflect(st):
r = llm_reflect.with_structured_output(Reflection).invoke([
SystemMessage(content=REFLECT_SYS),
HumanMessage(content=st[“final”])
])
note = Note(
note_id=str(time.time()),
title=r.note_title,
content=st[“final”],
tags=r.note_tags,
created_at_unix=time.time()
)
vec = np.array(emb.embed_query(note.title + note.content))
MEM.add_note(note, vec)
MEM.evolve_links(note.note_id, vec)
return {“last_note_id”: note.note_id}

We implement the core agentic behaviors as LangGraph nodes, including deliberation, action, tool execution, finalization, and reflection. We orchestrate how information flows between these stages and how decisions affect the execution path. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browserg = StateGraph(AgentState)
g.add_node(“deliberate”, deliberate)
g.add_node(“agent”, agent)
g.add_node(“tools”, tools_node)
g.add_node(“finalize”, finalize)
g.add_node(“reflect”, reflect)

g.add_edge(START, “deliberate”)
g.add_edge(“deliberate”, “agent”)
g.add_conditional_edges(“agent”, route, [“tools”, “finalize”])
g.add_edge(“tools”, “agent”)
g.add_edge(“finalize”, “reflect”)
g.add_edge(“reflect”, END)

graph = g.compile(checkpointer=InMemorySaver())

def run_agent(goal, constraints=None, thread_id=”demo”):
if constraints is None:
constraints = []
spec = RunSpec(
goal=goal,
constraints=constraints,
deliverable_format=”markdown”,
must_use_memory=True,
max_tool_calls=6
).model_dump()

return graph.invoke({
“run_spec”: spec,
“messages”: [],
“budget_calls_remaining”: 10,
“tool_calls_used”: 0,
“max_tool_calls”: 6
}, config={“configurable”: {“thread_id”: thread_id}})

We assemble all nodes into a LangGraph workflow and compile it with checkpointed state management. We also define a reusable runner function that executes the agent while preserving memory across runs.

In conclusion, we showed how an agent can continuously improve its behavior through reflection and memory rather than relying on static prompts or hard-coded logic. We used LangGraph to orchestrate deliberation, execution, tool governance, and reflexion as a coherent graph, while OpenAI models provide the reasoning and synthesis capabilities at each stage. This approach illustrated how agentic AI systems can move closer to autonomy by adapting their reasoning depth, reusing prior knowledge, and encoding lessons as persistent memory, forming a practical foundation for building scalable, self-improving agents in real-world applications.

Check out the FULL CODES here. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.

Check out our latest release of ai2025.dev, a 2025-focused analytics platform that turns model launches, benchmarks, and ecosystem activity into a structured dataset you can filter, compare, and export
The post How to Design an Agentic AI Architecture with LangGraph and OpenAI Using Adaptive Deliberation, Memory Graphs, and Reflexion Loops appeared first on MarkTechPost.

Zlab Princeton researchers have released LLM-Pruning Collection, a JAX based repository that consolidates major pruning algorithms for large language models into a single, reproducible framework. It targets one concrete goal, make it easy to compare block level, layer level and weight level pruning methods under a consistent training and evaluation stack on both GPUs and TPUs.

What LLM-Pruning Collection Contains?

It is described as a JAX based repo for LLM pruning. It is organized into three main directories:

pruning holds implementations for several pruning methods: Minitron, ShortGPT, Wanda, SparseGPT, Magnitude, Sheared Llama and LLM-Pruner.

training provides integration with FMS-FSDP for GPU training and MaxText for TPU training.

eval exposes JAX compatible evaluation scripts built around lm-eval-harness, with accelerate based support for MaxText that gives about 2 to 4 times speedup.

Pruning Methods Covered

LLM-Pruning Collection spans several families of pruning algorithms with different granularity levels:

Minitron

Minitron is a practical pruning and distillation recipe developed by NVIDIA that compresses Llama 3.1 8B and Mistral NeMo 12B to 4B and 8B while preserving performance. It explores depth pruning and joint width pruning of hidden sizes, attention and MLP, followed by distillation.

In LLM-Pruning Collection, the pruning/minitron folder provides scripts such as prune_llama3.1-8b.sh which run Minitron style pruning on Llama 3.1 8B.

ShortGPT

ShortGPT is based on the observation that many Transformer layers are redundant. The method defines Block Influence, a metric that measures the contribution of each layer and then removes low influence layers by direct layer deletion. Experiments show that ShortGPT outperforms previous pruning methods for multiple choice and generative tasks.

In the collection, ShortGPT is implemented through the Minitron folder with a dedicated script prune_llama2-7b.sh.

Wanda, SparseGPT, Magnitude

Wanda is a post training pruning method that scores weights by the product of weight magnitude and corresponding input activation on a per output basis. It prunes the smallest scores, requires no retraining and induces sparsity that works well even at billion parameter scale.

SparseGPT is another post training method that uses a second order inspired reconstruction step to prune large GPT style models at high sparsity ratios. Magnitude pruning is the classical baseline that removes weights with small absolute value.

In LLM-Pruning Collection, all three live under pruning/wanda with a shared installation path. The README includes a dense table of Llama 2 7B results that compares Wanda, SparseGPT and Magnitude across BoolQ, RTE, HellaSwag, Winogrande, ARC E, ARC C and OBQA, under unstructured and structured sparsity patterns such as 4:8 and 2:4.

Sheared Llama

Sheared LLaMA is a structured pruning method that learns masks for layers, attention heads and hidden dimensions and then retrains the pruned architecture. The original release provides models at multiple scales including 2.7B and 1.3B.

The pruning/llmshearing directory in LLM-Pruning Collection integrates this recipe. It uses a RedPajama subset for calibration, accessed through Hugging Face, and helper scripts to convert between Hugging Face and MosaicML Composer formats.

LLM-Pruner

LLM-Pruner is a framework for structural pruning of large language models. It removes non critical coupled structures, such as attention heads or MLP channels, using gradient based importance scores and then recovers performance with a short LoRA tuning stage that uses about 50K samples. The collection includes LLM-Pruner under pruning/LLM-Pruner with scripts for LLaMA, LLaMA 2 and Llama 3.1 8B.

Key Takeaways

LLM-Pruning Collection is a JAX based, Apache-2.0 repo from zlab-princeton that unifies modern LLM pruning methods with shared pruning, training and evaluation pipelines for GPUs and TPUs.

The codebase implements block, layer and weight level pruning approaches, including Minitron, ShortGPT, Wanda, SparseGPT, Sheared LLaMA, Magnitude pruning and LLM-Pruner, with method specific scripts for Llama family models.

Training integrates FMS-FSDP on GPU and MaxText on TPU with JAX compatible evaluation scripts built on lm-eval-harness, giving roughly 2 to 4 times faster eval for MaxText checkpoints via accelerate.

The repository reproduces key results from prior pruning work, publishing side by side “paper vs reproduced” tables for methods like Wanda, SparseGPT, Sheared LLaMA and LLM-Pruner so engineers can verify their runs against known baselines.

Check out the GitHub Repo. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
The post LLM-Pruning Collection: A JAX Based Repo For Structured And Unstructured LLM Compression appeared first on MarkTechPost.

In this tutorial, we build an advanced multi-agent incident response system using AgentScope. We orchestrate multiple ReAct agents, each with a clearly defined role such as routing, triage, analysis, writing, and review, and connect them through structured routing and a shared message hub. By integrating OpenAI models, lightweight tool calling, and a simple internal runbook, we demonstrate how complex, real-world agentic workflows can be composed in pure Python without heavy infrastructure or brittle glue code. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browser!pip -q install “agentscope>=0.1.5” pydantic nest_asyncio

import os, json, re
from getpass import getpass
from typing import Literal
from pydantic import BaseModel, Field
import nest_asyncio
nest_asyncio.apply()

from agentscope.agent import ReActAgent
from agentscope.message import Msg, TextBlock
from agentscope.model import OpenAIChatModel
from agentscope.formatter import OpenAIChatFormatter
from agentscope.memory import InMemoryMemory
from agentscope.tool import Toolkit, ToolResponse, execute_python_code
from agentscope.pipeline import MsgHub, sequential_pipeline

if not os.environ.get(“OPENAI_API_KEY”):
os.environ[“OPENAI_API_KEY”] = getpass(“Enter OPENAI_API_KEY (hidden): “)

OPENAI_MODEL = os.environ.get(“OPENAI_MODEL”, “gpt-4o-mini”)

We set up the execution environment and install all required dependencies so the tutorial runs reliably on Google Colab. We securely load the OpenAI API key and initialize the core AgentScope components that will be shared across all agents. Check out the FULL CODES here.

Copy CodeCopiedUse a different BrowserRUNBOOK = [
{“id”: “P0”, “title”: “Severity Policy”, “text”: “P0 critical outage, P1 major degradation, P2 minor issue”},
{“id”: “IR1”, “title”: “Incident Triage Checklist”, “text”: “Assess blast radius, timeline, deployments, errors, mitigation”},
{“id”: “SEC7”, “title”: “Phishing Escalation”, “text”: “Disable account, reset sessions, block sender, preserve evidence”},
]

def _score(q, d):
q = set(re.findall(r”[a-z0-9]+”, q.lower()))
d = re.findall(r”[a-z0-9]+”, d.lower())
return sum(1 for w in d if w in q) / max(1, len(d))

async def search_runbook(query: str, top_k: int = 2) -> ToolResponse:
ranked = sorted(RUNBOOK, key=lambda r: _score(query, r[“title”] + r[“text”]), reverse=True)[: max(1, int(top_k))]
text = “nn”.join(f”[{r[‘id’]}] {r[‘title’]}n{r[‘text’]}” for r in ranked)
return ToolResponse(content=[TextBlock(type=”text”, text=text)])

toolkit = Toolkit()
toolkit.register_tool_function(search_runbook)
toolkit.register_tool_function(execute_python_code)

We define a lightweight internal runbook and implement a simple relevance-based search tool over it. We register this function along with a Python execution tool, enabling agents to retrieve policy knowledge or compute results dynamically. It demonstrates how we augment agents with external capabilities beyond pure language reasoning. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browserdef make_model():
return OpenAIChatModel(
model_name=OPENAI_MODEL,
api_key=os.environ[“OPENAI_API_KEY”],
generate_kwargs={“temperature”: 0.2},
)

class Route(BaseModel):
lane: Literal[“triage”, “analysis”, “report”, “unknown”] = Field(…)
goal: str = Field(…)

router = ReActAgent(
name=”Router”,
sys_prompt=”Route the request to triage, analysis, or report and output structured JSON only.”,
model=make_model(),
formatter=OpenAIChatFormatter(),
memory=InMemoryMemory(),
)

triager = ReActAgent(
name=”Triager”,
sys_prompt=”Classify severity and immediate actions using runbook search when useful.”,
model=make_model(),
formatter=OpenAIChatFormatter(),
memory=InMemoryMemory(),
toolkit=toolkit,
)

analyst = ReActAgent(
name=”Analyst”,
sys_prompt=”Analyze logs and compute summaries using python tool when helpful.”,
model=make_model(),
formatter=OpenAIChatFormatter(),
memory=InMemoryMemory(),
toolkit=toolkit,
)

writer = ReActAgent(
name=”Writer”,
sys_prompt=”Write a concise incident report with clear structure.”,
model=make_model(),
formatter=OpenAIChatFormatter(),
memory=InMemoryMemory(),
)

reviewer = ReActAgent(
name=”Reviewer”,
sys_prompt=”Critique and improve the report with concrete fixes.”,
model=make_model(),
formatter=OpenAIChatFormatter(),
memory=InMemoryMemory(),
)

We construct multiple specialized ReAct agents and a structured router that decides how each user request should be handled. We assign clear responsibilities to the triage, analysis, writing, and review agents, ensuring separation of concerns. Check out the FULL CODES here.

Copy CodeCopiedUse a different BrowserLOGS = “””timestamp,service,status,latency_ms,error
2025-12-18T12:00:00Z,checkout,200,180,false
2025-12-18T12:00:05Z,checkout,500,900,true
2025-12-18T12:00:10Z,auth,200,120,false
2025-12-18T12:00:12Z,checkout,502,1100,true
2025-12-18T12:00:20Z,search,200,140,false
2025-12-18T12:00:25Z,checkout,500,950,true
“””

def msg_text(m: Msg) -> str:
blocks = m.get_content_blocks(“text”)
if blocks is None:
return “”
if isinstance(blocks, str):
return blocks
if isinstance(blocks, list):
return “n”.join(str(x) for x in blocks)
return str(blocks)

We introduce sample log data and a utility function that normalizes agent outputs into clean text. We ensure that downstream agents can safely consume and refine earlier responses without format issues. It focuses on making inter-agent communication robust and predictable. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browserasync def run_demo(user_request: str):
route_msg = await router(Msg(“user”, user_request, “user”), structured_model=Route)
lane = (route_msg.metadata or {}).get(“lane”, “unknown”)

if lane == “triage”:
first = await triager(Msg(“user”, user_request, “user”))
elif lane == “analysis”:
first = await analyst(Msg(“user”, user_request + “nnLogs:n” + LOGS, “user”))
elif lane == “report”:
draft = await writer(Msg(“user”, user_request, “user”))
first = await reviewer(Msg(“user”, “Review and improve:nn” + msg_text(draft), “user”))
else:
first = Msg(“system”, “Could not route request.”, “system”)

async with MsgHub(
participants=[triager, analyst, writer, reviewer],
announcement=Msg(“Host”, “Refine the final answer collaboratively.”, “assistant”),
):
await sequential_pipeline([triager, analyst, writer, reviewer])

return {“route”: route_msg.metadata, “initial_output”: msg_text(first)}

result = await run_demo(
“We see repeated 5xx errors in checkout. Classify severity, analyze logs, and produce an incident report.”
)
print(json.dumps(result, indent=2))

We orchestrate the full workflow by routing the request, executing the appropriate agent, and running a collaborative refinement loop using a message hub. We coordinate multiple agents in sequence to improve the final output before returning it to the user. It brings together all earlier components into a cohesive, end-to-end agentic pipeline.

In conclusion, we showed how AgentScope enables us to design robust, modular, and collaborative agent systems that go beyond single-prompt interactions. We routed tasks dynamically, invoked tools only when needed, and refined outputs through multi-agent coordination, all within a clean and reproducible Colab setup. This pattern illustrates how we can scale from simple agent experiments to production-style reasoning pipelines while maintaining clarity, control, and extensibility in our agentic AI applications.

Check out the FULL CODES here. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
The post A Coding Guide to Design and Orchestrate Advanced ReAct-Based Multi-Agent Workflows with AgentScope and OpenAI appeared first on MarkTechPost.

In this tutorial, we build an advanced yet practical multi-agent system using OpenAI Swarm that runs in Colab. We demonstrate how we can orchestrate specialized agents, such as a triage agent, an SRE agent, a communications agent, and a critic, to collaboratively handle a real-world production incident scenario. By structuring agent handoffs, integrating lightweight tools for knowledge retrieval and decision ranking, and keeping the implementation clean and modular, we show how Swarm enables us to design controllable, agentic workflows without heavy frameworks or complex infrastructure. Check out the FULL CODES HERE.

Copy CodeCopiedUse a different Browser!pip -q install -U openai
!pip -q install -U “git+https://github.com/openai/swarm.git”

import os

def load_openai_key():
try:
from google.colab import userdata
key = userdata.get(“OPENAI_API_KEY”)
except Exception:
key = None
if not key:
import getpass
key = getpass.getpass(“Enter OPENAI_API_KEY (hidden): “).strip()
if not key:
raise RuntimeError(“OPENAI_API_KEY not provided”)
return key

os.environ[“OPENAI_API_KEY”] = load_openai_key()

We set up the environment and securely load the OpenAI API key so the notebook can run safely in Google Colab. We ensure the key is fetched from Colab secrets when available and fall back to a hidden prompt otherwise. This keeps authentication simple and reusable across sessions. Check out the FULL CODES HERE.

Copy CodeCopiedUse a different Browserimport json
import re
from typing import List, Dict
from swarm import Swarm, Agent

client = Swarm()

We import the core Python utilities and initialize the Swarm client that orchestrates all agent interactions. This snippet establishes the runtime backbone that allows agents to communicate, hand off tasks, and execute tool calls. It serves as the entry point for the multi-agent workflow. Check out the FULL CODES HERE.

Copy CodeCopiedUse a different BrowserKB_DOCS = [
{
“id”: “kb-incident-001”,
“title”: “API Latency Incident Playbook”,
“text”: “If p95 latency spikes, validate deploys, dependencies, and error rates. Rollback, cache, rate-limit, scale. Compare p50 vs p99 and inspect upstream timeouts.”
},
{
“id”: “kb-risk-001”,
“title”: “Risk Communication Guidelines”,
“text”: “Updates must include impact, scope, mitigation, owner, and next update. Avoid blame and separate internal vs external messaging.”
},
{
“id”: “kb-ops-001”,
“title”: “On-call Handoff Template”,
“text”: “Include summary, timeline, current status, mitigations, open questions, next actions, and owners.”
},
]

def _normalize(s: str) -> List[str]:
return re.sub(r”[^a-z0-9s]”, ” “, s.lower()).split()

def search_kb(query: str, top_k: int = 3) -> str:
q = set(_normalize(query))
scored = []
for d in KB_DOCS:
score = len(q.intersection(set(_normalize(d[“title”] + ” ” + d[“text”]))))
scored.append((score, d))
scored.sort(key=lambda x: x[0], reverse=True)
docs = [d for s, d in scored[:top_k] if s > 0] or [scored[0][1]]
return json.dumps(docs, indent=2)

We define a lightweight internal knowledge base and implement a retrieval function to surface relevant context during agent reasoning. By using simple token-based matching, we allow agents to ground their responses in predefined operational documents. This demonstrates how Swarm can be augmented with domain-specific memory without external dependencies. Check out the FULL CODES HERE.

Copy CodeCopiedUse a different Browserdef estimate_mitigation_impact(options_json: str) -> str:
try:
options = json.loads(options_json)
except Exception as e:
return json.dumps({“error”: str(e)})
ranking = []
for o in options:
conf = float(o.get(“confidence”, 0.5))
risk = o.get(“risk”, “medium”)
penalty = {“low”: 0.1, “medium”: 0.25, “high”: 0.45}.get(risk, 0.25)
ranking.append({
“option”: o.get(“option”),
“confidence”: conf,
“risk”: risk,
“score”: round(conf – penalty, 3)
})
ranking.sort(key=lambda x: x[“score”], reverse=True)
return json.dumps(ranking, indent=2)

We introduce a structured tool that evaluates and ranks mitigation strategies based on confidence and risk. This allows agents to move beyond free-form reasoning and produce semi-quantitative decisions. We show how tools can enforce consistency and decision discipline in agent outputs. Check out the FULL CODES HERE.

Copy CodeCopiedUse a different Browserdef handoff_to_sre():
return sre_agent

def handoff_to_comms():
return comms_agent

def handoff_to_handoff_writer():
return handoff_writer_agent

def handoff_to_critic():
return critic_agent

We define explicit handoff functions that enable one agent to transfer control to another. This snippet illustrates how we model delegation and specialization within Swarm. It makes agent-to-agent routing transparent and easy to extend. Check out the FULL CODES HERE.

Copy CodeCopiedUse a different Browsertriage_agent = Agent(
name=”Triage”,
model=”gpt-4o-mini”,
instructions=”””
Decide which agent should handle the request.
Use SRE for incident response.
Use Comms for customer or executive messaging.
Use HandoffWriter for on-call notes.
Use Critic for review or improvement.
“””,
functions=[search_kb, handoff_to_sre, handoff_to_comms, handoff_to_handoff_writer, handoff_to_critic]
)

sre_agent = Agent(
name=”SRE”,
model=”gpt-4o-mini”,
instructions=”””
Produce a structured incident response with triage steps,
ranked mitigations, ranked hypotheses, and a 30-minute plan.
“””,
functions=[search_kb, estimate_mitigation_impact]
)

comms_agent = Agent(
name=”Comms”,
model=”gpt-4o-mini”,
instructions=”””
Produce an external customer update and an internal technical update.
“””,
functions=[search_kb]
)

handoff_writer_agent = Agent(
name=”HandoffWriter”,
model=”gpt-4o-mini”,
instructions=”””
Produce a clean on-call handoff document with standard headings.
“””,
functions=[search_kb]
)

critic_agent = Agent(
name=”Critic”,
model=”gpt-4o-mini”,
instructions=”””
Critique the previous answer, then produce a refined final version and a checklist.
“””
)

We configure multiple specialized agents, each with a clearly scoped responsibility and instruction set. By separating triage, incident response, communications, handoff writing, and critique, we demonstrate a clean division of labor. Check out the FULL CODES HERE.

Copy CodeCopiedUse a different Browserdef run_pipeline(user_request: str):
messages = [{“role”: “user”, “content”: user_request}]
r1 = client.run(agent=triage_agent, messages=messages, max_turns=8)
messages2 = r1.messages + [{“role”: “user”, “content”: “Review and improve the last answer”}]
r2 = client.run(agent=critic_agent, messages=messages2, max_turns=4)
return r2.messages[-1][“content”]

request = “””
Production p95 latency jumped from 250ms to 2.5s after a deploy.
Errors slightly increased, DB CPU stable, upstream timeouts rising.
Provide a 30-minute action plan and a customer update.
“””

print(run_pipeline(request))

We assemble the full orchestration pipeline that executes triage, specialist reasoning, and critical refinement in sequence. This snippet shows how we run the end-to-end workflow with a single function call. It ties together all agents and tools into a coherent, production-style agentic system.

In conclusion, we established a clear pattern for designing agent-oriented systems with OpenAI Swarm that emphasizes clarity, separation of responsibilities, and iterative refinement. We showed how to route tasks intelligently, enrich agent reasoning with local tools, and improve output quality via a critic loop, all while maintaining a simple, Colab-friendly setup. This approach allows us to scale from experimentation to real operational use cases, making Swarm a powerful foundation for building reliable, production-grade agentic AI workflows.

Check out the FULL CODES HERE. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
The post How to Build a Production-Ready Multi-Agent Incident Response System Using OpenAI Swarm and Tool-Augmented Agents appeared first on MarkTechPost.