IMPORTANT DISCLAIMER
What you are about to read describes a first prototype. It is a proof of
concept that demonstrates the feasibility of combining agentic AI with 3D
printing hardware. The code shown here requires further refinement, security
hardening, and testing before it is suitable for any production environment.
Treat every code snippet as a starting point, not a finished product. Edge
cases, hardware variations, network conditions, and model behaviour will all
require you to adapt and extend what is shown here. With that said, let us
build something genuinely exciting.
The Bambu Lab X1C used in the code below is a widely distributed 3D printer, Of course, some of the code needs to be rewritten when addressing a different 3D printer.
CHAPTER ONE: THE BIG PICTUREThere is a particular kind of frustration that every 3D printing enthusiast knows intimately. You start a six-hour print, walk away feeling confident, and return to find your build plate decorated with what the community has affectionately named "spaghetti": a chaotic tangle of melted filament that bears no resemblance whatsoever to the model you intended to produce. Hours of machine time and a spool of expensive filament are gone, and you are left wondering what went wrong and how you might have caught it sooner. This tutorial addresses that frustration directly. We are going to build an agentic AI application that runs entirely on a Raspberry Pi 5 and interacts with a Bambu Lab X1C 3D printer over a local network. The application does three things. First, it acts as a knowledgeable advisor before a print begins, analysing your intended slicer settings and suggesting improvements based on the filament type, model geometry, and known best practices for the X1C. Second, it watches your print in real time using a USB camera, using a Vision Language Model to detect failures like spaghetti, detachment, and warping, and it sends you a Telegram message the moment something looks wrong. Third, when a print fails, it consults with you about what might have caused the problem, drawing on the print history and the visual evidence captured during the failure. The word "agentic" is important here. We are not building a simple script that checks a camera feed and sends an alert. We are building a system of cooperating AI agents, each with a specific role, orchestrated by LangGraph into a coherent workflow. The agents can reason, use tools, call each other, and produce structured outputs that drive real actions in the physical world. This is the frontier of applied AI in 2026, and the Raspberry Pi 5 is surprisingly capable of hosting it. A note on the hardware choice: the Raspberry Pi 5 with 8 GB of RAM is the minimum recommended configuration for this project. If you have access to the Raspberry Pi AI HAT+ 2, released in January 2026 and featuring the Hailo-10H neural network accelerator with 40 TOPS of INT4 inference performance and 8 GB of dedicated on-board RAM, you will get significantly better performance for the LLM and VLM inference tasks. The tutorial works without it, but you should expect slower response times from the language models if you rely on the CPU alone.
CHAPTER TWO: UNDERSTANDING THE ARCHITECTUREBefore we write a single line of code, we need to understand how the pieces fit together. A well-designed system is one where each component has a single, clear responsibility and communicates with other components through well-defined interfaces. This is the clean architecture principle applied to an AI system. The application has five major layers. The hardware layer consists of the Raspberry Pi 5, a USB camera pointed at the X1C's build plate, and the X1C printer itself connected to the same local network. The camera does not need to be exotic. A Logitech C920 or similar 1080p USB webcam is perfectly adequate. What matters is that it has a clear, unobstructed view of the build plate from a fixed position. The connectivity layer handles communication with the X1C over MQTT. The Bambu Lab X1C runs a local MQTT broker on port 8883 using TLS 1.2 encryption. Our application connects to this broker, subscribes to status topics, and publishes command topics. This gives us real-time access to printer state: temperatures, print progress, error codes, AMS status, and more. Status messages arrive every 0.5 to 2 seconds during active printing. The inference layer is where the AI lives. We run Ollama on the Raspberry Pi 5 to serve language model inference. For the text-based reasoning tasks, we use a quantized model in the 1.5B to 3B parameter range, which the Pi 5 can handle at a usable speed of roughly 4 to 12 tokens per second. For the visual analysis tasks, we use Moondream2, a 2B parameter Vision Language Model specifically designed for efficient edge deployment, loaded via the Hugging Face Transformers library. Moondream2 can answer natural language questions about images, detect objects, and generate descriptions, all of which are exactly what we need for print monitoring. On the Raspberry Pi 5, a single image analysis cycle takes 20 to 90 seconds depending on the complexity of the question, which is why we check every 60 seconds rather than every frame. The orchestration layer is built with LangGraph 1.0, which reached its stable release in October 2025. LangGraph models our AI workflow as a directed graph where each node is a function and edges represent transitions between states. This gives us precise control over the agent's reasoning process, including loops, conditional branches, and tool calls. LangChain 1.0, released at the same time, provides the underlying abstractions for tools, memory, and LLM integration. The notification layer uses direct HTTP calls to the Telegram Bot API via the requests library. This approach is deliberately simple and synchronous: we do not need polling, handlers, or any of the Application machinery that the python-telegram-bot library provides. For our use case — sending a formatted alert message when something goes wrong — a direct POST request to the sendMessage endpoint is the most reliable and straightforward solution. It avoids the asyncio event loop complexity that arises when mixing asynchronous Telegram client code with synchronous LangGraph node functions. The flow through these layers looks like this. When the user is about to start a print, they invoke the pre-print advisor agent, which reads the intended slicer settings, reasons about them using the LLM, and produces a structured report with suggestions. Once the print starts, the monitoring agent begins capturing frames from the camera at regular intervals, passing each frame to Moondream2 with carefully crafted questions, and evaluating the responses. If Moondream2 reports a problem confirmed by the LLM in two consecutive cycles, the monitoring agent triggers the notification agent, which sends a Telegram message to the user. After a print failure, the user can invoke the consultation agent, which reviews the print history, the failure images, and the printer state at the time of failure, and produces a diagnosis with recommended corrective actions.
CHAPTER THREE: PROJECT STRUCTUREAll source files live in a single flat directory. This keeps imports simple and avoids packaging complexity for a prototype. The complete layout is: ~/print_agent/ ├── .env # Secret credentials (never commit this) ├── requirements.txt # Pinned Python dependencies ├── run_print_agent.sh # Launch script ├── print_agent.service # systemd service unit (optional) ├── bambu_mqtt_client.py # MQTT communication with the X1C ├── camera_capture.py # USB camera frame capture ├── vision_analyzer.py # Moondream2 visual analysis ├── notifier.py # Telegram notifications via requests ├── llm_setup.py # Ollama LLM configuration ├── pre_print_advisor.py # Pre-print settings advisor agent ├── monitoring_agent.py # Continuous print monitoring agent ├── failure_consultant.py # Post-failure diagnosis agent └── main.py # Application entry point Create the project directory and navigate into it: mkdir -p ~/print_agent cd ~/print_agent
CHAPTER FOUR: SETTING UP THE RASPBERRY PI 5We begin with a clean installation of Raspberry Pi OS 64-bit (Bookworm), which is the recommended operating system as of mid-2026. The 64-bit variant is essential because the AI models and inference engines we use require it. Flash your microSD card or, preferably, an NVMe SSD via a PCIe HAT, using the Raspberry Pi Imager tool. An NVMe SSD is strongly recommended because model files are large and loading them from a fast storage device makes a noticeable difference in startup time. After booting, update the system fully before installing anything else: sudo apt update && sudo apt full-upgrade -y Active cooling is not optional for this project. The Raspberry Pi 5 will run AI inference workloads continuously during print monitoring, and without active cooling it will thermal-throttle within minutes, dropping performance dramatically. Install the official Raspberry Pi Active Cooler or a third-party equivalent before you begin. Adding a swap file gives the operating system additional virtual memory to work with when models are being loaded. A 4 GB swap file is a reasonable choice: sudo fallocate -l 4G /swapfile sudo chmod 600 /swapfile sudo mkswap /swapfile sudo swapon /swapfile echo '/swapfile none swap sw 0 0' | sudo tee -a /etc/fstab Python 3.11 or newer is required for LangGraph 1.x. Raspberry Pi OS Bookworm ships with Python 3.11, so you should be in good shape. Verify your Python version: python3 --version Install the system-level dependencies needed by OpenCV, PyTorch, and other libraries on ARM64. The libopenmpi-dev and libomp-dev packages are required by PyTorch for its parallel processing support on ARM64: sudo apt install -y \ libopenblas-dev \ libatlas-base-dev \ libopenmpi-dev \ libomp-dev \ libjpeg-dev \ libpng-dev \ libavcodec-dev \ libavformat-dev \ libswscale-dev \ libv4l-dev \ v4l-utils \ git \ curl Create a dedicated virtual environment for the project: python3 -m venv ~/print_agent_env source ~/print_agent_env/bin/activate Now install Python dependencies. PyTorch for ARM64 must be installed from PyTorch's own CPU wheel index rather than the default PyPI index. This ensures you get the correct ARM64 CPU-optimised wheel. Install PyTorch first, before the rest of the requirements: pip install --upgrade pip pip install torch torchvision --index-url https://download.pytorch.org/whl/cpu The PyTorch download is approximately 200 MB for the ARM64 wheel, so allow several minutes on a typical home connection. Once PyTorch is installed, install the remaining dependencies from the requirements file: pip install -r requirements.txt We use opencv-python-headless rather than opencv-python because the Raspberry Pi in this application has no display attached. The headless variant omits the GUI dependencies and installs cleanly in a server environment. We use the requests library for Telegram notifications rather than the python-telegram-bot library. The python-telegram-bot library v20 and above is fully asynchronous and requires the Bot object to be used as an async context manager for proper httpx session initialisation. Mixing that asynchronous lifecycle with synchronous LangGraph node functions creates unnecessary complexity. A direct HTTP POST to the Telegram Bot API using requests is simpler, synchronous, and perfectly adequate for sending alert messages. Installing Ollama is a one-liner that downloads and configures the inference server. Ollama supports ARM64 and runs comfortably on the Raspberry Pi 5: curl -fsSL https://ollama.ai/install.sh | sh Once Ollama is installed, pull a suitable language model. For this project we use Qwen2.5:1.5b, which offers a good balance of reasoning capability and inference speed on the Pi 5. If you have the AI HAT+ 2, you can experiment with larger models: ollama pull qwen2.5:1.5b Verify that Ollama is running and the model is available: ollama list You should see qwen2.5:1.5b in the list. Ollama runs as a background service and exposes a REST API on localhost port 11434, which LangChain talks to directly through the langchain-ollama integration.
CHAPTER FIVE: REQUIREMENTS FILECreate requirements.txt in the project directory. Note that torch and torchvision must be installed separately before this file, using the PyTorch CPU wheel index as shown in Chapter Four. The remaining packages are installed from this file using pip install -r requirements.txt.
# requirements.txt
# Pinned dependencies for the 3D Print AI Agent.
# Generated for Raspberry Pi OS Bookworm 64-bit, Python 3.11, July 2026.
#
# IMPORTANT: Install PyTorch BEFORE running pip install -r requirements.txt:
# pip install torch torchvision --index-url https://download.pytorch.org/whl/cpu
#
# PyTorch must be installed from the PyTorch CPU wheel index to get the
# correct ARM64 CPU-optimised wheel for the Raspberry Pi 5. Installing from
# the default PyPI index may fetch an incompatible wheel.
#
# After installing PyTorch, run:
# pip install -r requirements.txt
# PyTorch (install separately first — see note above)
torch>=2.3.0
torchvision>=0.18.0
# Hugging Face ecosystem (used for Moondream2 local inference)
transformers==4.46.0
accelerate>=0.30.0
# LangChain and LangGraph (AI orchestration)
langchain==1.0.0
langgraph==1.0.0
langchain-ollama>=0.2.0
# MQTT client for Bambu Lab X1C communication
paho-mqtt>=2.0.0
# HTTP client for Telegram Bot API notifications (synchronous, no asyncio)
requests>=2.31.0
# Data validation and schema enforcement
pydantic>=2.0.0
# Computer vision (headless: no GUI dependencies)
opencv-python-headless>=4.9.0
# Image processing
pillow>=10.0.0
# Numerical computing
numpy>=1.26.0
# Environment variable management
python-dotenv>=1.0.0
CHAPTER SIX: ENVIRONMENT CONFIGURATION
Create a .env file in the project directory. This file holds all sensitive credentials and is never committed to version control. Add it to .gitignore immediately if you initialise a git repository:
echo ".env" >> .gitignore
Environment configuration for the 3D Print AI Agent.
Replace all placeholder values with your actual credentials.
NEVER commit this file to version control.
-----------------------------------------------------------------------
Bambu Lab X1C connection settings.
Find these in: Settings > Network on the printer's touchscreen.
Developer Mode must be enabled to reveal the LAN Access Code.
-----------------------------------------------------------------------
PRINTER_IP=192.168.1.100 PRINTER_SERIAL=YOUR_PRINTER_SERIAL_NUMBER PRINTER_ACCESS_CODE=YOUR_LAN_ACCESS_CODE
-----------------------------------------------------------------------
Telegram notification settings.
BOT_TOKEN: obtained from @BotFather on Telegram (/newbot command).
CHAT_ID: your numeric user ID, obtained from @userinfobot on Telegram.
-----------------------------------------------------------------------
TELEGRAM_BOT_TOKEN=YOUR_BOT_TOKEN_FROM_BOTFATHER TELEGRAM_CHAT_ID=YOUR_NUMERIC_CHAT_ID
-----------------------------------------------------------------------
Camera and monitoring settings.
CAMERA_INDEX: 0 for the first USB camera. Use v4l2-ctl --list-devices
to find the correct index if 0 does not work.
MONITORING_INTERVAL: seconds between each visual analysis cycle.
60 seconds is recommended for Pi 5 without HAT.
-----------------------------------------------------------------------
CAMERA_INDEX=0 MONITORING_INTERVAL=60
CHAPTER SEVEN: CONNECTING TO THE BAMBU LAB X1C OVER MQTTThe Bambu Lab X1C runs a local MQTT broker that allows third-party software to monitor and control the printer over your local network. To use this, you need to enable two settings on the printer itself. Navigate to the printer's touchscreen settings menu, find "Network" or "LAN", and enable both "LAN Only Mode" and "Developer Mode". Once Developer Mode is enabled, the printer will display its LAN Access Code, which is the password you will use to authenticate with the MQTT broker. Write down three pieces of information from the printer's settings screen: the printer's IP address on your local network, the device serial number, and the LAN Access Code. You will need all three. The MQTT broker on the X1C listens on port 8883 and requires TLS 1.2. The username is always the literal string "bblp" and the password is the LAN Access Code. The printer publishes its status to the topic: device/<serial_number>/report And it listens for commands on the topic: device/<serial_number>/request The status messages are JSON documents containing the printer's current state. A typical status message includes the nozzle temperature ("nozzle_temper"), bed temperature ("bed_temper"), print progress as a percentage ("mc_percent"), the name of the file being printed, the current layer number ("layer_num"), the total layer count, and the G-code execution state ("gcode_state"). The confirmed values for "gcode_state" are IDLE, RUNNING, PAUSE, FINISH, and FAILED. Note that Bambu Lab has not formally documented this API, and values may change with firmware updates. A separate field "gcode_file_prepare_percent" indicates preparation progress before printing begins. The following module encapsulates all MQTT communication with the X1C. It uses paho-mqtt 2.0 with CallbackAPIVersion.VERSION2, which is required for the updated callback signatures introduced in that release. The on_connect and on_disconnect callbacks each receive five parameters in VERSION2: client, userdata, flags, reason_code, and properties. The reason_code parameter is a ReasonCode object that compares equal to the integer 0 on success. Notice how the subscribe() call lives inside the _on_connect callback rather than in the connect() method: this ensures the subscription is automatically re-established whenever the connection drops and reconnects, which is essential for a long- running monitoring application.
# bambu_mqtt_client.py
#
# Handles all MQTT communication with the Bambu Lab X1C printer.
# Uses TLS 1.2 as required by the printer's local broker.
# Uses paho-mqtt 2.0 with CallbackAPIVersion.VERSION2.
# Thread-safe status storage allows the monitoring loop to read
# printer state without blocking the MQTT callback thread.
import json
import ssl
import threading
import paho.mqtt.client as mqtt
from paho.mqtt.enums import CallbackAPIVersion
class BambuMQTTClient:
"""
A client for communicating with the Bambu Lab X1C over its local MQTT
broker. Maintains a persistent connection and stores the latest printer
status for consumption by other parts of the application.
"""
MQTT_PORT = 8883
MQTT_USERNAME = "bblp"
def __init__(
self,
printer_ip: str,
serial_number: str,
access_code: str,
) -> None:
"""
Initialise the client with the printer's network credentials.
Args:
printer_ip: The local IP address of the X1C.
serial_number: The printer's serial number, used in topic names.
access_code: The LAN Access Code displayed on the printer.
"""
self._printer_ip = printer_ip
self._serial_number = serial_number
self._access_code = access_code
# The report topic is where the printer publishes its status.
self._report_topic = f"device/{serial_number}/report"
self._request_topic = f"device/{serial_number}/request"
# Thread lock protects the status dictionary from concurrent access.
self._status_lock = threading.Lock()
self._latest_status: dict = {}
self._connected = False
# Build the paho-mqtt 2.0 client with CallbackAPIVersion.VERSION2.
# This is required for the updated callback signatures in paho 2.0.
self._client = mqtt.Client(
callback_api_version=CallbackAPIVersion.VERSION2,
client_id=f"print_agent_{serial_number}",
protocol=mqtt.MQTTv311,
)
self._client.username_pw_set(
username=self.MQTT_USERNAME,
password=access_code,
)
# Configure TLS without certificate verification.
# The X1C uses a self-signed certificate, so we disable hostname
# checking. In a production system you would pin the certificate.
tls_context = ssl.create_default_context()
tls_context.check_hostname = False
tls_context.verify_mode = ssl.CERT_NONE
self._client.tls_set_context(tls_context)
# Register callback functions.
self._client.on_connect = self._on_connect
self._client.on_message = self._on_message
self._client.on_disconnect = self._on_disconnect
def connect(self) -> None:
"""
Establish the MQTT connection and start the background network loop.
The loop runs in a separate thread managed by paho-mqtt.
"""
self._client.connect(
host=self._printer_ip,
port=self.MQTT_PORT,
keepalive=60,
)
self._client.loop_start()
def disconnect(self) -> None:
"""Cleanly disconnect from the MQTT broker."""
self._client.loop_stop()
self._client.disconnect()
def get_printer_state(self) -> dict:
"""
Return a copy of the most recently received printer status.
Returns an empty dict if no status has been received yet.
"""
with self._status_lock:
return dict(self._latest_status)
def _on_connect(
self,
client: mqtt.Client,
userdata: object,
flags: mqtt.ConnectFlags,
reason_code: mqtt.ReasonCode,
properties: mqtt.Properties,
) -> None:
"""
Called by paho-mqtt 2.0 when the connection is established.
The VERSION2 signature adds 'properties' and changes the integer
'rc' to a ReasonCode object. ReasonCode compares equal to 0 on
success. Subscribing here ensures resubscription after reconnects.
"""
if reason_code == 0:
self._connected = True
client.subscribe(self._report_topic, qos=1)
print(f"[MQTT] Connected and subscribed to {self._report_topic}")
else:
print(f"[MQTT] Connection failed with reason code {reason_code}")
def _on_message(
self,
client: mqtt.Client,
userdata: object,
message: mqtt.MQTTMessage,
) -> None:
"""
Called by paho-mqtt when a message arrives on a subscribed topic.
Parses the JSON payload and merges it into the status dictionary.
The X1C wraps its print status inside a top-level "print" key.
"""
try:
payload = json.loads(message.payload.decode("utf-8"))
with self._status_lock:
if "print" in payload:
self._latest_status.update(payload["print"])
except json.JSONDecodeError as exc:
print(f"[MQTT] Failed to parse message: {exc}")
def _on_disconnect(
self,
client: mqtt.Client,
userdata: object,
flags: mqtt.DisconnectFlags,
reason_code: mqtt.ReasonCode,
properties: mqtt.Properties,
) -> None:
"""
Called by paho-mqtt 2.0 when the connection is lost.
The VERSION2 on_disconnect signature includes 'flags'
(type DisconnectFlags) and 'properties' in addition to the
reason_code. paho-mqtt will attempt reconnection automatically
when loop_start() is in use.
"""
self._connected = False
print(
f"[MQTT] Disconnected with reason code {reason_code}. "
"Will attempt reconnection."
)
CHAPTER EIGHT: CAPTURING FRAMES FROM THE USB CAMERA
The camera module is conceptually simple but deserves careful implementation. We need to capture frames reliably, handle camera disconnections gracefully, and provide frames to the VLM in a format it can process efficiently.
One subtlety worth explaining: OpenCV's VideoCapture.read() is a blocking call that waits for the next frame from the camera. If we call it directly in the monitoring loop, it will block the entire loop for the duration of the frame capture. To avoid this, the CameraCapture class runs a background thread that continuously reads frames and stores the most recent one. The monitoring loop can then call get_latest_frame() at any time and receive the most recent frame without blocking.
Moondream2 performs best on the Raspberry Pi 5 when images are no larger than 512x512 pixels. Larger images increase the encoding time substantially. The get_latest_frame() method therefore resizes the captured frame to 512x512 before returning it, which keeps analysis times predictable.
The conversion from OpenCV's BGR colour order to PIL's RGB order is a detail that trips up many developers. OpenCV stores colour channels in Blue-Green-Red order by convention, a legacy from its early days, while virtually every other image processing library uses Red-Green-Blue order. Moondream2 expects PIL Images in RGB format, so the conversion is essential for correct colour interpretation during visual analysis.
# camera_capture.py
#
# Provides a thread-safe interface for capturing frames from a USB camera.
# Uses a background thread to continuously read frames so that the caller
# always gets the most recent image without blocking.
# Frames are resized to 512x512 pixels for optimal Moondream2 performance
# on the Raspberry Pi 5.
import threading
import time
from typing import Optional
import cv2
import numpy as np
from PIL import Image
# Target resolution for Moondream2 on Raspberry Pi 5.
# Larger images increase encoding time significantly on CPU inference.
MOONDREAM_TARGET_SIZE = (512, 512)
class CameraCapture:
"""
Manages a USB camera connection and provides the latest captured frame
on demand. Runs a background thread to keep the frame buffer fresh.
"""
def __init__(
self,
camera_index: int = 0,
capture_width: int = 1280,
capture_height: int = 720,
) -> None:
"""
Initialise the camera capture.
Args:
camera_index: The V4L2 device index (0 for the first camera).
capture_width: Desired capture width in pixels (for the raw feed).
capture_height: Desired capture height in pixels (for the raw feed).
"""
self._camera_index = camera_index
self._capture_width = capture_width
self._capture_height = capture_height
self._capture: Optional[cv2.VideoCapture] = None
self._latest_frame: Optional[np.ndarray] = None
self._frame_lock = threading.Lock()
self._running = False
self._capture_thread: Optional[threading.Thread] = None
def start(self) -> bool:
"""
Open the camera and start the background capture thread.
Returns:
True if the camera was opened successfully, False otherwise.
"""
self._capture = cv2.VideoCapture(self._camera_index)
if not self._capture.isOpened():
print(
f"[Camera] Failed to open camera at index {self._camera_index}. "
"Try a different index or check the connection."
)
return False
# Request the desired resolution. The camera may not honour these
# values exactly; the actual resolution depends on the hardware.
self._capture.set(cv2.CAP_PROP_FRAME_WIDTH, self._capture_width)
self._capture.set(cv2.CAP_PROP_FRAME_HEIGHT, self._capture_height)
self._running = True
self._capture_thread = threading.Thread(
target=self._capture_loop,
daemon=True,
name="CaptureThread",
)
self._capture_thread.start()
print(f"[Camera] Started capture from device index {self._camera_index}.")
return True
def stop(self) -> None:
"""Stop the background thread and release the camera resource."""
self._running = False
if self._capture_thread is not None:
self._capture_thread.join(timeout=3.0)
if self._capture is not None:
self._capture.release()
print("[Camera] Capture stopped and camera released.")
def get_latest_frame(self) -> Optional[Image.Image]:
"""
Return the most recently captured frame as a PIL Image resized to
512x512 pixels, or None if no frame has been captured yet.
The frame is:
1. Converted from OpenCV BGR to RGB colour order.
2. Converted to a PIL Image.
3. Resized to MOONDREAM_TARGET_SIZE for efficient VLM inference.
Returns:
A PIL Image ready for Moondream2, or None if unavailable.
"""
with self._frame_lock:
if self._latest_frame is None:
return None
# Convert BGR (OpenCV convention) to RGB (PIL/Moondream convention).
rgb_frame = cv2.cvtColor(self._latest_frame, cv2.COLOR_BGR2RGB)
pil_image = Image.fromarray(rgb_frame)
# Resize outside the lock to minimise lock-hold time.
return pil_image.resize(MOONDREAM_TARGET_SIZE, Image.LANCZOS)
def _capture_loop(self) -> None:
"""
Background thread function. Continuously reads frames from the camera
and stores the most recent one. Sleeps briefly between reads to avoid
consuming 100% of a CPU core while the monitoring loop is idle.
"""
while self._running:
if self._capture is None or not self._capture.isOpened():
time.sleep(1.0)
continue
success, frame = self._capture.read()
if success:
with self._frame_lock:
self._latest_frame = frame
else:
print(
"[Camera] Frame read failed. "
"Camera may be disconnected or busy."
)
time.sleep(1.0)
If you are unsure which camera index to use, list the available video devices from the terminal:
v4l2-ctl --list-devices
This shows the device names and their corresponding /dev/video entries. The number at the end of the /dev/video path corresponds to the camera_index parameter. If your USB camera is the only camera connected, it will almost certainly be index 0.
CHAPTER NINE: THE VISION LANGUAGE MODEL — MOONDREAM2
Moondream2 is a 2B parameter Vision Language Model developed specifically for efficient deployment on edge devices. It was released in March 2024 and has received continuous updates, with a significant architecture update in June 2025. Its design philosophy is exactly what we need: a model that can answer natural language questions about images without requiring a GPU or a cloud connection.
On the Raspberry Pi 5, we load Moondream2 via the Hugging Face Transformers library using AutoModelForCausalLM and AutoTokenizer. This is the correct path for CPU inference on ARM64 hardware. The alternative — the moondream Python package with local=True — relies on the Photon inference engine, which requires an NVIDIA Ampere or newer GPU or Apple Silicon. Neither is present on the Raspberry Pi 5.
The key capability we use is Visual Question Answering. We pass Moondream2 an image of the print bed and ask it a specific, carefully worded question about what it sees. The quality of the question matters enormously. A vague question produces a vague answer. A precise question produces a precise, actionable answer.
For spaghetti detection, the question is: "Is there any tangled or chaotic filament visible on the print bed that does not appear to be part of the intended 3D print? Answer yes or no, then briefly describe what you see."
The two-step encode-then-query pattern is central to efficient use of Moondream2. The model processes images in two phases: first it encodes the raw pixel data into a compact internal representation, and then it uses that representation to answer questions. By separating these phases, we can ask multiple questions about the same image without paying the encoding cost more than once. The VisionAnalyzer class exposes an analyse_frame() method that encodes the image exactly once and then runs all three analysis questions against the encoded representation.
The answer_question method accepts the encoded image, the question string, and the tokenizer as a keyword argument (tokenizer=self._tokenizer). The tokenizer is responsible for converting the question text into token IDs that the model can process and for decoding the model's output token IDs back into readable text.
We also wrap all inference calls in a torch.no_grad() context manager. This disables PyTorch's gradient computation machinery, which is only needed during training and not during inference. Disabling it reduces memory usage and speeds up each forward pass. The context propagates into all functions called within the with block, so both encode_image and answer_question benefit from it.
On the Raspberry Pi 5 without an accelerator, encoding a 512x512 image takes approximately 8 to 25 seconds. Each subsequent question against the encoded image takes 5 to 15 seconds. Running three questions per cycle therefore takes roughly 25 to 70 seconds in total, which fits comfortably within a 60-second monitoring interval when the wait time is counted separately.
# vision_analyzer.py
#
# Wraps the Moondream2 Vision Language Model for use in print monitoring.
# Uses the Hugging Face Transformers library for CPU inference on ARM64,
# which is the correct path for Raspberry Pi 5 (no Photon engine support).
# The model is loaded once at initialisation and kept in memory.
# Provides an analyse_frame() method that encodes the image exactly once
# and runs all three analysis questions against the encoded representation.
# All inference is wrapped in torch.no_grad() to save memory and improve
# speed by disabling gradient computation. The no_grad context propagates
# into all functions called within the with block.
from typing import Optional
import torch
from PIL import Image
from transformers import AutoModelForCausalLM, AutoTokenizer
class VisionAnalyzer:
"""
Provides visual question answering capabilities using Moondream2
via the Hugging Face Transformers library.
Designed for CPU inference on the Raspberry Pi 5.
"""
# Hugging Face model identifier and pinned revision for reproducibility.
# Update the revision when a new stable checkpoint is released.
MODEL_ID = "vikhyatk/moondream2"
MODEL_REVISION = "2025-06-21"
# Questions used for different analysis scenarios.
# Phrasing matters: start with yes/no to make parsing unambiguous.
SPAGHETTI_QUESTION = (
"Is there any tangled, chaotic, or randomly distributed filament "
"on the print bed that does not appear to be part of an intentional "
"3D print structure? Answer 'yes' or 'no' first, then describe "
"what you observe in one or two sentences."
)
DETACHMENT_QUESTION = (
"Has any part of the 3D print detached or lifted from the build "
"plate? Answer 'yes' or 'no' first, then describe the location "
"and extent of any detachment you can see."
)
GENERAL_HEALTH_QUESTION = (
"Describe the current state of the 3D print on the build plate. "
"Is the print adhering to the bed? Are the layers stacking "
"correctly and uniformly? Is there any visible warping, "
"detachment, blobbing, or layer shifting? Be specific and concise."
)
def __init__(self) -> None:
"""
Initialise the vision analyzer. The model is not loaded here;
call load_model() explicitly to control when loading occurs.
"""
self._model: Optional[AutoModelForCausalLM] = None
self._tokenizer: Optional[AutoTokenizer] = None
self._loaded = False
def load_model(self) -> None:
"""
Load the Moondream2 model and tokenizer into memory via the
Hugging Face Transformers library. This operation takes
approximately 30 to 90 seconds on a Raspberry Pi 5 and requires
roughly 4 GB of RAM for the 2B parameter model. Call this once
at application startup.
"""
print(
f"[Vision] Loading Moondream2 ({self.MODEL_ID} @ "
f"{self.MODEL_REVISION}) via Hugging Face Transformers..."
)
self._tokenizer = AutoTokenizer.from_pretrained(
self.MODEL_ID,
revision=self.MODEL_REVISION,
trust_remote_code=True,
)
self._model = AutoModelForCausalLM.from_pretrained(
self.MODEL_ID,
revision=self.MODEL_REVISION,
trust_remote_code=True,
# Use float32 for CPU inference on ARM64.
# float16 is not reliably supported on all ARM CPU configurations.
torch_dtype=torch.float32,
)
# Set the model to evaluation mode to disable dropout and
# batch normalisation training behaviour.
self._model.eval()
self._loaded = True
print("[Vision] Moondream2 loaded successfully.")
def analyse_frame(self, image: Image.Image) -> dict:
"""
Perform a complete analysis of a single print frame.
Encodes the image exactly once, then runs all three questions
against the encoded representation to avoid redundant computation.
All inference is wrapped in torch.no_grad() to reduce memory
usage and improve speed. The context propagates into _query_encoded.
Args:
image: A PIL Image of the print bed (ideally 512x512 pixels).
Returns:
A dictionary with keys:
"spaghetti": {"detected": bool, "description": str}
"detachment": {"detected": bool, "description": str}
"general_health": str
"""
if not self._loaded:
raise RuntimeError("Model not loaded. Call load_model() first.")
with torch.no_grad():
# Encode the image once. All subsequent queries reuse this
# encoding, avoiding redundant computation.
encoded_image = self._model.encode_image(image)
spaghetti_answer = self._query_encoded(
encoded_image, self.SPAGHETTI_QUESTION
)
detachment_answer = self._query_encoded(
encoded_image, self.DETACHMENT_QUESTION
)
health_answer = self._query_encoded(
encoded_image, self.GENERAL_HEALTH_QUESTION
)
return {
"spaghetti": {
"detected": spaghetti_answer.lower().startswith("yes"),
"description": spaghetti_answer,
},
"detachment": {
"detected": detachment_answer.lower().startswith("yes"),
"description": detachment_answer,
},
"general_health": health_answer,
}
def _query_encoded(self, encoded_image: object, question: str) -> str:
"""
Ask a question about an already-encoded image.
Uses Moondream2's answer_question method, passing the tokenizer
as a keyword argument as required by the Moondream2 API.
Must be called from within a torch.no_grad() context.
Args:
encoded_image: The encoded image from model.encode_image().
question: The natural language question to ask.
Returns:
The model's answer as a stripped string.
"""
answer = self._model.answer_question(
encoded_image,
question,
tokenizer=self._tokenizer,
)
return answer.strip()
CHAPTER TEN: THE NOTIFICATION SYSTEM
When a problem is detected, we need to tell the user immediately. Telegram is an excellent choice for this because it has a robust bot API, messages are delivered reliably even on mobile networks, and the API is simple to call directly over HTTP.
Before writing any code, you need to create a Telegram bot and obtain two pieces of information. First, talk to @BotFather on Telegram and use the /newbot command to create a new bot. BotFather will give you an HTTP API token that looks like a long string of numbers and letters separated by a colon. Second, you need your own Telegram user ID, which you can obtain by messaging @userinfobot on Telegram. This numeric ID tells the bot where to send messages.
We implement the notifier using the requests library to make direct HTTP POST calls to the Telegram Bot API's sendMessage endpoint. This is a deliberate design choice. The python-telegram-bot library version 20 and above is fully asynchronous: the Bot object must be used as an async context manager (async with Bot(...) as bot:) for proper initialisation of its underlying httpx session. Mixing that asynchronous lifecycle with the synchronous LangGraph node functions that call the notifier would require either running a persistent event loop in a background thread or using asyncio.run() to create a new event loop on every call. Both approaches add complexity and potential failure modes.
The requests-based approach is synchronous, has no event loop requirements, and is perfectly adequate for our use case. The Telegram Bot API is a standard REST API and the requests library is the most widely used HTTP client in the Python ecosystem.
Telegram limits messages to 4096 characters. The notifier automatically truncates messages that exceed 4000 characters and appends a truncation notice, ensuring that long diagnostic reports are always delivered rather than silently rejected by the API.
# notifier.py
#
# Handles sending notifications to the user via the Telegram Bot API.
# Uses the requests library for direct, synchronous HTTP POST calls.
# This avoids the asyncio lifecycle complexity of the python-telegram-bot
# library, which requires the Bot object to be used as an async context
# manager for proper httpx session initialisation.
#
# Telegram limits messages to 4096 characters. Messages exceeding 4000
# characters are automatically truncated with a notice appended.
import requests
from requests.exceptions import RequestException
# Telegram's hard limit is 4096 characters. We truncate at 4000 to leave
# room for the truncation notice itself.
_MAX_MESSAGE_LENGTH = 4000
_TRUNCATION_NOTICE = "\n...[message truncated]"
class TelegramNotifier:
"""
Sends notification messages to a specific Telegram user or chat
by making direct HTTP POST requests to the Telegram Bot API.
All methods are synchronous and safe to call from LangGraph nodes.
"""
def __init__(self, bot_token: str, chat_id: str) -> None:
"""
Initialise the notifier with Telegram credentials.
Args:
bot_token: The HTTP API token provided by BotFather.
chat_id: The Telegram user or chat ID to send messages to.
"""
self._bot_token = bot_token
self._chat_id = chat_id
self._send_url = (
f"https://api.telegram.org/bot{bot_token}/sendMessage"
)
def send_alert(self, message: str) -> bool:
"""
Send an alert message to the configured Telegram chat.
Truncates the message if it exceeds the Telegram character limit.
Args:
message: The HTML-formatted text content of the alert.
Returns:
True if the message was sent successfully, False otherwise.
"""
if len(message) > _MAX_MESSAGE_LENGTH:
message = message[:_MAX_MESSAGE_LENGTH] + _TRUNCATION_NOTICE
payload = {
"chat_id": self._chat_id,
"text": message,
"parse_mode": "HTML",
}
try:
response = requests.post(
self._send_url,
json=payload,
timeout=10,
)
response.raise_for_status()
print("[Notifier] Alert sent successfully.")
return True
except RequestException as exc:
print(f"[Notifier] Failed to send alert: {exc}")
return False
def send_print_failure_alert(
self,
failure_type: str,
description: str,
print_progress: int,
layer_number: int,
) -> bool:
"""
Send a structured alert for print failure events.
Args:
failure_type: A short label for the type of failure detected.
description: The VLM's description of what it observed.
print_progress: The print completion percentage at failure time.
layer_number: The layer number at which failure was detected.
Returns:
True if the message was sent successfully, False otherwise.
"""
message = (
"<b>PRINT FAILURE DETECTED</b>\n\n"
f"<b>Type:</b> {failure_type}\n"
f"<b>Progress:</b> {print_progress}% complete\n"
f"<b>Layer:</b> {layer_number}\n\n"
f"<b>What the AI saw:</b>\n{description}\n\n"
"Please check your printer. The print may need to be cancelled."
)
return self.send_alert(message)
def send_pre_print_advice(self, advice_report: str) -> bool:
"""
Send the pre-print settings analysis report to the user.
Args:
advice_report: The formatted advice report from the LLM agent.
Returns:
True if the message was sent successfully, False otherwise.
"""
message = f"<b>PRE-PRINT ANALYSIS REPORT</b>\n\n{advice_report}"
return self.send_alert(message)
The parse_mode="HTML" parameter allows us to use basic HTML formatting in our messages: bold text with tags, italic with tags, and code formatting with tags. This makes the alerts much more readable on a mobile phone screen. Telegram's HTML mode is simpler and more predictable than its Markdown mode, which is why we prefer it here.
CHAPTER ELEVEN: THE LLM CONFIGURATION
All three agents share a single LLM configuration function. Centralising this in its own module means that changing the model, temperature, or context window requires editing exactly one file. The function creates a new ChatOllama instance on each call, which is lightweight since ChatOllama is a thin wrapper around HTTP calls to the Ollama server.
A note on structured output reliability: when using with_structured_output() with smaller models like Qwen2.5:1.5b, you may occasionally encounter Pydantic validation errors. Smaller models sometimes produce output that deviates from the requested JSON schema — omitting a required field, using the wrong data type, or including extraneous text. The advisor and consultant agents include try/except blocks around their structured output calls and return a fallback response when validation fails. If you find validation errors occurring frequently, consider switching to a larger model (such as qwen2.5:3b or llama3.2:3b) or simplifying the Pydantic schemas.
# llm_setup.py
#
# Configures the language model connection for all agents.
# Uses Ollama running locally on the Raspberry Pi 5.
# All agents call create_llm() to obtain a configured ChatOllama instance.
from langchain_ollama import ChatOllama
def create_llm(
model_name: str = "qwen2.5:1.5b",
temperature: float = 0.1,
base_url: str = "http://localhost:11434",
) -> ChatOllama:
"""
Create and return a configured ChatOllama instance.
Args:
model_name: The Ollama model to use for inference.
temperature: Controls randomness. Lower values produce more
deterministic, focused responses. 0.1 is appropriate
for analytical tasks like print diagnosis.
base_url: The URL of the Ollama API server.
Returns:
A configured ChatOllama instance ready for use with LangChain.
"""
return ChatOllama(
model=model_name,
temperature=temperature,
base_url=base_url,
# num_ctx sets the context window size. 4096 tokens is sufficient
# for our use cases and keeps memory usage manageable on the Pi 5.
num_ctx=4096,
)
CHAPTER TWELVE: THE PRE-PRINT ADVISOR AGENT
The pre-print advisor agent receives a dictionary of slicer settings and produces a structured report with specific, actionable recommendations. We use Pydantic to define the output schema, which ensures that the LLM's response is always in a predictable format that the rest of the application can process reliably.
The with_structured_output() method on the LLM is one of LangChain 1.0's most powerful features. When you pass a Pydantic model to this method, LangChain instructs the underlying model to produce output that conforms to the schema defined by that model. We explicitly pass method='json_schema' to use Ollama's native structured output API. Since langchain-ollama 0.3.0, json_schema is the default method, but we pass it explicitly to make the intent clear and to ensure the correct method is used regardless of future default changes. The response is automatically parsed and validated, and you receive a proper Python object rather than a raw string.
The system prompt encodes a comprehensive knowledge base of X1C-specific settings, their valid ranges, and their interactions. This is where the advisor's expertise lives. The more specific and accurate the knowledge in the prompt, the more useful the advisor's output will be.
LangGraph node functions return partial state updates: only the keys that changed, not the entire state. LangGraph merges these partial updates into the existing state automatically using reducer functions. When no reducer is specified for a key, the default behaviour is last-write-wins, meaning the returned value overwrites the existing value. This is the correct pattern and it is what all node functions in this application follow.
# pre_print_advisor.py
#
# Implements the pre-print settings advisor as a LangGraph graph.
# Accepts slicer settings as input and produces a structured advice
# report using an LLM with Pydantic schema validation.
# Uses method='json_schema' for reliable structured output with Ollama.
# Includes a fallback response for cases where the model fails to produce
# valid structured output (common with smaller quantized models).
# Node functions return partial state updates (only changed keys).
from typing import List, Optional, TypedDict
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import END, START, StateGraph
from pydantic import BaseModel, Field
from llm_setup import create_llm
# ---------------------------------------------------------------------------
# Output schema definition
# ---------------------------------------------------------------------------
class PrintSettingIssue(BaseModel):
"""Represents a single identified issue with the print settings."""
setting_name: str = Field(
description="The name of the slicer setting that has an issue."
)
current_value: str = Field(
description="The current value of the setting as provided."
)
recommended_value: str = Field(
description="The recommended value for this setting."
)
reason: str = Field(
description="A clear explanation of why this change is recommended."
)
priority: str = Field(
description=(
"The priority of this recommendation: "
"'critical', 'important', or 'optional'."
)
)
class PrePrintAdviceReport(BaseModel):
"""The complete pre-print analysis report produced by the advisor agent."""
overall_assessment: str = Field(
description="A brief overall assessment of the print settings quality."
)
issues_found: List[PrintSettingIssue] = Field(
description="A list of specific issues identified in the settings.",
default_factory=list,
)
general_tips: List[str] = Field(
description="General tips for this specific print job.",
default_factory=list,
)
estimated_success_likelihood: str = Field(
description=(
"An estimate of print success likelihood: 'high', 'medium', "
"or 'low', with a brief justification."
)
)
# ---------------------------------------------------------------------------
# Agent state definition
# ---------------------------------------------------------------------------
class AdvisorState(TypedDict):
"""The state that flows through the pre-print advisor graph."""
# Input: the slicer settings to analyse.
slicer_settings: dict
# Output: the structured advice report (None until the node runs).
advice_report: Optional[PrePrintAdviceReport]
# ---------------------------------------------------------------------------
# System prompt with X1C-specific knowledge base
# ---------------------------------------------------------------------------
ADVISOR_SYSTEM_PROMPT = """
You are an expert 3D printing advisor specialising in the Bambu Lab X1C.
You have deep knowledge of the following X1C-specific characteristics and
best practices:
FILAMENT-SPECIFIC GUIDANCE:
- PLA: Nozzle 190-220C, Bed 55-65C (textured PEI) or 45-55C (smooth PEI).
Fan speed 80-100%. Prone to stringing above 220C.
- PETG: Nozzle 230-250C, Bed 70-85C. Fan speed 30-50%. Sticks aggressively
to smooth PEI; use textured PEI or release agent. Prone to stringing.
- ABS: Nozzle 240-260C, Bed 90-110C. Fan speed 0-20%. Requires enclosure.
High warping risk without brim on large flat parts.
- TPU: Nozzle 220-240C, Bed 40-60C. Fan speed 30-50%. Print slowly (max
30 mm/s outer wall). Disable retraction or use very short retraction.
- PA (Nylon): Nozzle 260-280C, Bed 70-90C. Must be dried before printing.
Highly hygroscopic; moisture causes bubbling and poor layer adhesion.
SPEED GUIDELINES FOR QUALITY:
- Outer wall: 30-60 mm/s for quality prints. Above 80 mm/s causes ringing.
- Inner wall: 60-120 mm/s acceptable.
- Infill: 100-200 mm/s acceptable for most patterns.
- First layer: 20-40 mm/s for reliable adhesion.
- Bridges: 20-40 mm/s for clean bridge surfaces.
LAYER HEIGHT GUIDELINES:
- 0.08 mm: Ultra-fine detail, very slow, best for miniatures.
- 0.12 mm: High quality, good for visible surfaces.
- 0.20 mm: Standard quality, good balance of speed and appearance.
- 0.28 mm: Draft quality, fast, for functional parts.
- Maximum layer height should not exceed 75% of nozzle diameter.
COOLING RULES:
- First layer fan speed should always be 0% for bed adhesion.
- For overhangs beyond 45 degrees, increase fan speed.
- For small features, set minimum layer time to 10-15 seconds.
- ABS and ASA require minimal or no cooling to prevent warping.
SUPPORT SETTINGS:
- Tree supports are generally preferred for organic models.
- Support Z distance of 0.2 mm for PLA, 0.15 mm for PETG.
- Support interface layers improve surface quality where supports contact model.
INFILL PATTERNS:
- Gyroid: Best for strength-to-weight ratio, no nozzle scraping.
- Grid: Fast, good for flat-topped parts.
- Honeycomb: Good strength, moderate speed.
- 10-20% density is sufficient for most non-structural parts.
- 40-60% for functional parts under moderate load.
COMMON FAILURE MODES AND THEIR SETTINGS CAUSES:
- Spaghetti: Usually caused by print detachment from insufficient bed
temperature, contaminated build plate, or too-fast first layer.
- Warping: Insufficient bed temperature, no brim, too much cooling for
ABS/ASA, printing in a cold environment.
- Stringing: Nozzle temperature too high, retraction distance too short,
travel speed too slow.
- Layer shifting: Print speed too high causing vibrations, loose belts,
or mechanical obstruction.
- Poor layer adhesion: Nozzle temperature too low, layer height too large,
print speed too high.
- Elephant foot: First layer squish too high, bed temperature too high,
cooling insufficient on first layer.
Analyse the provided settings against this knowledge base and identify any
settings that deviate from best practices. Prioritise issues by their
potential impact on print success. Be specific about why each change matters.
Respond with a JSON object matching the PrePrintAdviceReport schema exactly.
"""
# ---------------------------------------------------------------------------
# Graph node functions
# ---------------------------------------------------------------------------
def analyse_settings_node(state: AdvisorState) -> dict:
"""
The main analysis node. Sends the slicer settings to the LLM and
receives a structured advice report in return.
Uses method='json_schema' for reliable structured output with Ollama.
Falls back to a default report if the model produces invalid output,
which can occur with smaller quantized models.
Returns a partial state update containing only the 'advice_report' key.
"""
llm = create_llm()
# Bind the output schema to the LLM using json_schema method for
# reliable structured output via Ollama's native schema enforcement.
# json_schema is the default since langchain-ollama 0.3.0, but we
# pass it explicitly to make the intent clear.
structured_llm = llm.with_structured_output(
PrePrintAdviceReport,
method="json_schema",
)
settings_text = "\n".join(
f" {key}: {value}"
for key, value in state["slicer_settings"].items()
)
messages = [
SystemMessage(content=ADVISOR_SYSTEM_PROMPT),
HumanMessage(
content=(
"Please analyse these Bambu Lab X1C slicer settings and "
f"provide your recommendations:\n\n{settings_text}"
)
),
]
try:
report: PrePrintAdviceReport = structured_llm.invoke(messages)
except Exception as exc:
print(
f"[Advisor] Structured output failed ({type(exc).__name__}: {exc}). "
"Returning fallback report. Consider using a larger model."
)
report = PrePrintAdviceReport(
overall_assessment=(
"Analysis could not be completed due to a model output "
"validation error. Please review your settings manually."
),
issues_found=[],
general_tips=[
"Verify bed temperature is appropriate for your filament type.",
"Check that print speeds are within recommended ranges.",
"Ensure the build plate is clean before printing.",
],
estimated_success_likelihood=(
"unknown — analysis failed; review settings manually"
),
)
# Return only the changed key as a partial state update.
return {"advice_report": report}
# ---------------------------------------------------------------------------
# Graph construction
# ---------------------------------------------------------------------------
def build_pre_print_advisor():
"""
Construct and compile the pre-print advisor LangGraph graph.
Returns:
The compiled LangGraph graph ready for invocation.
"""
graph = StateGraph(AdvisorState)
graph.add_node("analyse_settings", analyse_settings_node)
graph.add_edge(START, "analyse_settings")
graph.add_edge("analyse_settings", END)
return graph.compile()
CHAPTER THIRTEEN: THE MONITORING AGENT
The monitoring agent is the most complex of the three agents because it needs to run continuously, make decisions about what it sees, and trigger alerts when necessary. It is implemented as a LangGraph graph with a loop that continues until the print finishes or a failure is detected.
A critical configuration detail for any LangGraph graph that contains a loop: the recursion_limit. LangGraph counts every node execution as one step toward this limit and raises a GraphRecursionError when the limit is reached. The default limit is 25 steps. Our monitoring graph has four nodes in its main loop: capture_and_analyse, assess_with_llm, decide_action, and wait. Each full monitoring cycle therefore consumes four steps. With the default limit of 25, the graph would exhaust its budget after approximately six monitoring cycles, which at a 60-second interval means the application would crash after roughly six minutes of monitoring — far too short for any real print job. We therefore pass config={"recursion_limit": 10000} when invoking the monitoring agent, which provides capacity for approximately 2,499 full cycles, or over 41 hours of monitoring at 60-second intervals. This is more than sufficient for even the longest print jobs.
The consecutive_issue_count mechanism is one of the most important design decisions in the entire application. A single frame analysed by Moondream2 can produce a false positive for many reasons: a shadow that looks like tangled filament, a reflection on the build plate, a piece of debris that was there before the print started, or simply an ambiguous angle of view. By requiring two consecutive ALERT assessments before triggering a notification, we dramatically reduce the false positive rate. The user will only be disturbed when the AI has seen the same problem in two successive frames separated by the check interval, which is strong evidence that something is genuinely wrong.
The should_continue conditional edge function checks failure_detected before print_complete. This ordering is critical: if the printer itself reports a FAILED gcode_state, both flags are set to True, and we must route to the alert path rather than silently ending the monitoring loop. Checking failure_detected first ensures that printer-reported failures always trigger a user notification.
The capture_and_analyse_node and assess_with_llm_node both include try/except blocks to handle transient inference errors gracefully. If Moondream2 fails for any reason during a particular cycle, the monitoring loop continues rather than crashing. The error is logged and the cycle is treated as if no frame was available. When the LLM assessment fails, the consecutive_issue_count is preserved at its current value rather than being reset to zero, since an LLM failure is not evidence that the print is healthy.
All node functions return partial state updates — only the keys that changed — rather than the entire state. LangGraph merges these partial updates into the existing state using the last-write-wins reducer for all keys in MonitoringState.
# monitoring_agent.py
#
# Implements the continuous print monitoring agent as a LangGraph graph.
# Captures camera frames, analyses them with Moondream2 (encoding the
# image exactly once per cycle), evaluates the results with an LLM,
# and triggers alerts when failures are detected.
#
# Key design decisions:
# - Encodes each frame exactly once and runs all VLM questions against
# the encoded representation to avoid redundant computation.
# - Requires two consecutive ALERT assessments before notifying the user
# to reduce false positives from transient visual artefacts.
# - Checks failure_detected before print_complete in the routing function
# so that printer-reported FAILED states always trigger an alert.
# - All node functions return partial state updates (only changed keys).
# - try/except in capture and LLM nodes prevents transient errors from
# crashing the monitoring loop.
# - On LLM failure, consecutive_issue_count is preserved (not reset)
# because an LLM error is not evidence that the print is healthy.
# - CRITICAL: invoke() must be called with config={"recursion_limit": 10000}
# because the default limit of 25 steps would exhaust after ~6 cycles.
import time
from typing import List, Literal, TypedDict
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import END, START, StateGraph
from pydantic import BaseModel, Field
from bambu_mqtt_client import BambuMQTTClient
from camera_capture import CameraCapture
from llm_setup import create_llm
from notifier import TelegramNotifier
from vision_analyzer import VisionAnalyzer
# ---------------------------------------------------------------------------
# State and schema definitions
# ---------------------------------------------------------------------------
class MonitoringObservation(BaseModel):
"""A single observation made during the monitoring loop."""
timestamp: float = Field(description="Unix timestamp of the observation.")
layer_number: int = Field(description="Current layer at observation time.")
print_progress: int = Field(description="Print completion percentage.")
spaghetti_detected: bool = Field(
description="Whether spaghetti was detected in this frame."
)
detachment_detected: bool = Field(
description="Whether print detachment was detected in this frame."
)
vision_description: str = Field(
description="The VLM's general health description of the print."
)
llm_assessment: str = Field(
description="The LLM's assessment of whether action is needed."
)
alert_triggered: bool = Field(
description="Whether an alert was sent for this observation."
)
class MonitoringState(TypedDict):
"""The state that flows through the monitoring graph."""
# Configuration: how often to check, in seconds.
check_interval_seconds: int
# The current printer state from MQTT.
printer_state: dict
# The latest vision analysis result from Moondream2.
latest_vision_result: dict
# History of all observations made during this print.
observations: List[MonitoringObservation]
# Whether a critical failure has been detected.
failure_detected: bool
# Whether the print has completed (normally or with failure).
print_complete: bool
# Number of consecutive cycles showing a potential issue.
# We require at least 2 before alerting to reduce false positives.
consecutive_issue_count: int
# ---------------------------------------------------------------------------
# Shared resources (injected at graph construction time via closure)
# ---------------------------------------------------------------------------
class MonitoringResources:
"""
Container for the shared resources used by the monitoring agent.
Passed to node functions via closure to avoid global state.
"""
def __init__(
self,
mqtt_client: BambuMQTTClient,
camera: CameraCapture,
vision_analyzer: VisionAnalyzer,
notifier: TelegramNotifier,
) -> None:
self.mqtt_client = mqtt_client
self.camera = camera
self.vision_analyzer = vision_analyzer
self.notifier = notifier
# ---------------------------------------------------------------------------
# LLM assessment system prompt
# ---------------------------------------------------------------------------
_MONITOR_SYSTEM_PROMPT = """You are an AI system monitoring a 3D print in
progress on a Bambu Lab X1C printer. You receive observations from a vision
system and printer telemetry, and you must decide whether the situation
requires immediate user notification.
Be conservative: only recommend alerting the user if there is clear evidence
of a print failure. Normal printing artefacts, shadows, and minor imperfections
should not trigger alerts. A single ambiguous observation should not trigger
an alert; wait for confirmation across multiple observations.
Respond with either 'ALERT' followed by a brief reason, or 'OK' followed by
a brief description of the current print state. Your entire response must
begin with exactly 'ALERT' or 'OK'."""
# ---------------------------------------------------------------------------
# Graph construction function
# ---------------------------------------------------------------------------
def build_monitoring_agent(resources: MonitoringResources):
"""
Construct and compile the monitoring agent LangGraph graph.
Uses closures to give node functions access to shared resources.
IMPORTANT: When invoking the compiled graph, always pass a high
recursion_limit to prevent GraphRecursionError during long prints:
graph.invoke(state, config={"recursion_limit": 10000})
The default limit of 25 steps is exhausted after approximately 6
monitoring cycles, which at 60-second intervals is only 6 minutes.
Args:
resources: The MonitoringResources instance with all dependencies.
Returns:
The compiled LangGraph graph ready for invocation.
"""
def capture_and_analyse_node(state: MonitoringState) -> dict:
"""
Capture a camera frame, encode it once with Moondream2, run all
three analysis questions against the encoded representation, and
return the results as a partial state update.
If the camera has no frame or if Moondream2 raises an exception,
returns an error marker so the LLM assessment node can skip this
cycle gracefully without crashing the monitoring loop.
"""
frame = resources.camera.get_latest_frame()
if frame is None:
print("[Monitor] No frame available from camera.")
return {"latest_vision_result": {"error": "no_frame"}}
print("[Monitor] Analysing frame with Moondream2...")
try:
# analyse_frame() encodes the image once and runs all three
# questions against the encoded representation.
vision_result = resources.vision_analyzer.analyse_frame(frame)
print(
f"[Monitor] Spaghetti: "
f"{vision_result['spaghetti']['detected']}, "
f"Detachment: {vision_result['detachment']['detected']}"
)
return {"latest_vision_result": vision_result}
except Exception as exc:
print(f"[Monitor] Vision analysis failed: {exc}")
return {"latest_vision_result": {"error": str(exc)}}
def assess_with_llm_node(state: MonitoringState) -> dict:
"""
Pass the vision results and printer telemetry to the LLM for a
higher-level assessment of whether action is needed. Records the
observation and updates the consecutive issue counter.
If the vision node produced an error this cycle, returns an empty
partial update (no state changes) and skips the LLM call.
On LLM failure, preserves the existing consecutive_issue_count
rather than resetting it to zero, because an LLM error is not
evidence that the print is healthy.
Returns a partial state update with 'consecutive_issue_count'
and 'observations' when an assessment is made.
"""
vision = state["latest_vision_result"]
# If the camera or VLM produced an error, skip this cycle.
if "error" in vision:
print(
f"[Monitor] Skipping LLM assessment due to vision error: "
f"{vision['error']}"
)
return {}
printer = state["printer_state"]
try:
llm = create_llm()
context = (
f"Printer state: layer {printer.get('layer_num', 'unknown')}, "
f"progress {printer.get('mc_percent', 'unknown')}%, "
f"nozzle temp {printer.get('nozzle_temper', 'unknown')}C, "
f"bed temp {printer.get('bed_temper', 'unknown')}C.\n\n"
"Vision analysis results:\n"
f"- Spaghetti detected: "
f"{vision['spaghetti']['detected']}\n"
f"- Spaghetti description: "
f"{vision['spaghetti']['description']}\n"
f"- Detachment detected: "
f"{vision['detachment']['detected']}\n"
f"- Detachment description: "
f"{vision['detachment']['description']}\n"
f"- General health: {vision['general_health']}\n\n"
f"Consecutive issue count so far: "
f"{state['consecutive_issue_count']}"
)
messages = [
SystemMessage(content=_MONITOR_SYSTEM_PROMPT),
HumanMessage(content=context),
]
response = llm.invoke(messages)
assessment = response.content.strip()
print(f"[Monitor] LLM assessment: {assessment[:80]}")
# Update the consecutive issue counter based on the assessment.
if assessment.upper().startswith("ALERT"):
new_count = state["consecutive_issue_count"] + 1
else:
new_count = 0
except Exception as exc:
# On LLM failure, preserve the existing count rather than
# resetting it. An LLM error is not evidence of a healthy print.
print(
f"[Monitor] LLM assessment failed: {exc}. "
"Preserving existing consecutive issue count."
)
assessment = "OK (LLM assessment unavailable this cycle)"
new_count = state["consecutive_issue_count"]
# Record this observation in the history.
observation = MonitoringObservation(
timestamp=time.time(),
layer_number=int(printer.get("layer_num", 0)),
print_progress=int(printer.get("mc_percent", 0)),
spaghetti_detected=vision["spaghetti"]["detected"],
detachment_detected=vision["detachment"]["detected"],
vision_description=vision["general_health"],
llm_assessment=assessment,
alert_triggered=False,
)
return {
"consecutive_issue_count": new_count,
"observations": state["observations"] + [observation],
}
def decide_action_node(state: MonitoringState) -> dict:
"""
Decide whether to alert, continue monitoring, or end the session.
Checks the printer's reported gcode_state and the consecutive issue
counter. Returns a partial state update with the decision flags.
Important: when the printer reports FAILED, both failure_detected
and print_complete are set to True. The routing function checks
failure_detected first, ensuring an alert is always sent for
printer-reported failures rather than silently ending the session.
"""
printer_state = state["printer_state"]
gcode_state = printer_state.get("gcode_state", "")
if gcode_state == "FINISH":
print("[Monitor] Print completed normally (FINISH).")
return {"print_complete": True, "failure_detected": False}
if gcode_state == "FAILED":
print("[Monitor] Printer reported FAILED state.")
# Set both flags; routing checks failure_detected first to
# ensure the user is always notified of printer-reported failures.
return {"print_complete": True, "failure_detected": True}
# Require two consecutive ALERT assessments before triggering.
if state["consecutive_issue_count"] >= 2:
print(
f"[Monitor] {state['consecutive_issue_count']} consecutive "
"issues detected. Triggering alert."
)
return {"failure_detected": True}
return {}
def send_alert_node(state: MonitoringState) -> dict:
"""
Send a Telegram alert to the user about the detected failure.
Marks the most recent observation as having triggered an alert.
Uses the synchronous TelegramNotifier (safe in a sync node function).
Returns a partial state update with the updated observations list,
or an empty update if there are no observations to mark.
"""
printer = state["printer_state"]
vision = state["latest_vision_result"]
# Determine the failure type from the vision analysis.
if vision.get("spaghetti", {}).get("detected"):
failure_type = "Spaghetti / Filament Tangle"
elif vision.get("detachment", {}).get("detected"):
failure_type = "Print Detachment from Bed"
elif printer.get("gcode_state") == "FAILED":
failure_type = "Printer-Reported Failure"
else:
failure_type = "Unknown Print Anomaly"
# Get the description from the most recent observation if available.
description = "No visual description available."
if state["observations"]:
description = state["observations"][-1].vision_description
resources.notifier.send_print_failure_alert(
failure_type=failure_type,
description=description,
print_progress=int(printer.get("mc_percent", 0)),
layer_number=int(printer.get("layer_num", 0)),
)
# Mark the most recent observation as having triggered an alert.
if not state["observations"]:
return {}
updated_observations = list(state["observations"])
latest = updated_observations[-1]
updated_observations[-1] = MonitoringObservation(
**{**latest.model_dump(), "alert_triggered": True}
)
return {"observations": updated_observations}
def wait_node(state: MonitoringState) -> dict:
"""
Wait for the configured interval before the next check, then
refresh the printer state from MQTT. Returns a partial state
update with the refreshed printer_state.
"""
print(
f"[Monitor] Waiting {state['check_interval_seconds']}s "
"before next check..."
)
time.sleep(state["check_interval_seconds"])
refreshed_state = resources.mqtt_client.get_printer_state()
return {"printer_state": refreshed_state}
def should_continue(
state: MonitoringState,
) -> Literal["send_alert", "end", "wait"]:
"""
Conditional edge function. Determines the next node based on the
current state. Checks failure_detected BEFORE print_complete so
that printer-reported FAILED states always route to the alert path
rather than silently ending the monitoring session.
Returns:
"send_alert" if a failure has been detected.
"end" if the print has completed without failure.
"wait" to continue monitoring.
"""
# Check failure first: a FAILED printer state sets both flags,
# and we must alert the user rather than silently ending.
if state["failure_detected"]:
return "send_alert"
if state["print_complete"]:
return "end"
return "wait"
# Build the graph.
graph = StateGraph(MonitoringState)
graph.add_node("capture_and_analyse", capture_and_analyse_node)
graph.add_node("assess_with_llm", assess_with_llm_node)
graph.add_node("decide_action", decide_action_node)
graph.add_node("send_alert", send_alert_node)
graph.add_node("wait", wait_node)
# Define the linear edges.
graph.add_edge(START, "capture_and_analyse")
graph.add_edge("capture_and_analyse", "assess_with_llm")
graph.add_edge("assess_with_llm", "decide_action")
# Define the conditional edge from decide_action.
graph.add_conditional_edges(
"decide_action",
should_continue,
{
"send_alert": "send_alert",
"end": END,
"wait": "wait",
},
)
# After alerting, end the monitoring session.
graph.add_edge("send_alert", END)
# After waiting, loop back to capture the next frame.
graph.add_edge("wait", "capture_and_analyse")
return graph.compile()
CHAPTER FOURTEEN: THE FAILURE CONSULTANT AGENT
When a print fails, the user is often left wondering what went wrong. The failure consultant agent addresses this by reviewing all available evidence: the print history, the observations recorded during monitoring, the printer state at the time of failure, and the visual description captured near the failure event. It then produces a structured diagnosis with probable causes ranked by likelihood and specific corrective actions for each.
The compile_evidence_node serves an important architectural purpose even though it performs no computation in this prototype. In a more mature version of this application, this node would be the place to enrich the failure evidence with additional context: perhaps querying a local knowledge base of known X1C failure patterns, retrieving the print file to check for problematic geometry, or looking up the filament manufacturer's recommended settings. By having this node in the graph from the beginning, we establish a clear extension point without needing to restructure the graph later.
As with the pre-print advisor, the analyse_failure_node includes a try/except block around the structured output call to handle cases where the model fails to produce a valid response conforming to the FailureDiagnosis schema.
# failure_consultant.py
#
# Implements the post-failure consultation agent as a LangGraph graph.
# Analyses print failure evidence and produces a structured diagnosis
# with probable causes and corrective actions.
# Uses method='json_schema' for reliable structured output with Ollama.
# Includes a fallback diagnosis for model output validation failures.
# Node functions return partial state updates (only changed keys).
from typing import List, Optional, TypedDict
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import END, START, StateGraph
from pydantic import BaseModel, Field
from llm_setup import create_llm
# ---------------------------------------------------------------------------
# Output schema definitions
# ---------------------------------------------------------------------------
class ProbableCause(BaseModel):
"""A single probable cause of the print failure."""
cause: str = Field(
description="A concise description of the probable cause."
)
likelihood: str = Field(
description=(
"The likelihood of this being the actual cause: "
"'high', 'medium', or 'low'."
)
)
evidence: str = Field(
description=(
"The specific evidence from the print history or observations "
"that supports this cause."
)
)
corrective_action: str = Field(
description=(
"A specific, actionable step the user can take to prevent "
"this cause from recurring."
)
)
class FailureDiagnosis(BaseModel):
"""The complete failure diagnosis produced by the consultant agent."""
failure_summary: str = Field(
description="A brief, plain-language summary of what happened."
)
probable_causes: List[ProbableCause] = Field(
description="A ranked list of probable causes, most likely first.",
default_factory=list,
)
immediate_actions: List[str] = Field(
description=(
"Actions the user should take right now, before attempting "
"another print."
),
default_factory=list,
)
prevention_strategy: str = Field(
description=(
"A broader strategy for preventing similar failures "
"in future prints."
)
)
# ---------------------------------------------------------------------------
# State definition
# ---------------------------------------------------------------------------
class ConsultantState(TypedDict):
"""The state that flows through the failure consultant graph."""
# Input: evidence about the failure.
print_settings: dict
printer_state_at_failure: dict
monitoring_observations: List[dict]
vision_description_at_failure: str
# Output: the structured diagnosis (Optional because it starts as None).
diagnosis: Optional[FailureDiagnosis]
# ---------------------------------------------------------------------------
# System prompt for the failure consultant
# ---------------------------------------------------------------------------
_CONSULTANT_SYSTEM_PROMPT = """You are an expert 3D printing failure analyst
specialising in FDM printers, particularly the Bambu Lab X1C. You receive
evidence about a print failure and produce a structured diagnosis.
Your analysis should consider: bed adhesion issues (temperature, surface
condition, first layer settings), filament quality and moisture content,
cooling problems, speed-related issues (vibrations, layer shifting), thermal
issues (heat creep, clogs), mechanical issues (belt tension, calibration),
and slicer setting errors.
Be specific and practical. Users want to understand what went wrong and how
to fix it. Rank your probable causes by likelihood based on the evidence
provided. Respond with a JSON object matching the FailureDiagnosis schema."""
# ---------------------------------------------------------------------------
# Graph node functions
# ---------------------------------------------------------------------------
def compile_evidence_node(state: ConsultantState) -> dict:
"""
Validates and prepares the failure evidence before analysis.
Returns an empty partial update (no state changes needed here).
In future versions, this node would enrich the evidence with data
from a local knowledge base or the print file.
"""
# No state changes in this prototype; return empty partial update.
return {}
def analyse_failure_node(state: ConsultantState) -> dict:
"""
The main analysis node. Sends all available failure evidence to the
LLM and receives a structured diagnosis in return.
Uses method='json_schema' for reliable structured output with Ollama.
Falls back to a default diagnosis if the model produces invalid output.
Returns a partial state update containing only the 'diagnosis' key.
"""
llm = create_llm()
structured_llm = llm.with_structured_output(
FailureDiagnosis,
method="json_schema",
)
# Format the monitoring observations into a readable summary.
obs_lines = []
for i, obs in enumerate(state["monitoring_observations"], 1):
obs_lines.append(
f" Observation {i}: "
f"Layer {obs.get('layer_number', '?')}, "
f"{obs.get('print_progress', '?')}% complete. "
f"Spaghetti: {obs.get('spaghetti_detected', False)}. "
f"Detachment: {obs.get('detachment_detected', False)}. "
f"Description: {obs.get('vision_description', 'N/A')}"
)
obs_summary = (
"\n".join(obs_lines) if obs_lines else " No observations recorded."
)
settings_summary = "\n".join(
f" {k}: {v}" for k, v in state["print_settings"].items()
)
printer_summary = "\n".join(
f" {k}: {v}"
for k, v in state["printer_state_at_failure"].items()
)
evidence_text = (
f"PRINT SETTINGS:\n{settings_summary}\n\n"
f"PRINTER STATE AT FAILURE:\n{printer_summary}\n\n"
f"MONITORING OBSERVATIONS:\n{obs_summary}\n\n"
f"VISION DESCRIPTION AT FAILURE:\n"
f"{state['vision_description_at_failure']}"
)
messages = [
SystemMessage(content=_CONSULTANT_SYSTEM_PROMPT),
HumanMessage(
content=(
"Please analyse this print failure and provide your "
f"diagnosis:\n\n{evidence_text}"
)
),
]
try:
diagnosis: FailureDiagnosis = structured_llm.invoke(messages)
except Exception as exc:
print(
f"[Consultant] Structured output failed "
f"({type(exc).__name__}: {exc}). "
"Returning fallback diagnosis. Consider using a larger model."
)
diagnosis = FailureDiagnosis(
failure_summary=(
"Automated diagnosis could not be completed due to a model "
"output validation error. Please review the monitoring "
"observations manually."
),
probable_causes=[],
immediate_actions=[
"Inspect the build plate for adhesion issues.",
"Check the nozzle for clogs or partial blockages.",
"Review the monitoring observations for visual clues.",
],
prevention_strategy=(
"Review your slicer settings against the X1C best practices "
"guide and consider running the pre-print advisor before "
"your next attempt."
),
)
# Return only the changed key as a partial state update.
return {"diagnosis": diagnosis}
# ---------------------------------------------------------------------------
# Graph construction
# ---------------------------------------------------------------------------
def build_failure_consultant():
"""
Construct and compile the failure consultant LangGraph graph.
Returns:
The compiled LangGraph graph ready for invocation.
"""
graph = StateGraph(ConsultantState)
graph.add_node("compile_evidence", compile_evidence_node)
graph.add_node("analyse_failure", analyse_failure_node)
graph.add_edge(START, "compile_evidence")
graph.add_edge("compile_evidence", "analyse_failure")
graph.add_edge("analyse_failure", END)
return graph.compile()
CHAPTER FIFTEEN: THE MAIN APPLICATION
With all the individual components built, we can now assemble them into a coherent application. The main module is responsible for initialising all components, managing their lifecycle, and providing the entry points for each of the three main features.
All sensitive configuration values are read from environment variables, which are populated from the .env file by python-dotenv. This keeps credentials out of the source code entirely. The load_dotenv() call at the top of the module reads the .env file before any configuration values are accessed. When the application is run under systemd with an EnvironmentFile directive, the variables are already present in the process environment before Python starts, and load_dotenv() will not overwrite them because its default behaviour is to respect existing environment variables.
The run_monitoring() method passes config={"recursion_limit": 10000} to the monitoring agent's invoke() call. This is essential: without it, the LangGraph graph would raise a GraphRecursionError after approximately six monitoring cycles due to the default limit of 25 steps.
The run_monitoring() method returns a boolean indicating whether a failure was detected. The mainblock uses this return value to decide whether to run the failure consultation: consultation is only meaningful after a failure, so we skip it when the print completes normally.
The example slicer settings in the main block contain deliberate issues that the pre-print advisor should catch. The bed temperature of 35 degrees Celsius is too low for PLA on the X1C's textured PEI plate, which typically requires 55 to 65 degrees for good adhesion. The outer wall speed of 100 mm/s is also quite high and may cause vibration artefacts. The advisor agent should identify both of these issues and explain why they matter.
# main.py
#
# The main application entry point for the 3D Print AI Agent.
# Reads configuration from the .env file via python-dotenv.
# Initialises all components and provides three main workflows:
# 1. Pre-print settings analysis
# 2. Continuous print monitoring
# 3. Post-failure consultation
#
# run_monitoring() passes config={"recursion_limit": 10000} to the
# LangGraph invoke() call. This is essential: the default limit of 25
# steps would exhaust after ~6 monitoring cycles (~6 minutes at 60s
# intervals), causing a GraphRecursionError during normal operation.
#
# run_monitoring() returns True if a failure was detected, False if the
# print completed normally. The __main__ block uses this to decide whether
# to invoke the failure consultant.
#
# IMPORTANT: This is a first prototype. Error handling, configuration
# management, and robustness improvements are needed before production use.
import json
import os
import time
from dotenv import load_dotenv
from bambu_mqtt_client import BambuMQTTClient
from camera_capture import CameraCapture
from failure_consultant import build_failure_consultant
from monitoring_agent import MonitoringResources, build_monitoring_agent
from notifier import TelegramNotifier
from pre_print_advisor import build_pre_print_advisor
from vision_analyzer import VisionAnalyzer
# Load environment variables from .env before reading any config values.
# When running under systemd with EnvironmentFile=, variables are already
# in the environment and load_dotenv() will not overwrite them (default
# behaviour is to respect existing environment variables).
load_dotenv()
# ---------------------------------------------------------------------------
# Configuration (read from environment variables populated by .env)
# ---------------------------------------------------------------------------
PRINTER_IP = os.environ["PRINTER_IP"]
PRINTER_SERIAL = os.environ["PRINTER_SERIAL"]
PRINTER_ACCESS_CODE = os.environ["PRINTER_ACCESS_CODE"]
TELEGRAM_BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
TELEGRAM_CHAT_ID = os.environ["TELEGRAM_CHAT_ID"]
CAMERA_INDEX = int(os.environ.get("CAMERA_INDEX", "0"))
MONITORING_INTERVAL_SECONDS = int(os.environ.get("MONITORING_INTERVAL", "60"))
# ---------------------------------------------------------------------------
# Application class
# ---------------------------------------------------------------------------
class PrintAgentApplication:
"""
The top-level application class. Manages the lifecycle of all
components and provides the three main agent workflows.
"""
def __init__(self) -> None:
"""Initialise all application components."""
print("[App] Initialising Print Agent Application...")
self._mqtt_client = BambuMQTTClient(
printer_ip=PRINTER_IP,
serial_number=PRINTER_SERIAL,
access_code=PRINTER_ACCESS_CODE,
)
self._camera = CameraCapture(
camera_index=CAMERA_INDEX,
capture_width=1280,
capture_height=720,
)
self._vision_analyzer = VisionAnalyzer()
self._notifier = TelegramNotifier(
bot_token=TELEGRAM_BOT_TOKEN,
chat_id=TELEGRAM_CHAT_ID,
)
# Build the LangGraph agents.
self._pre_print_advisor = build_pre_print_advisor()
self._failure_consultant = build_failure_consultant()
# Store monitoring observations for post-failure consultation.
self._monitoring_history: list = []
print("[App] Initialisation complete.")
def startup(self) -> None:
"""
Start all background services: MQTT connection, camera capture,
and Moondream2 model loading. Call this before running any agent.
"""
print("[App] Starting background services...")
self._mqtt_client.connect()
# Allow time for the MQTT connection to establish and the first
# status message to arrive from the printer.
time.sleep(2)
camera_started = self._camera.start()
if not camera_started:
print(
"[App] WARNING: Camera failed to start. "
"Visual monitoring will be unavailable."
)
# Loading Moondream2 takes 30-90 seconds on the Raspberry Pi 5.
self._vision_analyzer.load_model()
print("[App] All services started successfully.")
def shutdown(self) -> None:
"""Stop all background services and release resources."""
print("[App] Shutting down...")
self._camera.stop()
self._mqtt_client.disconnect()
print("[App] Shutdown complete.")
def run_pre_print_analysis(self, slicer_settings: dict) -> dict:
"""
Run the pre-print advisor agent on the provided slicer settings.
Sends the resulting report to the user via Telegram and returns
the report as a dictionary.
Args:
slicer_settings: A dictionary of slicer setting names and values.
Returns:
The advice report as a dictionary.
"""
print("[App] Running pre-print analysis...")
result = self._pre_print_advisor.invoke(
{
"slicer_settings": slicer_settings,
"advice_report": None,
}
)
report = result["advice_report"]
# Format the report for Telegram.
report_lines = [f"Assessment: {report.overall_assessment}", ""]
for issue in report.issues_found:
report_lines.append(
f"[{issue.priority.upper()}] {issue.setting_name}: "
f"Change from {issue.current_value} to "
f"{issue.recommended_value}. {issue.reason}"
)
report_lines.append("")
report_lines.append(
f"Success likelihood: {report.estimated_success_likelihood}"
)
self._notifier.send_pre_print_advice("\n".join(report_lines))
return report.model_dump()
def run_monitoring(self) -> bool:
"""
Start the continuous print monitoring loop. This method blocks
until the print completes or a failure is detected and alerted.
The monitoring history is stored for post-failure consultation.
The recursion_limit is set to 10000 to support long print jobs.
Without this, the default limit of 25 steps would cause a
GraphRecursionError after approximately 6 monitoring cycles.
Returns:
True if a failure was detected and an alert was sent.
False if the print completed normally.
"""
print("[App] Starting print monitoring...")
resources = MonitoringResources(
mqtt_client=self._mqtt_client,
camera=self._camera,
vision_analyzer=self._vision_analyzer,
notifier=self._notifier,
)
monitoring_agent = build_monitoring_agent(resources)
initial_state = {
"check_interval_seconds": MONITORING_INTERVAL_SECONDS,
"printer_state": self._mqtt_client.get_printer_state(),
"latest_vision_result": {},
"observations": [],
"failure_detected": False,
"print_complete": False,
"consecutive_issue_count": 0,
}
# CRITICAL: Pass recursion_limit=10000 to support long print jobs.
# The default limit of 25 steps is exhausted after ~6 monitoring
# cycles (approximately 6 minutes at a 60-second interval).
# At 10000 steps with 4 nodes per cycle, the graph supports
# approximately 2499 full cycles (~41 hours at 60-second intervals).
final_state = monitoring_agent.invoke(
initial_state,
config={"recursion_limit": 10000},
)
# Store the observations for potential post-failure consultation.
self._monitoring_history = [
obs.model_dump()
for obs in final_state.get("observations", [])
]
failure_detected = final_state.get("failure_detected", False)
if failure_detected:
print("[App] Monitoring ended: failure detected and alert sent.")
else:
print("[App] Monitoring ended: print completed normally.")
return failure_detected
def run_failure_consultation(self, slicer_settings: dict) -> dict:
"""
Run the failure consultant agent using the stored monitoring history.
Should be called after run_monitoring() has returned True.
Sends the diagnosis to the user via Telegram.
Args:
slicer_settings: The slicer settings used for the failed print.
Returns:
The failure diagnosis as a dictionary.
"""
print("[App] Running failure consultation...")
# Extract the vision description from the last recorded observation.
last_vision_desc = "No visual description available."
if self._monitoring_history:
last_obs = self._monitoring_history[-1]
last_vision_desc = last_obs.get(
"vision_description", last_vision_desc
)
result = self._failure_consultant.invoke(
{
"print_settings": slicer_settings,
"printer_state_at_failure": (
self._mqtt_client.get_printer_state()
),
"monitoring_observations": self._monitoring_history,
"vision_description_at_failure": last_vision_desc,
"diagnosis": None,
}
)
diagnosis = result["diagnosis"]
# Format and send the diagnosis via Telegram.
diag_lines = [
f"FAILURE SUMMARY:\n{diagnosis.failure_summary}", ""
]
for i, cause in enumerate(diagnosis.probable_causes, 1):
diag_lines.append(
f"Cause {i} [{cause.likelihood.upper()}]: {cause.cause}"
)
diag_lines.append(f"Evidence: {cause.evidence}")
diag_lines.append(f"Fix: {cause.corrective_action}")
diag_lines.append("")
diag_lines.append(
f"PREVENTION:\n{diagnosis.prevention_strategy}"
)
self._notifier.send_alert(
"<b>FAILURE DIAGNOSIS</b>\n\n" + "\n".join(diag_lines)
)
return diagnosis.model_dump()
# ---------------------------------------------------------------------------
# Entry point with example usage
# ---------------------------------------------------------------------------
if __name__ == "__main__":
# Example slicer settings for a PLA print on the X1C.
# In a real application, these would be parsed from the Bambu Studio
# project file or entered by the user through a simple interface.
# Note: bed_temperature_c=35 and outer_wall_speed_mm_s=100 are
# intentionally suboptimal — the advisor should flag both.
example_settings = {
"filament_type": "PLA",
"layer_height_mm": 0.2,
"outer_wall_speed_mm_s": 100,
"inner_wall_speed_mm_s": 200,
"infill_density_percent": 15,
"infill_pattern": "gyroid",
"nozzle_temperature_c": 220,
"bed_temperature_c": 35,
"cooling_fan_speed_percent": 100,
"support_type": "tree",
"brim_width_mm": 0,
"first_layer_speed_mm_s": 50,
}
app = PrintAgentApplication()
try:
app.startup()
print("\n--- PRE-PRINT ANALYSIS ---")
advice = app.run_pre_print_analysis(example_settings)
print(json.dumps(advice, indent=2))
input(
"\nReview the advice above, adjust your slicer settings if "
"needed, start the print in Bambu Studio, then press Enter "
"to begin monitoring...\n"
)
print("\n--- PRINT MONITORING ---")
failure_occurred = app.run_monitoring()
if failure_occurred:
print("\n--- FAILURE CONSULTATION ---")
diagnosis = app.run_failure_consultation(example_settings)
print(json.dumps(diagnosis, indent=2))
else:
print(
"\nPrint completed successfully. "
"No failure consultation needed."
)
finally:
app.shutdown()
CHAPTER SIXTEEN: LAUNCH SCRIPT AND DEPLOYMENT
Create the launch script in the project directory:
```bash
#!/usr/bin/env bash
# run_print_agent.sh
#
# Launch script for the 3D Print AI Agent.
# Activates the virtual environment and runs the main application.
# The .env file in the project directory is loaded automatically by
# python-dotenv inside main.py; no export statements are needed here.
#
# Usage:
# cd ~/print_agent
# chmod +x run_print_agent.sh
# ./run_print_agent.sh
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
VENV_DIR="$HOME/print_agent_env"
# Verify the virtual environment exists.
if [ ! -f "$VENV_DIR/bin/activate" ]; then
echo "ERROR: Virtual environment not found at $VENV_DIR"
echo "Run the setup steps in Chapter Four first."
exit 1
fi
# Activate the virtual environment.
source "$VENV_DIR/bin/activate"
# Verify Ollama is running.
if ! curl -sf http://localhost:11434/api/tags > /dev/null 2>&1; then
echo "WARNING: Ollama does not appear to be running."
echo "Start it with: ollama serve"
fi
# Change to the project directory and run the application.
cd "$SCRIPT_DIR"
echo "Starting 3D Print AI Agent..."
python3 main.py
Make the script executable:
chmod +x ~/print_agent/run_print_agent.sh
For unattended operation, you can run the application as a systemd service so that it starts automatically on boot. Create the service unit file:
```ini
# print_agent.service
# systemd service unit for the 3D Print AI Agent.
#
# Installation:
# sudo cp ~/print_agent/print_agent.service /etc/systemd/system/
# sudo systemctl daemon-reload
# sudo systemctl enable print_agent
# sudo systemctl start print_agent
#
# Monitoring:
# sudo systemctl status print_agent
# journalctl -u print_agent -f
#
# Note: Replace 'pi' with your actual username if you configured a
# different default user during Raspberry Pi OS setup.
[Unit]
Description=3D Print AI Agent
After=network-online.target ollama.service
Wants=network-online.target
[Service]
Type=simple
User=pi
WorkingDirectory=/home/pi/print_agent
ExecStart=/home/pi/print_agent_env/bin/python3 /home/pi/print_agent/main.py
Restart=on-failure
RestartSec=10
StandardOutput=journal
StandardError=journal
# Load environment variables from the .env file.
# systemd EnvironmentFile supports KEY=VALUE pairs and # comments.
# python-dotenv's load_dotenv() will not overwrite variables already
# present in the environment (default behaviour respects existing vars),
# so both mechanisms coexist safely: systemd sets the variables first,
# and load_dotenv() skips them.
EnvironmentFile=/home/pi/print_agent/.env
[Install]
WantedBy=multi-user.target
Install and start the service:
sudo cp ~/print_agent/print_agent.service /etc/systemd/system/
sudo systemctl daemon-reload
sudo systemctl enable print_agent
sudo systemctl start print_agent
Check the service status:
sudo systemctl status print_agent
View live logs:
journalctl -u print_agent -f
CHAPTER SEVENTEEN: A WORKED EXAMPLE OF THE PRE-PRINT ADVISOR IN ACTION
To make the pre-print advisor's behaviour concrete, let us trace through what happens when it analyses the example settings from the main module. The advisor receives the settings dictionary, formats it into a prompt, and sends it to the LLM with the system prompt that establishes the advisor's role and expertise.
The LLM, running as Qwen2.5:1.5b through Ollama, reasons about the settings and produces a response that LangChain parses into a PrePrintAdviceReport object. On a Raspberry Pi 5, this process takes roughly 30 to 90 seconds depending on the complexity of the analysis and the current system load. This is acceptable for a pre-print check that the user performs once before starting a potentially multi-hour print.
A realistic output from the advisor for the example settings might look like this in the Telegram message:
PRE-PRINT ANALYSIS REPORT
Assessment: The settings have two significant issues that should be
corrected before printing. The bed temperature is too low for reliable
PLA adhesion, and the outer wall speed is too high for good surface
quality.
[CRITICAL] bed_temperature_c: Change from 35 to 60. PLA requires a
bed temperature of 55-65C on the X1C's textured PEI plate for reliable
first-layer adhesion. At 35C, the first layer is likely to detach during
printing, especially for models with small contact areas.
[IMPORTANT] outer_wall_speed_mm_s: Change from 100 to 40. An outer wall
speed of 100 mm/s will cause visible vibration artefacts (ringing) on
the print surface. Reducing to 40 mm/s will significantly improve surface
finish without substantially increasing print time.
[OPTIONAL] first_layer_speed_mm_s: Change from 50 to 30. For better
first-layer adhesion, a slower first layer speed of 30 mm/s allows the
filament more time to bond with the build plate surface.
Success likelihood: medium - After correcting the critical bed temperature
issue, success likelihood rises to high.
This kind of specific, prioritised, actionable advice is exactly what a knowledgeable friend with 3D printing expertise would tell you. The LLM brings that expertise to every print, available at any time of day or night, running entirely on your local network without sending any data to the cloud.
CHAPTER EIGHTEEN: RUNNING THE APPLICATION AND WHAT TO EXPECT
Ensure you are in the project directory and the virtual environment is active:
cd ~/print_agent
source ~/print_agent_env/bin/activate
Verify that Ollama is running and the model is available:
ollama list
If Ollama is not running, start it:
ollama serve &
Run the application:
./run_print_agent.sh
When you run the application for the first time, you will see a sequence of startup messages as each component initialises. The Moondream2 model loading step takes the longest, typically 30 to 90 seconds on the Pi 5, and you will see a confirmation message when it completes. The MQTT connection to the X1C establishes quickly, usually within a second or two, and you will see a confirmation message showing the topic it has subscribed to.
The pre-print analysis runs first and produces its report within 30 to 90 seconds. The report is both printed to the console and sent to your Telegram account. After reviewing the report, you can adjust your slicer settings accordingly, start the print in Bambu Studio, and then press Enter in the terminal to begin monitoring.
During monitoring, the application captures a frame every 60 seconds (or whatever interval you configured), analyses it with Moondream2, and then passes the results to the LLM for a higher-level assessment. You will see log messages for each cycle showing what the vision system observed. If everything looks normal, the cycle repeats. If a problem is detected in two consecutive cycles, a Telegram alert is sent immediately.
When the print completes normally, the monitoring loop exits and you will see a completion message. If you experienced a failure, the failure consultant runs automatically and sends you a detailed diagnosis via Telegram.
CHAPTER NINETEEN: LIMITATIONS, KNOWN ISSUES, AND FUTURE IMPROVEMENTS
This prototype, while functional, has numerous limitations that must be acknowledged honestly. Understanding these limitations is essential for using the system safely and for planning future improvements.
The inference speed on the Raspberry Pi 5 without an AI accelerator is the most significant practical limitation. Moondream2 takes 20 to 90 seconds to analyse a single frame, and the LLM adds another 30 to 90 seconds on top of that. This means each monitoring cycle can take up to three minutes from frame capture to assessment completion. For most print failures, this is adequate, but a very rapid failure that develops in under three minutes might not be caught in time. The Raspberry Pi AI HAT+ 2 with its Hailo-10H accelerator would dramatically improve this, potentially reducing the cycle time to under 30 seconds.
Structured output validation errors are an occasional reality when using small quantized models with Pydantic schemas. The advisor and consultant agents include fallback responses for these cases, so the application will not crash, but the fallback responses are generic rather than specific to your settings. If you find validation errors occurring frequently, switch to qwen2.5:3b or llama3.2:3b, which are more reliable at schema adherence while still running acceptably on the Pi 5.
The MQTT integration with the X1C is based on the printer's unofficial local API, which Bambu Lab has not formally documented. While the community has reverse-engineered the protocol and it works reliably, Bambu Lab could change it in a firmware update. The X1C reached end-of-life for manufacturing in March 2026 and will receive firmware updates until May 2027, so this risk is real but limited. Monitor the Bambu Lab community forums and the pybambu library repository for any changes.
The Moondream2 model, while impressive for its size, is not infallible. It can produce false positives, particularly in low-light conditions or when the camera angle creates ambiguous shadows. It can also produce false negatives for subtle failures like very slight warping or early-stage detachment. The two- consecutive-detection requirement mitigates false positives but does not eliminate them entirely.
The error handling in this prototype is minimal. Network disconnections, camera failures, model loading errors, and Telegram API failures are all handled with simple print statements and graceful degradation, but a production system would need comprehensive error handling, automatic retry logic, and alerting when the monitoring system itself fails.
The configuration system is rudimentary. Environment variables work for a single-user prototype, but a proper application would have a web interface for setup and secure credential storage.
Future improvements that would significantly enhance this system include: automatic print pausing via MQTT when a failure is detected, which the X1C supports through its command API; a local web dashboard showing the camera feed, current print status, and monitoring history; integration with Bambu Studio's project file format to automatically extract slicer settings rather than requiring manual entry; a local database of print history to allow the consultant agent to identify recurring failure patterns; and support for multiple printers monitored simultaneously.
The most impactful single improvement would be adding the Raspberry Pi AI HAT+ 2. With its Hailo-10H accelerator and 8 GB of dedicated on-board RAM, the monitoring cycle time would drop dramatically, enabling more frequent checks and faster failure detection. The HAT also enables running larger, more capable LLMs locally, which would improve the quality of both the pre-print advice and the failure diagnosis.
CHAPTER TWENTY: PUTTING IT ALL TOGETHER
Let us step back and appreciate what we have built. On a single-board computer the size of a credit card, costing roughly 80 US dollars, we have deployed:
A Vision Language Model with 2 billion parameters that can look at a photograph of a 3D print in progress and answer natural language questions about what it sees, detecting spaghetti, detachment, and other failure modes with no cloud connection required.
A Large Language Model with 1.5 billion parameters that can reason about complex 3D printing settings, identify potential problems before they occur, and diagnose failures after they happen, drawing on a rich knowledge base encoded in its system prompt.
An agentic orchestration layer built with LangGraph 1.0 that coordinates these models in a principled workflow with explicit state management, conditional routing, and structured outputs validated by Pydantic schemas.
A real-time MQTT connection to the Bambu Lab X1C that provides continuous telemetry about the printer's state, including temperatures, progress, and error codes.
A Telegram notification system that delivers formatted, actionable alerts to the user's mobile phone within seconds of a failure being detected, implemented as direct HTTP calls to the Telegram Bot API using the requests library — clean, synchronous, and free of event loop complexity.
All of this runs locally, privately, and without any subscription fees. The only ongoing cost is electricity, and the Raspberry Pi 5 draws roughly 5 to 8 watts under typical load.
This is the power of the current generation of small, efficient AI models combined with capable edge hardware. The Raspberry Pi 5 is not a toy for this application; it is a genuinely capable AI inference platform that happens to be inexpensive, power-efficient, and widely available. The models we have used, Qwen2.5:1.5b and Moondream2, are not compromises forced by hardware limitations; they are purpose-built for exactly this kind of edge deployment and they perform their roles admirably.
The agentic architecture built with LangGraph is equally significant. By modelling the application as a graph of cooperating agents rather than a monolithic script, we have created a system that is easy to understand, easy to extend, and easy to debug. Each node in the graph has a single responsibility. Each edge represents a clear decision. The state flows through the graph in a transparent, traceable way. When something goes wrong, and in a prototype something always does, you can inspect the state at each node and understand exactly what happened.
This is a first prototype. It needs further testing and error corrections before it is ready for unattended operation. But it is a solid foundation, and the path from here to a production-ready system is clear. Each limitation identified in Chapter Nineteen points to a specific, achievable improvement. The architecture supports those improvements without requiring fundamental redesign.
Start with the hardware. Get a Raspberry Pi 5 with 8 GB of RAM, add active cooling, install an NVMe SSD, and mount a USB camera with a clear view of your X1C's build plate. Follow the setup instructions in this tutorial. Run the application on a test print first, something you would not mind failing, and observe how the system behaves. Adjust the monitoring interval, the detection thresholds, and the system prompts based on what you observe. Iterate.
That is, after all, how all good software is built: one prototype at a time, each one a little better than the last.
APPENDIX: QUICK REFERENCE
PROJECT FILES SUMMARY
bambu_mqtt_client.py MQTT client for X1C (paho-mqtt 2.0, VERSION2)
camera_capture.py Thread-safe USB camera frame capture
vision_analyzer.py Moondream2 via HuggingFace Transformers (CPU)
notifier.py Telegram notifications via requests (synchronous)
llm_setup.py Ollama LLM configuration (Qwen2.5:1.5b)
pre_print_advisor.py Pre-print settings advisor LangGraph agent
monitoring_agent.py Continuous monitoring LangGraph agent
failure_consultant.py Post-failure diagnosis LangGraph agent
main.py Application entry point
requirements.txt Pinned Python dependencies
.env Secret credentials (never commit)
run_print_agent.sh Launch script
print_agent.service systemd service unit
KEY DEPENDENCIES AND VERSIONS (MID-2026)
Python: 3.11+ (required by LangGraph 1.0)
langchain: 1.0.0
langgraph: 1.0.0
langchain-ollama: latest compatible with langchain 1.0
transformers: 4.46.0 (for Moondream2 HuggingFace path)
torch: 2.3.0+ (ARM64 CPU wheel via PyTorch index)
accelerate: 0.30.0+
paho-mqtt: 2.0+ (CallbackAPIVersion.VERSION2 required)
requests: 2.31.0+ (Telegram Bot API HTTP calls)
pydantic: 2.0+
opencv-python-headless: 4.9+
pillow: 10.0+
numpy: 1.26+
python-dotenv: 1.0+
Ollama: latest stable (ARM64 supported)
Model: qwen2.5:1.5b (via Ollama)
VLM: vikhyatk/moondream2 @ 2025-06-21 (HuggingFace)
OS: Raspberry Pi OS Bookworm 64-bit
PYTORCH INSTALLATION (ARM64 — CRITICAL)
Install PyTorch BEFORE pip install -r requirements.txt:
pip install torch torchvision \
--index-url https://download.pytorch.org/whl/cpu
System prerequisites:
sudo apt install -y libopenblas-dev libopenmpi-dev libomp-dev
LANGGRAPH RECURSION LIMIT (CRITICAL)
Default recursion_limit: 25 steps (exhausted after ~6 monitoring cycles)
Required for monitoring: config={"recursion_limit": 10000}
At 60s interval / 4 nodes per cycle: supports ~2499 cycles (~41 hours)
Set in main.py run_monitoring() via monitoring_agent.invoke(state, config=...)
BAMBU LAB X1C MQTT REFERENCE
Broker port: 8883 (TLS 1.2, self-signed certificate)
Username: bblp
Password: LAN Access Code (from printer settings)
Status topic: device/<serial>/report
Command topic: device/<serial>/request
Status update rate: 0.5-2 seconds during active printing
Confirmed gcode_state: IDLE, RUNNING, PAUSE, FINISH, FAILED
Related field: gcode_file_prepare_percent (preparation progress)
Note: API is unofficial and may change with firmware
updates. Bambu Lab has not formally documented it.
MOONDREAM2 ON RASPBERRY PI 5 (CPU INFERENCE)
Loading time: 30-90 seconds
RAM required: ~4 GB (2B model)
Image size: 512x512 pixels recommended
Encoding time: 8-25 seconds per image
Per-question time: 5-15 seconds per question
Full cycle (3 questions): 25-70 seconds
Inference path: HuggingFace Transformers (not Photon)
Photon engine: NOT supported on ARM64 (requires NVIDIA/Apple)
torch.no_grad(): Used throughout to reduce memory and improve speed
tokenizer parameter: Passed as keyword arg: tokenizer=self._tokenizer
TELEGRAM NOTIFICATION APPROACH
Library: requests (direct HTTP POST to Bot API)
Reason: python-telegram-bot v20+ requires Bot as async
context manager for httpx initialisation;
mixing with sync LangGraph nodes adds complexity
Endpoint: https://api.telegram.org/bot{token}/sendMessage
Parse mode: HTML (<b>, <i>, <code> tags supported)
Message limit: 4096 chars; auto-truncated at 4000 with notice
Timeout: 10 seconds per request
STRUCTURED OUTPUT TROUBLESHOOTING
Method used: json_schema (Ollama native schema enforcement)
Default since langchain-ollama 0.3.0; passed
explicitly for clarity and future-proofing
Fallback: Generic response returned on validation error
If errors are frequent: Switch to qwen2.5:3b or llama3.2:3b
Temperature: 0.1 (low, for deterministic schema adherence)
CAMERA TROUBLESHOOTING
List devices: v4l2-ctl --list-devices
Test capture: v4l2-ctl --device=/dev/video0 --all
Default index: 0 (first USB camera)
Recommended camera: Logitech C920 or similar USB webcam 
No comments:
Post a Comment