The web is full of articles that do not want to tell you what happened too soon. The headline hints at something. The first paragraphs add suspense. The useful information is somewhere below the fold, after the cookie banner, the newsletter box, a couple of related links, and enough scrolling to make the advertising model happy.
That is annoying when all we want is the news.
That’s my PoC. A small command-line application that receives the URL of a news article, converts the page into clean Markdown, and asks an AI agent to rewrite it as clear journalism: direct headline, concise lead, short paragraphs, no clickbait.
The CLI does not scrape the page directly. It gives the URL to a Strands Agent. The agent has one tool, fetch_url_as_markdown, and the model decides when to use it. Once the article is available as Markdown, the agent rewrites it following a focused system prompt.
The architecture
The flow is straightforward:
The important part is the boundary between the agent and the tool. Fetching a web page, removing navigation, and converting HTML into Markdown is deterministic Python code. Deciding how to rewrite the story is the LLM’s job.
This keeps the PoC small and easy to reason about.
Project structure
I like to keep configuration in settings.py. It is a pattern I borrowed years ago from Django and I still use it in small prototypes because it keeps things simple:
src/commands/rewrite.py contains the Click command.
src/lib/tools.py contains the Strands tool and the HTML-to-Markdown pipeline.
src/lib/agent.py wires Strands Agents with AWS Bedrock.
src/lib/prompts.py keeps the editor prompt and the user task prompt.
src/lib/ui.py renders Markdown in the terminal with Rich.
Fetching a URL as Markdown
The agent only gets one tool. It fetches the URL, removes noisy page elements, selects the main content, converts it to Markdown, and truncates the result to 100K characters:
@tool
deffetch_url_as_markdown(url: str) -> str:
"""
Fetch an HTTP or HTTPS URL, remove navigation, ads, scripts and layout noise,
extract the main article content, convert it to Markdown, and return up to
100K characters of clean text.
Use this tool when the user pastes a URL or asks you to analyze a web page.
I am not trying to build a perfect browser engine here. This is a PoC. The goal is to get enough readable article content for the agent to work with. For many news pages, removing scripts, navigation, cookie boxes, newsletter blocks, related links and advertising containers is enough.
The agent
The agent uses Claude on AWS Bedrock through Strands Agents:
defcreate_agent(*,settings: Settings) -> Agent:
boto_session=create_boto_session(settings)
returnAgent(
model=BedrockModel(
boto_session=boto_session,
model_id=settings.resolved_bedrock_model_id,
),
tools=[fetch_url_as_markdown],
system_prompt=SYSTEM_PROMPT,
)
The system prompt is the editorial policy. It tells the model to preserve only facts supported by the fetched article, answer in the requested output language, put the most important information first, remove suspense and filler, and write in a neutral tone.
The output format is Markdown:
a direct H1 headline
a concise lead paragraph
short factual paragraphs
a final What changed section, translated to the requested output language, explaining
what noise was removed
That last section is useful during development. It gives us a quick sanity check: did the model actually remove clickbait, or did it just paraphrase the article?
The CLI
The command is intentionally small:
@click.command(name="rewrite")
@click.argument("url")
@runtime_options
defrewrite_command(
url: str,
aws_profile: str|None,
region: str|None,
model: str|None,
language: str,
) -> None:
ifnotis_supported_url(url):
raiseclick.ClickException("URL must start with http:// or https://")
The CLI validates the URL, creates the agent, sends the URL in the prompt, and renders the final Markdown with Rich.
The tool is not called manually from the command. That is the point of this PoC: the URL is part of the task, and the agent decides to call fetch_url_as_markdown because the tool description says it should be used when the user pastes a URL or asks to analyze a web page.
Usage
Run the command:
poetry run plainnews rewrite "https://example.com/news/article"
By default, PlainNews writes the rewritten article in English. You can choose a
different output language with --language:
poetry run plainnews rewrite "https://example.com/news/article" --language Spanish
The output is rendered as Markdown in the terminal.
Example terminal output:
Tech stack
Python with Poetry
Strands Agents for tool-based agent orchestration
AWS Bedrock for the LLM runtime
BeautifulSoup for HTML cleanup
markdownify for HTML-to-Markdown conversion
Click for the command-line interface
Rich for Markdown terminal rendering
pytest for tests
A couple of notes
This is not a product and it is not a universal paywall remover. It is a small agentic workflow for a very specific frustration: articles that make readers work too hard to understand the basic facts.
Even in this small version, the pattern is useful: deterministic Python code prepares clean context, and the AI agent performs the editorial rewrite with a tight prompt.
And that’s all. Full source code available on GitHub.
What if you could describe the music you want to hear and have an AI produce it in real-time, sending MIDI notes directly to your DAW? That’s exactly what I built: a Python application that uses AI agents to generate Eurobeat and 90s techno patterns, outputting them as live MIDI to Akai’s MPC Beats.
I’m not a musician. I enjoy playing guitar from time to time, but I have zero experience with music production software. However, I’m gifted myself a Akai MPK mini Plus MIDI controller, which has 8 knobs and 8 pads, and I experimented with using it to control a music generation agent. No idea what I’m doing, but it’s fun.
As Akai MIDI controller can be connected to a laptop, and there I’ve got Python, this saturday morning I decided to build a simple prototype that connects an AI agent to MIDI output. The idea is simple. You write a prompt like “Energetic eurobeat in Am, Daft Punk style”, and an AI agent powered by Claude on AWS Bedrock generates patterns for 8 tracks: two drum kits, bass, rhodes, pluck, pad, and a lead melody. The patterns are sent as MIDI messages to MPC Beats, where each track is routed to a different virtual instrument. You can then modify the music live by writing new instructions, and use the physical knobs and pads on an Akai MPK Mini Plus to mute/unmute tracks, regenerate patterns, or reset the session.
I’m using the MPC Beats because it’s free and has a simple MIDI setup, but in theory this could work with any DAW that accepts MIDI input. The whole system is built in Python using Strands Agents for the AI orchestration, mido + python-rtmidi for MIDI I/O, and Rich for the terminal UI.
The Architecture
The flow is straightforward:
Project Structure
src/
settings.py # Configuration: BPM, tracks, MIDI devices
cli.py # Click CLI entry point
commands/play.py # Main play command
agent/
prompts.py # System prompts for the AI producer
tools.py # PatternStore + @tool functions
factory.py # Agent creation
midi/
device.py # MIDI device detection
melody_player.py # Threaded melody loop player
drum_player.py # Threaded drum loop player
session/
state.py # State machine (IDLE/GENERATING/PLAYING)
session.py # Session orchestrator
ui/
menu.py # Interactive terminal menu
Configuration
Everything starts with settings.py. The MIDI devices and AWS region are loaded from environment variables, while the musical parameters are defined as constants:
Each track maps to a MIDI channel. Tracks 1-2 are drum kits (offset-based timing), tracks 3-8 are melodic instruments (duration-based timing). The MPC Beats “House Template” provides the virtual instruments: a Classic drum kit, a Detroit percussion kit, Electric Rhodes, Tube Pluck, Bassline, Organ Bass, Tube Pad, and an Instant Go lead synth.
The Bridge Between AI and MIDI: PatternStore and Tools
The core of the system is the PatternStore, a simple shared store where the AI writes patterns and the MIDI players read them:
classPatternStore:
def__init__(self):
self._patterns: dict[int,list]={}
defset(self,track_id: int,pattern: list) -> None:
self._patterns[track_id]=pattern
defget(self,track_id: int) -> list|None:
returnself._patterns.get(track_id)
defclear(self) -> None:
self._patterns.clear()
The Strands @tool functions are created via a factory that closes over the store:
"""Define a drum pattern for a specific drum track."""
data=json.loads(pattern)
store.set(track_id,data)
name=TRACKS[track_id]["name"]
returnf"OK - {name}: {len(data)} hits"
return[set_drum_pattern,set_melody_pattern]
A melody pattern is a JSON array of {note, duration, velocity} objects where the sum of durations must equal LOOP_DURATION (4 bars). A drum pattern uses {note, velocity, offset} where offset is the time in seconds from the loop start. The note value -1 represents silence, which is crucial for creating space in the arrangement.
The Agent
The agent is a Strands Agent using Claude Sonnet on AWS Bedrock. The system prompt is heavily detailed with music production instructions: frequency ranges for each track, velocity guidelines, and structural rules. The key instruction is “less is more” – not all tracks should play notes all the time:
defcreate_agent(store: PatternStore) -> Agent:
returnAgent(
model=BedrockModel(
model_id=Models.CLAUDE_SONNET,
region_name=AWS_REGION,
),
tools=create_tools(store),
system_prompt=SYSTEM_PROMPT,
callback_handler=None,
)
There are two agents: one for initial generation (calls all 8 tools) and one for live modifications (only modifies the tracks that need to change). A third, lighter agent using Haiku generates the menu suggestions to keep latency and cost low.
MIDI Players
Two player classes handle the actual MIDI output. The MelodyLoopPlayer iterates through note events with durations:
The DrumLoopPlayer uses offset-based timing instead, scheduling hits at specific points within the loop. Both players read from the PatternStore on each loop iteration, which enables hot-swapping patterns during live modifications.
The Session
The Session class orchestrates everything. It manages the state machine (IDLE -> GENERATING -> PLAYING), owns the PatternStore, creates the agents, and handles MIDI input from the controller:
classSession:
def__init__(self):
self.state=State.IDLE
self.store=PatternStore()
self.agent=create_agent(self.store)
self.live_agent=create_live_agent(self.store)
self._agent_busy=threading.Lock()
When generation completes, playback starts with a progressive intro – tracks are unmuted one by one with a 2-bar delay between each, creating a build-up effect:
Predicting the future is something we all try to do. Whether it’s energy consumption, sensor readings, or production metrics, having a reliable forecast helps us make better decisions. The problem is that building a good forecasting model traditionally requires deep statistical knowledge, and a lot of tuning. What if we could just hand our data to an AI agent and ask “what’s going to happen next”?
That’s exactly what this project does. It combines Strands Agents with Amazon Chronos-Bolt, a foundation model for time series forecasting available on AWS Bedrock Marketplace, to create an AI agent that can forecast any numerical time series through natural language.
The architecture
The idea is simple. We have a Strands Agent powered by Claude (via AWS Bedrock) that understands natural language. When the user asks for a forecast, the agent calls a custom tool that invokes Chronos-Bolt to generate predictions. The agent then interprets the results and explains them in plain language.
The key here is that the agent doesn’t just return raw numbers. It understands the context, explains trends, and presents the confidence intervals in a way that makes sense.
The forecast tool
The tool is defined using the @tool decorator from Strands. This decorator turns a regular Python function into something the agent can discover and invoke on its own:
@tool
defforecast_time_series(
values: Annotated[
list[float],
"Historical time series values in chronological order. "
"Values should be evenly spaced (e.g., hourly, daily). Minimum 10 values.",
],
prediction_length: Annotated[
int,
"Number of future steps to predict. "
"Uses the same time unit as the input data.",
],
quantile_levels: Annotated[
Optional[list[float]],
"Quantile levels for confidence intervals. Default: [0.1, 0.5, 0.9]. "
"0.5 is the median forecast, 0.1 and 0.9 define the 80% confidence band.",
]=None,
) -> dict:
The Annotated type hints serve a dual purpose: they validate types at runtime and provide descriptions that the LLM reads to understand how to use the tool. This means the agent knows it needs a list of floats, a prediction length, and optionally custom quantile levels, all from the type annotations alone.
The tool validates the input (minimum 10 values, maximum 50,000, prediction length between 1 and 1,000), filters out NaN values, and then calls the Chronos-Bolt client:
result=invoke_chronos(
values=clean_values,
prediction_length=prediction_length,
quantile_levels=quantile_levels,
)
return{
"status": "success",
"content": [{"text": "\n".join(summary_lines)}],
"metadata": {
"quantiles": result.quantiles,
"prediction_length": result.prediction_length,
"history_length": result.history_length,
},
}
The response includes both a human-readable summary (in content) and the raw quantile data (in metadata), so the agent can reference exact numbers when explaining the forecast.
The Chronos-Bolt client
Chronos-Bolt is accessed through the Bedrock runtime API. The client sends the historical values and receives predictions at different quantile levels:
definvoke_chronos(
values: list[float],
prediction_length: int,
quantile_levels: list[float]|None=None,
) -> ForecastResult:
client=_get_bedrock_runtime_client()
payload={
"inputs": [{"target": values}],
"parameters": {
"prediction_length": prediction_length,
"quantile_levels": quantiles,
},
}
response=client.invoke_model(
modelId=CHRONOS_ENDPOINT_ARN,
body=json.dumps(payload),
contentType="application/json",
accept="application/json",
)
The invoke_model call uses the SageMaker endpoint ARN deployed through Bedrock Marketplace. Chronos-Bolt returns predictions organized by quantile levels, by default, the 10th, 50th (median), and 90th percentiles. This gives us not just a single forecast line, but a confidence band: the 80% interval between the 10th and 90th percentiles tells us how uncertain the model is about its predictions.
The Bedrock runtime client is configured with generous timeouts (120s read, 30s connect) and automatic retries, since inference on time series data can take a moment depending on the history length:
def_get_bedrock_runtime_client():
returnboto3.client(
"bedrock-runtime",
region_name=AWS_REGION,
config=Config(
read_timeout=120,
connect_timeout=30,
retries={"max_attempts": 3},
),
)
The agent
Wiring everything together is straightforward. We create a BedrockModel pointing to Claude and pass our forecast tool to the Agent:
fromstrandsimportAgent
fromstrands.models.bedrockimportBedrockModel
fromsettingsimportAWS_REGION,Models
fromforecastimportforecast_time_series
SYSTEM_PROMPT="""You are a time series forecasting assistant powered by Amazon Chronos-Bolt.
You help users predict future values from historical numerical data. When a user provides
time series data or describes a scenario, use the forecast_time_series tool to generate
predictions.
When presenting results:
- Show the median forecast (quantile 0.5) as the main prediction
- Explain the confidence band (quantiles 0.1 and 0.9) as the uncertainty range
- Summarize trends in plain language
"""
defcreate_agent() -> Agent:
bedrock_model=BedrockModel(
model_id=Models.CLAUDE_SONNET,
region_name=AWS_REGION,
)
returnAgent(
model=bedrock_model,
system_prompt=SYSTEM_PROMPT,
tools=[forecast_time_series],
)
The system prompt is important here. It tells Claude that it has forecasting capabilities and how to present the results. Without it, the agent would still call the tool correctly (thanks to the Annotated descriptions), but it might not explain the confidence bands or summarize trends as clearly.
Running it
The CLI entry point (cli.py) registers commands and wires everything together. The forecast command generates synthetic hourly data (a sine wave with noise) by default and asks the agent to forecast. You can also pass a custom prompt.
The entry point is minimal:
importclick
fromcommands.forecastimportrunasforecast
@click.group()
defcli():
pass
cli.add_command(cmd=forecast,name="forecast")
if__name__=="__main__":
cli()
The actual command lives in commands/forecast.py:
@click.command()
@click.option("--prompt","-p",default=None,help="Custom prompt for the agent.")
defrun(prompt: str|None):
agent=create_agent()
ifpromptisNone:
values=generate_sample_data(num_points=100)
values_str=", ".join(f"{v:.2f}"forvinvalues)
prompt=(
f"I have the following hourly sensor readings from the last 100 hours:\n"
f"[{values_str}]\n\n"
f"Please forecast the next 24 hours and explain the predicted trend."
)
response=agent(prompt)
click.echo(response)
The sine wave is a good choice for a demo because it has a clear periodic pattern that Chronos-Bolt should capture well. With 100 hours of history (about 4 full cycles of a 24-hour pattern), the model has enough data to identify the periodicity and project it forward.
Example
(venv) ➜ src python cli.py forecast
2026-02-2714:11:16,471 - INFO - Found credentials in shared credentials file: ~/.aws/credentials
2026-02-2714:11:16,506 - INFO - Creating Strands MetricsClient
Sure! Let me run the forecast on your 100-hour sensor readings right away.
Tool #1: forecast_time_series
2026-02-2714:11:22,981 - INFO - Starting forecast: history=100, prediction_length=24
2026-02-2714:11:22,994 - INFO - Found credentials in shared credentials file: ~/.aws/credentials
2026-02-2714:11:23,697 - INFO - Forecast complete: 24 steps, 4 quantiles
Here are the results of your 24-hour forecast:
Script response
📊 24-Hour Sensor Forecast
Hour
Low (10th %)
Median (50th %)
High (90th %)
+1
~-0.29
~0.85
~0.99
+2
~0.29
~0.95
~0.99
+3
~0.49
~0.99
~0.99
+4
~0.47
~0.83
~0.99
+5
~0.24
~0.72
~0.96
+6
~0.09
~0.53
~0.83
+7
~-0.25
~0.19
~0.59
+8
~-0.60
~-0.22
~0.25
+9
~-0.91
~-0.51
~-0.04
+10
~-0.99
~-0.76
~-0.30
+11
~-0.99
~-0.87
~-0.46
+12
~-0.99
~-0.93
~-0.56
+13
~-0.99
~-0.96
~-0.57
+14
~-0.99
~-0.93
~-0.52
+15
~-0.99
~-0.87
~-0.44
+16
~-0.99
~-0.79
~-0.35
+17
~-0.96
~-0.63
~-0.16
+18
~-0.86
~-0.46
~0.02
+19
~-0.66
~-0.24
~0.21
+20
~-0.44
~-0.01
~0.42
+21
~-0.21
~0.22
~0.60
+22
~-0.04
~0.45
~0.78
+23
~0.16
~0.66
~0.90
+24
~0.31
~0.80
~0.96
🔍 Trend Explanation
Your sensor data clearly exhibits a sinusoidal/oscillating pattern — cycling between peaks near +1.0 and troughs near -1.0 roughly every ~24 hours. The model has picked this up beautifully! Here’s what to expect:
Hours +1 to +4 — Continued Peak 🔼: The signal is currently rising (ending at ~0.67–0.70), and the model predicts it will crest near ~+0.85 to +0.99 in the first few hours — completing the current positive phase.
Hours +5 to +13 — Descent & Trough 🔽: The signal then descends sharply, hitting a trough around hours +12 to +14 with a median near -0.93 to -0.96 — matching the negative peaks seen in the historical data.
Hours +14 to +24 — Recovery 🔼: After bottoming out, the signal climbs back up, reaching approximately +0.80 by hour +24, setting up the next positive cycle.
📐 Confidence Band
The 80% confidence interval (low–high columns) is relatively tight, reflecting the model’s high confidence in the periodic nature of this signal. The widest uncertainty occurs around the transition zones (hours +7–+9 and +17–+19), which is typical for oscillating signals near the zero-crossing points.
In short: your sensor is behaving like a clean oscillating signal with an ~24-hour period, and the next full cycle looks very consistent with historical behavior.Here are the results of your 24-hour forecast:
📊 24-Hour Sensor Forecast
Hour
Low (10th %)
Median (50th %)
High (90th %)
+1
~-0.29
~0.85
~0.99
+2
~0.29
~0.95
~0.99
+3
~0.49
~0.99
~0.99
+4
~0.47
~0.83
~0.99
+5
~0.24
~0.72
~0.96
+6
~0.09
~0.53
~0.83
+7
~-0.25
~0.19
~0.59
+8
~-0.60
~-0.22
~0.25
+9
~-0.91
~-0.51
~-0.04
+10
~-0.99
~-0.76
~-0.30
+11
~-0.99
~-0.87
~-0.46
+12
~-0.99
~-0.93
~-0.56
+13
~-0.99
~-0.96
~-0.57
+14
~-0.99
~-0.93
~-0.52
+15
~-0.99
~-0.87
~-0.44
+16
~-0.99
~-0.79
~-0.35
+17
~-0.96
~-0.63
~-0.16
+18
~-0.86
~-0.46
~0.02
+19
~-0.66
~-0.24
~0.21
+20
~-0.44
~-0.01
~0.42
+21
~-0.21
~0.22
~0.60
+22
~-0.04
~0.45
~0.78
+23
~0.16
~0.66
~0.90
+24
~0.31
~0.80
~0.96
🔍 Trend Explanation
Your sensor data clearly exhibits a sinusoidal/oscillating pattern — cycling between peaks near +1.0 and troughs near -1.0 roughly every ~24 hours. The model has picked this up beautifully! Here’s what to expect:
Hours +1 to +4 — Continued Peak 🔼: The signal is currently rising (ending at ~0.67–0.70), and the model predicts it will crest near ~+0.85 to +0.99 in the first few hours — completing the current positive phase.
Hours +5 to +13 — Descent & Trough 🔽: The signal then descends sharply, hitting a trough around hours +12 to +14 with a median near -0.93 to -0.96 — matching the negative peaks seen in the historical data.
Hours +14 to +24 — Recovery 🔼: After bottoming out, the signal climbs back up, reaching approximately +0.80 by hour +24, setting up the next positive cycle.
📐 Confidence Band
The 80% confidence interval (low–high columns) is relatively tight, reflecting the model’s high confidence in the periodic nature of this signal. The widest uncertainty occurs around the transition zones (hours +7–+9 and +17–+19), which is typical for oscillating signals near the zero-crossing points.
In short: your sensor is behaving like a clean oscillating signal with an ~24-hour period, and the next full cycle looks very consistent with historical behavior.
We live in the era of Large Language Models (LLMs) with massive context windows. Claude 3.5 Sonnet offers 200k tokens, and Gemini 1.5 Pro goes up to 2 million. So, why do we still need to worry about document processing strategies? The answer is yes, we do. For example, AWS Bedrock has a strict limit of 4.5MB for documents, regardless of token count. That’s means we can’t just stuff file greater than 4.5MB into a prompt. Today we’ll show you how I built a production-ready document processing agent that handles large files by implementing a Map-Reduce pattern using Python, AWS Bedrock, and Strands Agents.
The core idea is simple: instead of asking the LLM to “read this book and answer” we break the book into chapters, analyze each chapter in parallel, and then synthesize the results.
Here is the high-level flow:
The heart of the implementation is the DocumentProcessor class. It decides whether to process a file as a whole or split it based on a size threshold. We define a threshold (e.g., 4.3MB) to stay safely within Bedrock’s limits. If the file is larger, we trigger the _process_big method.
# src/lib/processor/processor.py
BYTES_THRESHOLD = 4_300_000
async def _process_file(self, file: DocumentFile, question: str, with_callback=True):
file_bytes = Path(file.path).read_bytes()
# Strategy pattern: Choose the right processor based on file size
processor = self._process_big if len(file_bytes) > BYTES_THRESHOLD else self._process
async for chunk in processor(file_bytes, file, question, with_callback):
yield chunk
To increase the performance, we use asyncio to process the file in parallel and we use a semaphore to control the number of workers.
async def _process_big(self, file_bytes: bytes, file: DocumentFile, question: str, with_callback=True) -> AsyncIterator[str]:
# ... splitting logic ...
semaphore = asyncio.Semaphore(self.max_workers)
# Create async tasks for each chunk
tasks = [
self._process_chunk(chunk, i, file_name, question, handler.format, semaphore)
for i, chunk in enumerate(chunks, 1)
]
# Run in parallel
results = await asyncio.gather(*tasks)
# Sort results to maintain document order
results.sort(key=lambda x: x[0])
responses_from_chunks = [response for _, response in results]
Each chunk is processed by an isolated agent instance that only sees that specific fragment and the user’s question. Once we have the partial analyses, we consolidate them. This acts as a compression step: we’ve turned raw pages into relevant insights.
def _consolidate_and_truncate(self, responses: list[str], num_chunks: int) -> str:
consolidated = "\n\n".join(responses)
if len(consolidated) > MAX_CONTEXT_CHARS:
# Safety mechanism to ensure we don't overflow the final context
return consolidated[:MAX_CONTEXT_CHARS] + "\n... [TRUNCATED]"
return consolidated
Finally, we feed this consolidated context to the agent for the final answer. In a long-running async process, feedback is critical. I implemented an Observer pattern to decouple the processing logic from the UI/Logging.
By breaking down large tasks, we not only bypass technical limits but often get better results. The model focuses on smaller sections, reducing hallucinations, and the final answer is grounded in a pre-processed summary of facts.
We don’t just send text; we send the raw document bytes. This allows the model (Claude 4.5 Sonnet via Bedrock) to use its native document processing capabilities. Here is how we construct the message payload:
When processing chunks, we don’t want the model to be chatty. We need raw information extraction. We use a “Spartan” system prompt that enforces brevity and objectivity, ensuring the consolidation phase receives high-signal input.
# src/lib/processor/prompts.py
SYSTEM_CHUNK_PROMPT = f"""
You are an artificial intelligence assistant specialized in reading and analyzing files.
You have received a chunk of a large file.
...
If the user's question cannot be answered with the information in the current chunk, do not answer it directly.
{SYSTEM_PROMPT_SPARTAN}
The SYSTEM_PROMPT_SPARTAN (injected above) explicitly forbids conversational filler, ensuring we maximize the token budget for actual data.
The project handles pdf and xlsx files. The rest of the file types are not processed and are given to the LLM as-is.
With this architecture, we can process large files in a production environment. This allows us to easily plug in different interfaces, whether it’s a CLI logger (as shown) or a WebSocket update for a UI frontend like Chainlit.
We all know LLMs are powerful, but their true potential is unlocked when they can see your data. While RAG (Retrieval-Augmented Generation) is great for massive knowledge bases, sometimes you just want to drag and drop a file and ask questions about it.
Today we’ll build a “File-Aware” AI agent that can natively understand a wide range of document formats—from PDFs and Excel sheets to Word docs and Markdown files. We’ll use AWS Bedrock with Claude 4.5 Sonnet for the reasoning engine and Chainlit for the conversational UI.
The idea is straightforward: Upload a file, inject it into the model’s context, and let the LLM do the rest. No vector databases, no complex indexing pipelines—just direct context injection for immediate analysis.
The architecture is simple yet effective. We intercept file uploads in the UI, process them into a format the LLM understands, and pass them along with the user’s query.
AWS Bedrock with Claude 4.5 Sonnet for high-quality reasoning and large context windows.
Chainlit for a chat-like interface with native file upload support.
Python for the backend logic.
The core challenge is handling different file types and presenting them to the LLM. We support a variety of formats by mapping them to Bedrock’s expected input types.
To enable file uploads in Chainlit, you need to configure the [features.spontaneous_file_upload] section in your .chainlit/config.toml. This is where you define which MIME types are accepted.
The main agent loop handles the conversation. It checks for uploaded files, processes them, and constructs the message payload for the LLM. We also include robust error handling to manage context window limits gracefully.
def get_question_from_message(message: cl.Message):
content_blocks = None
if message.elements:
content_blocks = get_content_blocks_from_message(message)
if content_blocks:
content_blocks.append({"text": message.content or "Write a summary of the document"})
question = content_blocks
else:
question = message.content
return question
def get_content_blocks_from_message(message: cl.Message):
docs = [f for f in message.elements if f.type == "file" and f.mime in MIME_MAP]
content_blocks = []
for doc in docs:
file = Path(doc.path)
file_bytes = file.read_bytes()
shutil.rmtree(file.parent)
content_blocks.append({
"document": {
"name": sanitize_filename(doc.name),
"format": MIME_MAP[doc.mime],
"source": {"bytes": file_bytes}
}
})
return content_blocks
@cl.on_message
async def handle_message(message: cl.Message):
task = asyncio.create_task(process_user_task(
question=get_question_from_message(message),
debug=DEBUG))
cl.user_session.set("task", task)
try:
await task
except asyncio.CancelledError:
logger.info("User task was cancelled.")
This pattern allows for ad-hoc analysis. You don’t need to pre-ingest data. You can:
Analyze Financials: Upload an Excel sheet and ask for trends.
Review Contracts: Upload a PDF and ask for clause summaries.
Debug Code: Upload a source file and ask for a bug fix.
By leveraging the large context window of modern models like Claude 4.5 Sonnet, we can feed entire documents directly into the prompt, providing the model with full visibility without the information loss often associated with RAG chunking.
And that's all. With tools like Chainlit and powerful APIs like AWS Bedrock, we can create robust, multi-modal assistants that integrate seamlessly into our daily workflows.
We can build simple AI agents that handle specific tasks quite easily today. But what about building AI systems that can handle multiple domains effectively? One approach is to create a single monolithic agent that tries to do everything, but this quickly runs into problems of context pollution, maintenance complexity, and scaling limitations. In this article, we’ll show a production-ready pattern for building multi-purpose AI systems using an orchestrator architecture that coordinates domain-specific agents.
The idea is simple: Don’t build one agent to rule them all instead, create specialized agents that excel in their domains and coordinate them through an intelligent orchestrator. The solution is an orchestrator agent that routes requests to specialized sub-agents, each with focused expertise and dedicated tools. Think of it as a smart router that understands intent and delegates accordingly.
That’s the core of the Orchestrator Pattern for multi-agent systems:
Weather Agent: Expert in meteorological data and weather patterns. It uses external weather APIs to fetch historical and current weather data.
Logistics Agent: Specialist in supply chain and shipping operations. Fake logistics data is generated to simulate shipment tracking, route optimization, and delivery performance analysis.
Production Agent: Focused on manufacturing operations and production metrics. Also, fake production data is generated to analyze production KPIs.
AWS Bedrock with Claude 4.5 Sonnet for agent reasoning
Strands Agents framework for agent orchestration
Chainlit for the conversational UI
FastAPI for the async backend
PostgreSQL for storing conversation history and domain data
The orchestrator’s job is simple but critical: understand the user’s intent and route to the right specialist(s).
MAIN_SYSTEM_PROMPT = """You are an intelligent orchestrator agent
responsible for routing user requests to specialized sub-agents
based on their domain expertise.
## Available Specialized Agents
### 1. Production Agent
**Domain**: Manufacturing operations, production metrics, quality control
**Handles**: Production KPIs, machine performance, downtime analysis
### 2. Logistics Agent
**Domain**: Supply chain, shipping, transportation operations
**Handles**: Shipment tracking, route optimization, delivery performance
### 3. Weather Agent
**Domain**: Meteorological data and weather patterns
**Handles**: Historical weather, atmospheric conditions, climate trends
## Your Decision Process
1. Analyze the request for key terms and domains
2. Determine scope (single vs multi-domain)
3. Route to appropriate agent(s)
4. Synthesize results when multiple agents are involved
"""
The orchestrator receives specialized agents as tools:
def get_orchestrator_tools() -> List[Any]:
from tools.logistics.agent import logistics_assistant
from tools.production.agent import production_assistant
from tools.weather.agent import weather_assistant
tools = [
calculator,
think,
current_time,
AgentCoreCodeInterpreter(region=AWS_REGION).code_interpreter,
logistics_assistant, # Specialized agent as tool
production_assistant, # Specialized agent as tool
weather_assistant # Specialized agent as tool
]
return tools
Each specialized agent follows a consistent pattern. Here’s the weather agent:
@tool
@stream_to_step("weather_assistant")
async def weather_assistant(query: str):
"""
A research assistant specialized in weather topics with streaming support.
"""
try:
tools = [
calculator,
think,
current_time,
AgentCoreCodeInterpreter(region=AWS_REGION).code_interpreter
]
# Domain-specific tools
tools += WeatherTools(latitude=MY_LATITUDE, longitude=MY_LONGITUDE).get_tools()
research_agent = get_agent(
system_prompt=WEATHER_ASSISTANT_PROMPT,
tools=tools
)
async for token in research_agent.stream_async(query):
yield token
except Exception as e:
yield f"Error in research assistant: {str(e)}"
Each agent has access to domain-specific tools. For example, the weather agent uses external APIs:
class WeatherTools:
def __init__(self, latitude: float, longitude: float):
self.latitude = latitude
self.longitude = longitude
def get_tools(self) -> List[tool]:
@tool
def get_hourly_weather_data(from_date: date, to_date: date) -> MeteoData:
"""Get hourly weather data for a specific date range."""
url = (f"https://api.open-meteo.com/v1/forecast?"
f"latitude={self.latitude}&longitude={self.longitude}&"
f"hourly=temperature_2m,relative_humidity_2m...")
response = requests.get(url)
return parse_weather_response(response.json())
return [get_hourly_weather_data]
The logistics and production agents use synthetic data generators for demonstration:
User: “Analyze production efficiency trends and correlate with weather and logistics performance based in yesterday’s data.”
Flow:
Orchestrator coordinates all three agents
Production agent retrieves manufacturing KPIs
Weather agent provides environmental data
Logistics agent supplies delivery metrics
Orchestrator synthesizes multi-domain analysis
This architecture scales naturally in multiple dimensions. We can easily add new specialized agents without disrupting existing functionality. WE only need to create the new agent and register it as a tool with the orchestratortrator prompt with new domain description. That’s it.
The orchestrator pattern transforms multi-domain AI from a monolithic challenge into a composable architecture. Each agent focuses on what it does best, while the orchestrator provides intelligent coordination.
In industrial environments, data analysis is crucial for optimizing processes, detecting anomalies, and making informed
decisions. Manufacturing plants, energy systems, and industrial IoT generate massive amounts of data from sensors,
machines, and control systems. Traditionally, analyzing this data requires specialized knowledge in both industrial
processes and data science, creating a bottleneck for quick insights.
I’ve been exploring agentic AI frameworks lately, particularly for complex data analysis tasks. While working on
industrial data problems, I realized that combining the reasoning capabilities of Large Language Models with specialized
tools could create a powerful solution for industrial data analysis. This project demonstrates how to build a ReAct (
Reasoning and Acting) AI agent using LangGraph that can analyze manufacturing data, understand industrial processes, and
provide actionable insights.
The goal of this project is to create an AI agent that can analyze industrial datasets (manufacturing metrics, sensor
readings, process control data) and provide expert-level insights about production optimization, quality control, and
process efficiency. Using LangGraph’s ReAct agent framework with AWS Bedrock, the system can execute Python code
dynamically in a sandboxed environment, process large datasets, and reason about industrial contexts.
The dataset is a fake sample of industrial data with manufacturing metrics like temperature, speed, humidity, pressure,
operator experience, scrap rates, and unplanned stops. In fact, I’ve generated the dataset using chatgpt
This project uses several key components:
LangGraph ReAct Agent: For building the multi-tool AI agent with ReAct (Reasoning and Acting) patterns that can
dynamically choose tools and reason about results
AWS Bedrock: Claude Sonnet 4 as the underlying LLM for reasoning and code generation
Sandboxed Code Interpreter: Secure execution of Python code for data analysis using AWS Agent Core. One tool taken
from strands-agents-tools library.
Industrial Domain Expertise: Specialized system prompts with knowledge of manufacturing processes, quality
control, and industrial IoT
The agent has access to powerful tools:
Code Interpreter: Executes Python code safely in a sandboxed AWS environment using pandas, numpy, scipy, and other
scientific libraries
Data Processing: Handles large industrial datasets with memory-efficient strategies
Industrial Context: Understands manufacturing processes, sensor data, and quality metrics
The system uses AWS Agent Core’s sandboxed code interpreter, which means:
Python code is executed in an isolated environment
No risk to the host system
Access to scientific computing libraries (pandas, numpy, scipy)
Memory management for large datasets
The core of the system is surprisingly simple. The ReAct agent is built using LangGraph’s create_react_agent with
custom tools:
from langgraph.prebuilt import create_react_agent
from typing import List
import pandas as pd
from langchain_core.callbacks import BaseCallbackHandler
def analyze_df(df: pd.DataFrame, system_prompt: str, user_prompt: str,
callbacks: List[BaseCallbackHandler], streaming: bool = False):
code_interpreter_tools = CodeInterpreter()
tools = code_interpreter_tools.get_tools()
agent = create_react_agent(
model=get_llm(model=DEFAULT_MODEL, streaming=streaming,
budget_tokens=12288, callbacks=callbacks),
tools=tools,
prompt=system_prompt
)
agent_prompt = f"""
I have a DataFrame with the following data:
- Columns: {list(df.columns)}
- Shape: {df.shape}
- data: {df}
The output must be an executive summary with the key points.
The response must be only markdown, not plots.
"""
messages = [
("user", agent_prompt),
("user", user_prompt)
]
agent_input = {"messages": messages}
return agent. Invoke(agent_input)
The ReAct pattern (Reasoning and Acting) allows the agent to:
Reason about what analysis is needed
Act by calling the appropriate tools (in this case: code interpreter)
Observe the results of code execution
Re-reason and potentially call more tools if needed
This creates a dynamic loop where the agent can iteratively analyze data, examine results, and refine its approach –
much more powerful than a single code execution.
The magic happens in the system prompt, which provides the agent with industrial domain expertise:
SYSTEM_PROMPT = """
# Industrial Data Analysis Agent - System Prompt
You are an expert AI agent specialized in industrial data analysis and programming.
You excel at solving complex data problems in manufacturing, process control,
energy systems, and industrial IoT environments.
## Core Capabilities
- Execute Python code using pandas, numpy, scipy
- Handle large datasets with chunking strategies
- Process time-series data, sensor readings, production metrics
- Perform statistical analysis, anomaly detection, predictive modeling
## Industrial Domain Expertise
- Manufacturing processes and production optimization
- Process control systems (PID controllers, SCADA, DCS)
- Industrial IoT sensor data and telemetry
- Quality control and Six Sigma methodologies
- Energy consumption analysis and optimization
- Predictive maintenance and failure analysis
"""
The code interpreter tool is wrapped with safety validations:
def validate_code_ast(code: str) -> bool:
"""Validate Python code using AST to ensure safety."""
try:
ast.parse(code)
return True
except SyntaxError:
return False
@tool
def code_interpreter(code: str) -> str:
"""Executes Python code in a sandboxed environment."""
if not validate_code_ast(code):
raise UnsafeCodeError("Unsafe code or syntax errors.")
return code_tool(code_interpreter_input={
"action": {
"type": "executeCode",
"session_name": session_name,
"code": code,
"language": "python"
}
})
The system uses Claude Sonnet 4 through AWS Bedrock with optimized parameters for industrial analysis:
The project includes fake sample industrial data with manufacturing metrics:
- `machine_id`: Equipment identifier - `shift`: Production shift (A/M/N for morning/afternoon/night) - `temperature`, `speed`, `humidity`, `pressure`: Process parameters - `operator_experience`: Years of operator experience - `scrap_kg`: Quality metric (waste produced) - `unplanned_stop`: Equipment failure indicator
A typical analysis query might be: "Do temperature and speed setpoints vary across shifts?" The agent will stream the response as it generates it.
The agent will:
1. Load and examine the dataset structure 2. Generate appropriate Python code for analysis 3. Execute the code in a sandboxed environment 4. Provide insights about shift-based variations 5. Suggest process optimization recommendations
import logging
import pandas as pd
from langchain_core.callbacks import StreamingStdOutCallbackHandler
from modules.df_analyzer import analyze_df
from prompts import SYSTEM_PROMPT
logging.basicConfig(
format='%(asctime)s [%(levelname)s] %(message)s',
level='INFO',
datefmt='%d/%m/%Y %X')
logger = logging.getLogger(__name__)
class StreamingCallbackHandler(StreamingStdOutCallbackHandler):
def on_llm_new_token(self, token: str, **kwargs):
print(token, end='', flush=True)
df = pd.read_csv('fake_data.csv')
user_prompt = "Do temperature and speed setpoints vary across shifts?"
for chunk in analyze_df(
user_prompt=user_prompt,
df=df,
system_prompt=SYSTEM_PROMPT,
callbacks=[StreamingCallbackHandler()],
streaming=True):
logger.debug(chunk)
This project demonstrates the power of agentic AI for specialized domains. Instead of building custom analytics
dashboards or writing specific analysis scripts, we provide the agent with:
Domain Knowledge: Through specialized system prompts
Tools: Safe code execution capabilities
Context: The actual data to analyze
The agent can then:
Generate appropriate analysis code
Execute it safely
Interpret results with industrial context
Provide actionable recommendations
The result is a flexible system that can handle various industrial analysis tasks without pre-programmed solutions. The
agent reasons about the problem, writes the necessary code (sandboxed), and provides expert-level insights.
Today we’re going to build an AI agent that can predict the weather using Strands-Agents framework and Python. This project is designed to show how to integrate external data sources, advanced computational tools, and AI capabilities into a cohesive system. For this experiment we’re going to use Strands-Agents framework, which provides a robust foundation for building intelligent agents that can interact with various tools and APIs. Strands-Agents comes with built-in tools that allow us to create agents that can perform complex tasks by orchestrating multiple tools and APIs. For this project we’re going to use the following tools:
calculator: for performing mathematical and financial calculations.
think: for reflecting on data and generating ideas.
file_write: for saving results and analyses to files.
python_repl: for executing Python code and performing advanced analyses.
The last one is particularly useful for overcoming the limitations of large language models (LLMs) when it comes to deterministic calculations. By using a Python REPL, we can ensure that our agent can perform precise computations without relying solely on the LLM’s probabilistic outputs. We have Pandas and Scikit-learn for statistical analysis, which allows us to perform advanced data manipulation and machine learning tasks, and the agent will be able to use these libraries to analyze weather data and generate forecasts. Also, I’ve created a custom tool to fetch hourly weather data from the Open-Meteo API, which provides real-time weather information for specific locations.
import logging
from datetime import datetime, date
from typing import List
import requests
from strands import tool
from modules.weather.models import (
TemperatureReading, HumidityReading, ApparentTemperatureReading,
PrecipitationReading, EvapotranspirationReading, SurfacePressureReading, MeteoData)
logger = logging.getLogger(__name__)
class Tools:
def __init__(self, latitude: float, longitude: float):
self.latitude = latitude
self.longitude = longitude
def get_tools(self) -> List[tool]:
@tool
def get_hourly_weather_data(from_date: date, to_date: date) -> MeteoData:
"""
Get hourly weather data for a specific date range.
Notes:
- The response is a MeteoData object containing lists of readings for temperature, humidity,
apparent temperature, precipitation, evapotranspiration, and surface pressure.
- Each reading has a timestamp and a value.
Returns:
MeteoData: Object containing weather readings for the specified date range
"""
start_date = from_date.strftime('%Y-%m-%d')
end_date = to_date.strftime('%Y-%m-%d')
url = (f"https://api.open-meteo.com/v1/forecast?"
f"latitude={self.latitude}&"
f"longitude={self.longitude}&"
f"hourly=temperature_2m,relative_humidity_2m,apparent_temperature,precipitation,evapotranspiration,surface_pressure&"
f"start_date={start_date}&"
f"end_date={end_date}")
response = requests.get(url)
meteo = MeteoData(
temperature=[],
humidity=[],
apparent_temperature=[],
precipitation=[],
evapotranspiration=[],
surface_pressure=[]
)
data = response.json()
weather_data_time = data['hourly']['time']
logger.info(f"[get_hourly_weather_data] Fetched weather data from {start_date} to {end_date}. {len(weather_data_time)} records found.")
for iso in weather_data_time:
time = datetime.fromisoformat(iso)
meteo.temperature.append(TemperatureReading(
time=time,
value=data['hourly']['temperature_2m'][data['hourly']['time'].index(iso)]))
meteo.humidity.append(HumidityReading(
time=time,
value=data['hourly']['relative_humidity_2m'][data['hourly']['time'].index(iso)]))
meteo.apparent_temperature.append(ApparentTemperatureReading(
time=time,
value=data['hourly']['apparent_temperature'][data['hourly']['time'].index(iso)]))
meteo.precipitation.append(PrecipitationReading(
time=time,
value=data['hourly']['precipitation'][data['hourly']['time'].index(iso)]))
meteo.evapotranspiration.append(EvapotranspirationReading(
time=time,
value=data['hourly']['evapotranspiration'][data['hourly']['time'].index(iso)]))
meteo.surface_pressure.append(SurfacePressureReading(
time=time,
value=data['hourly']['surface_pressure'][data['hourly']['time'].index(iso)]))
return meteo
return [get_hourly_weather_data, ]
To allow the LLM to interact with this tool, we define a Pydantic model that describes the expected input and output formats. This ensures that the agent can correctly interpret the data it receives from the API and use it effectively in its analyses.
from datetime import datetime
from pydantic import BaseModel, Field
class TemperatureReading(BaseModel):
"""Temperature reading at 2 meters"""
time: datetime = Field(..., description="Timestamp")
value: float = Field(description="Temperature in °C")
class HumidityReading(BaseModel):
"""Relative humidity reading at 2 meters"""
time: datetime = Field(..., description="Timestamp")
value: int = Field(..., ge=0, le=100, description="Relative humidity in %")
class ApparentTemperatureReading(BaseModel):
"""Apparent temperature reading"""
time: datetime = Field(..., description="Timestamp")
value: float = Field(..., description="Apparent temperature in °C")
class PrecipitationReading(BaseModel):
"""Precipitation reading"""
time: datetime = Field(..., description="Timestamp")
value: float = Field(..., ge=0, description="Precipitation in mm")
class EvapotranspirationReading(BaseModel):
"""Evapotranspiration reading"""
time: datetime = Field(..., description="Timestamp")
value: float = Field(..., description="Evapotranspiration in mm")
class SurfacePressureReading(BaseModel):
"""Surface pressure reading"""
time: datetime = Field(..., description="Timestamp")
value: float = Field(..., gt=0, description="Surface pressure in hPa")
class MeteoData(BaseModel):
"""Model to store meteorological data"""
temperature: list[TemperatureReading] = Field(..., description="List of temperature readings")
humidity: list[HumidityReading] = Field(..., description="List of humidity readings")
apparent_temperature: list[ApparentTemperatureReading] = Field(..., description="List of apparent temperature readings")
precipitation: list[PrecipitationReading] = Field(..., description="List of precipitation readings")
evapotranspiration: list[EvapotranspirationReading] = Field(..., description="List of evapotranspiration readings")
surface_pressure: list[SurfacePressureReading] = Field(..., description="List of surface pressure readings")
The use of Strands-Agents is very simple. I’ve encapsulated the agent logic in a single function that initializes the agent with the necessary tools and prompts. The agent can then be used to generate weather forecasts or answer specific weather-related questions.
_ = ai(
system_prompt=SYSTEM_PROMPT,
user_prompt="What will the weather be like tomorrow?")
If I run this code, agent will use the provided tools to generate the answer. In the output of the command, you will see the agent’s reasoning, the tools it used, and the final answer. First it uses the current time tool to get the current date and time (using current_time tool), then it uses the get_hourly_weather_data tool to fetch the weather data, after that uses think tool to analyze the data, and finally it uses the python_repl tool to generate the needed calculations, using Pandas, and produce the final weather forecast. Here’s an example of the output you might see when running the agent:
12/07/2025 15:28:18 [INFO] Check agent weather.
12/07/2025 15:28:18 [INFO] Found credentials in shared credentials file: ~/.aws/credentials
12/07/2025 15:28:18 [INFO] Creating Strands MetricsClient
I'll help you get the weather forecast for tomorrow. First, let me get the current time to determine what "tomorrow" is, and then I'll fetch the weather data for you.
Tool #1: current_time
Now I'll get the hourly weather data for tomorrow (July 13, 2025):
Tool #2: get_hourly_weather_data
12/07/2025 15:28:24 [INFO] [get_hourly_weather_data] Fetched weather data from 2025-07-13 to 2025-07-13. 24 records found.
Let me analyze this weather data to provide you with a comprehensive forecast for tomorrow (July 13, 2025):
Tool #3: think
Let me analyze the weather data directly to provide you with tomorrow's forecast:
Tool #4: python_repl
WEATHER FORECAST FOR TOMORROW (July 13, 2025)
==================================================
Temperature Range: 20.5°C to 23.6°C
Average Temperature: 22.2°C
Feels Like Temperature: Up to 24.6°C
Humidity Range: 70% to 83%
Average Humidity: 77%
Total Precipitation: 1.0mm
Precipitation Timing: Brief shower 4-5 PM
Pressure: Rising from 1012.1 to 1019.9 hPa
Pressure Change: +7.8 hPa (improving conditions)
Tool #5: python_repl
DETAILED HOURLY BREAKDOWN
==============================
🌅 MORNING (6 AM - 12 PM):
Temperature: 20.6°C to 23.0°C
Humidity: 70% to 82%
Conditions: Clear, comfortable temperatures
☀️ AFTERNOON (12 PM - 6 PM):
Temperature: 22.6°C to 23.6°C
Humidity: 71% to 81%
Precipitation: 1.0mm (brief shower 4-5 PM)
Conditions: Warmest part of day, brief afternoon shower
🌙 EVENING (6 PM - 12 AM):
Temperature: 22.0°C to 22.4°C
Humidity: 75% to 82%
Conditions: Pleasant evening temperatures, increasing humidity
📋 SUMMARY & RECOMMENDATIONS:
• Mild and pleasant day with comfortable temperatures
• Brief shower expected between 4-5 PM (carry umbrella)
• Rising atmospheric pressure indicates stable weather
• High humidity in morning and evening - may feel muggy
• Good day for outdoor activities, except during afternoon shower
• Light layers recommended due to temperature variation
## Weather Forecast for Tomorrow (July 13, 2025)
**Overall Conditions:** A pleasant, mild day with mostly clear skies and a brief afternoon shower.
### 🌡️ **Temperature**
- **Range:** 20.5°C to 23.6°C (69°F to 74°F)
- **Feels like:** Up to 24.6°C (76°F)
- **Coolest:** Early morning around 2-6 AM
- **Warmest:** Mid-afternoon around 2 PM
### 🌧️ **Precipitation**
- **Total:** 1.0mm of rain
- **Timing:** Brief shower between 4-5 PM
- **Duration:** Short-lived, about 1 hour
- **Recommendation:** Carry an umbrella for the afternoon
### 💨 **Atmospheric Conditions**
- **Humidity:** 70-83% (higher in morning/evening)
- **Pressure:** Rising from 1012 to 1020 hPa (+7.8 hPa increase)
- **Trend:** Improving and stabilizing weather conditions
### ⏰ **Time-Based Breakdown**
**Morning (6 AM - 12 PM):** Clear and comfortable, 20.6-23°C
**Afternoon (12 PM - 6 PM):** Warmest period with brief shower, 22.6-23.6°C
**Evening (6 PM - 12 AM):** Pleasant and stable, 22-22.4°C
### 📝 **Recommendations**
- **Clothing:** Light layers, as temperatures vary throughout the day
- **Activities:** Great for outdoor plans, just avoid 4-5 PM for the shower
- **Comfort:** May feel slightly muggy due to higher humidity
- **Overall:** A very pleasant summer day with stable, improving weather conditions
The rising atmospheric pressure indicates this will be a stable weather day with good conditions for most outdoor activities!
Process finished with exit code 0
Here you can see the ai function.
import logging
from botocore.config import Config
from strands import Agent
from strands.agent import AgentResult
from strands.models import BedrockModel
from strands_tools import calculator, file_write, current_time, think, python_repl
from core.aws import get_aws_session
from modules.weather.tools import Tools
from settings import (
IA_MODEL, IA_TEMPERATURE, LLM_READ_TIMEOUT, LLM_CONNECT_TIMEOUT,
LLM_MAX_ATTEMPTS, MY_LATITUDE, MY_LONGITUDE, )
logger = logging.getLogger(__name__)
def get_agent(
system_prompt: str,
read_timeout: int = LLM_READ_TIMEOUT,
connect_timeout: int = LLM_CONNECT_TIMEOUT,
max_attempts: int = LLM_MAX_ATTEMPTS) -> Agent:
config = Config(
read_timeout=read_timeout,
connect_timeout=connect_timeout,
retries={'max_attempts': max_attempts}
)
session = get_aws_session()
base_tools = [calculator, think, python_repl, file_write, current_time]
custom_tools = Tools(latitude=MY_LATITUDE, longitude=MY_LONGITUDE).get_tools()
all_tools = base_tools + custom_tools
bedrock_model = BedrockModel(
model_id=IA_MODEL,
temperature=IA_TEMPERATURE,
boto_session=session,
boto_client_config=config,
)
return Agent(
model=bedrock_model,
tools=all_tools,
system_prompt=system_prompt
)
def ai(
system_prompt: str,
user_prompt: str,
read_timeout: int = 300,
connect_timeout: int = 60,
max_attempts: int = 5) -> AgentResult:
agent = get_agent(
system_prompt=system_prompt,
read_timeout=read_timeout,
connect_timeout=connect_timeout,
max_attempts=max_attempts)
return agent(user_prompt)
As you can see, the agent is only a few lines of code. The magic is in the prompts and the tools that it uses. The agent can be used to generate weather forecasts, analyze historical weather data, and provide practical recommendations based on the weather conditions. This is the main prompt:
FORECAST_PROMPT = f"""
## Instructions for the weather forecast
Your mission is to analyze weather data and provide accurate and useful forecasts for the next {{days}} days.
You have access to a tool called `get_hourly_weather_data` that allows you to obtain hourly weather data.
As a meteorology expert, you must thoroughly analyze the data and provide accurate and useful forecasts.
Take into account possible extreme heat days, especially in summer.
Remember that extreme heat is considered when maximum and minimum temperatures exceed local temperature thresholds for several consecutive days,
often during a heatwave. These temperatures, along with humidity, can be harmful to health, especially for vulnerable groups.
## Report style
All reports must be written in English.
The report must be clear, concise, and easy to understand.
It should include:
- A summary of current weather conditions.
- A detailed forecast for the coming days, including temperature, precipitation, wind, and any other relevant data.
- Practical recommendations based on the forecast, such as precautions to take or recommended activities.
- Be creative and innovative in your approach, using advanced data visualization techniques to enhance the report.
## Data visualization
The report, in markdown, must be visually appealing and innovative.
You will use tables, lists, and other formatting elements to enhance readability.
### Graph format
- Generate the graph configuration in JSON format, compatible with the Vegalite library.
- Ensure the JSON is valid and compatible with the Vegalite library.
- The graphs must be innovative, leveraging the library's potential. Do not limit yourself to simple bar or line charts. Aim for a wow effect.
- Required JSON structure:
* title: main title of the graph, at the top of the graph. The title must be brief and descriptive.
* the title must be in the layout.title.text directive
* layout.showlegend will be true/false, to show the graph legend. Some graphs do not need a legend, such as simple line charts.
- After each graph, generate a blockquote briefly explaining what the graph shows and its context.
...
For the visualization I’m using MkDocs , a simple static site generator for Markdown files. To have more advanced visualizations, I’m using the Vega-Lite library, which allows you to create interactive and visually appealing charts. The agent generates the JSON configuration for the graphs in a format compatible with Vega-Lite, which can then be rendered in the Markdown reports.
For AI, I’m using Claude 3.5 Sonnet, provided by Amazon Bedrock. For the experiment it’s enough, but if you create a cron job to run the agent every day, you’ll have your 5-day forecasting system ready to go. The project tries to show how to use AI agents to solve real-world problems, and how to integrate them with external data sources and tools. The agent can be extended to include more advanced features, such as integrating with other APIs or using more complex machine learning models for weather prediction.
Today we are going to build an agent with IA. It is just an example of how to build a agent with LangChain and AWS Bedrock and Claude 4 Sonnet. The agent will be a “mathematical expert” capable of performing complex calculations and providing detailed explanations of its reasoning process. The idea is to provide the agent with the ability to perform mathematical operations like addition, subtraction. In fact, with additions and subtractions, we can perform all the mathematical operations, like multiplication, division, exponentiation, square root, etc. The agent will be able to perform these operations step by step, providing a detailed explanation of its reasoning process. I know that we don’t need to use AI to perform these operations, but the idea is to show how to build an agent with LangChain and AWS Bedrock and Claude 4 Sonnet.
The mathematical agent implements the tool-calling pattern, allowing the LLM to dynamically select and execute mathematical operations:
Tools are defined using LangChain’s @tool decorator, providing automatic schema generation and type validation. Really we don’t need to create a class for the tools, but I have done it because I want to add an extra feature to the agent: the ability to keep a history of the operations performed. This will allow the agent to provide a detailed explanation of its reasoning process, showing the steps taken to arrive at the final result.
import logging
from typing import List
from langchain.tools import tool
logger = logging.getLogger(__name__)
class MathTools:
def __init__(self):
self.history = []
def _diff_values(self, a: int, b: int) -> int:
result = a - b
self.history.append(f"{a} - {b} = {result}")
return result
def _sum_values(self, a: int, b: int) -> int:
result = a + b
self.history.append(f"{a} + {b} = {result}")
return result
def _get_history(self) -> str:
if not self.history:
return "No previous operations"
return "\n".join(self.history[-5:]) # Last 5
def get_tools(self) -> List:
@tool
def diff_values(a: int, b: int) -> int:
"""Calculates the difference between two numbers
Args:
a (int): first number
b (int): second number
Returns:
int: difference of a - b
"""
logger.info(f"Calculating difference: {a} - {b}")
return self._diff_values(a, b)
@tool
def sum_values(a: int, b: int) -> int:
"""Sums two numbers
Args:
a (int): first number
b (int): second number
Returns:
int: sum of a + b
"""
logger.info(f"Calculating sum: {a} + {b}")
return self._sum_values(a, b)
@tool
def get_history() -> str:
"""Gets the operation history
Returns:
str: last operations
"""
logger.info("Retrieving operation history")
return self._get_history()
return [diff_values, sum_values, get_history]
The system prompt is carefully crafted to guide the agent’s behavior and establish clear operational boundaries:
AGENT_SYSTEM_PROMPT = """
You are an expert mathematical agent specialized in calculations.
You have access to the following tools:
- diff_values: Calculates the difference between two numbers
- sum_values: Sums two numbers
- get_history: Gets the operation history
Guidelines:
1. Only answer questions related to mathematical operations.
2. For complex operations, use step-by-step calculations:
- Multiplication: Repeated addition
- Division: Repeated subtraction
- Exponentiation: Repeated multiplication
- Square root: Use methods like Babylonian method or prime factorization.
"""
Now we can invoke our agent by asking questions such as ‘What’s the square root of 16 divided by two, squared?’. The agent will iterate using only the provided tools to obtain the result.
And that’s all. This project demonstrates how to build a production-ready AI agent using LangChain and AWS Bedrock. It’s just a boilerplate, but it can be extended to create more complex agents with additional capabilities and understand how AI agents work.