Realtime Suspicious Activity Detection

Introduction

30 Second Summary

Security cameras are everywhere, but most of them just record passively. They capture footage that someone reviews after an incident has already happened. What if a camera could actually think about what it sees and decide how to respond in real time?

In this project, you will build a real-time surveillance system that detects human body poses from video, classifies activities as normal or suspicious, and forwards suspicious detections to an AI agent that reasons about what to do next.

What You'll Build

A complete surveillance intelligence system that combines computer vision with agentic AI to detect, classify, and reason about human activities in real-time video.

By the end of this project, you'll have:

• A real-time pose estimation pipeline using YOLOv8 that detects 17 body keypoints per person from webcam or video input.
• A trained activity classifier that distinguishes normal behavior from suspicious actions like fighting, shoplifting, or falling.
• An AI agent in Langflow powered by Google Gemini that autonomously assesses threats and decides on appropriate actions.
• A Flask monitoring dashboard displaying live annotated video, activity classifications, and agent decisions.
• Secret Mission: Add multi-person tracking with persistent IDs and a threat escalation system that responds more aggressively to repeat offenders.

Are there any prerequisites?

This project assumes intermediate Python knowledge and a basic understanding of machine learning concepts. You will also need a Google account to get a free Gemini API key.

Set Up Your Configuration

Your development environment is ready with all the packages installed and Langflow running. Before you start building the detection pipeline, you need a central place for all the settings your system will use.

A configuration file keeps magic numbers out of your code. It means you can adjust thresholds, swap models, or change API endpoints without hunting through multiple files.

In this step, get ready to:

• Create a centralized configuration file for all system settings.
• Define detection thresholds, activity labels, and API connection details.

Create the configuration file

Your surveillance system has many moving parts: a pose model, a classifier, Langflow, and a Flask dashboard. Each component needs specific settings like confidence thresholds and API URLs.

• Create a new file called config.py in your project folder.
• Paste the following code into config.py:

import os

# Gemini API Configuration
GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY", "your-api-key-here")

# Langflow Configuration
LANGFLOW_BASE_URL = "http://127.0.0.1:7860"
LANGFLOW_FLOW_ID = "your-flow-id-here"

# Detection Configuration
POSE_MODEL = "yolov8n-pose.pt"
CONFIDENCE_THRESHOLD = 0.5
SUSPICIOUS_THRESHOLD = 0.7
CONSECUTIVE_FRAMES_REQUIRED = 5

# Activity Labels
ACTIVITIES = ["walking", "standing", "running", "fighting", "shoplifting", "falling"]
SUSPICIOUS_ACTIVITIES = ["fighting", "shoplifting", "falling"]

# Flask Configuration
FLASK_HOST = "0.0.0.0"
FLASK_PORT = 5000

What does this code do?

This file defines every tunable setting for your system in one place.

• GEMINI_API_KEY reads from your environment variable so you never hardcode secrets into source files.
• LANGFLOW_BASE_URL and LANGFLOW_FLOW_ID tell your Python code where to send suspicious activity reports. You will update the flow ID once you build your Langflow agent later.
• CONFIDENCE_THRESHOLD (0.5) filters out low-confidence keypoint detections. SUSPICIOUS_THRESHOLD (0.7) means the classifier must be at least 70% confident before flagging an activity.
• CONSECUTIVE_FRAMES_REQUIRED (5) prevents false alarms by requiring suspicious classification for 5 frames in a row before notifying the agent.
• ACTIVITIES lists all recognizable actions. SUSPICIOUS_ACTIVITIES is the subset that triggers the AI agent.

• Save config.py.
• Verify the file loads without errors by running this command in your terminal:

python -c "import config; print(config.ACTIVITIES)"

You should see the list ['walking', 'standing', 'running', 'fighting', 'shoplifting', 'falling'] printed in your terminal.

Seeing a ModuleNotFoundError?

• Make sure your terminal is in the same folder where you saved config.py. Run ls (or dir on Windows) to confirm config.py is listed.
• Check that you named the file exactly config.py (not Config.py or configuration.py).
• If you see a syntax error, double-check that all strings use matching quotes and the list brackets are closed.

Why can't Python find my config file?

Why read the API key from an environment variable?

Hardcoding API keys directly in source files is a security risk. If you ever share your code or push it to GitHub, anyone could see your key. Reading from an environment variable keeps the secret outside your codebase.

✔️ Awesome, I've got everything!

Great. Double check you have saved config.py and can import it successfully.

ⓧ I'd like to double check the full code

Here is the complete config.py file. Make sure yours matches exactly:

import os

# Gemini API Configuration
GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY", "your-api-key-here")

# Langflow Configuration
LANGFLOW_BASE_URL = "http://127.0.0.1:7860"
LANGFLOW_FLOW_ID = "your-flow-id-here"

# Detection Configuration
POSE_MODEL = "yolov8n-pose.pt"
CONFIDENCE_THRESHOLD = 0.5
SUSPICIOUS_THRESHOLD = 0.7
CONSECUTIVE_FRAMES_REQUIRED = 5

# Activity Labels
ACTIVITIES = ["walking", "standing", "running", "fighting", "shoplifting", "falling"]
SUSPICIOUS_ACTIVITIES = ["fighting", "shoplifting", "falling"]

# Flask Configuration
FLASK_HOST = "0.0.0.0"
FLASK_PORT = 5000

Your configuration is locked in. Every module you build from here will import these constants, so changes propagate automatically. Next up, you will build the pose estimation pipeline that detects human keypoints in real time.

Build the Pose Estimation Pipeline

Your environment is ready. Now it's time to build the core of your surveillance system: understanding where people's bodies are in each video frame.

YOLOv8's pose estimation model detects 17 body keypoints per person (nose, eyes, ears, shoulders, elbows, wrists, hips, knees, and ankles). In this step, you'll build a PoseDetector class that loads this model, extracts keypoints from video frames, and draws them as a visible skeleton overlay.

In this step, get ready to:

• Load the YOLOv8 nano pose model and configure it for inference.
• Build a detection method that extracts keypoints and bounding boxes from each frame.
• Draw skeleton connections on frames and verify the pipeline with live video.

Create the PoseDetector class

Pose estimation works by running an image through a neural network that outputs coordinates for each body joint it finds. You'll wrap this logic in a reusable class that the rest of your project can call.

• Create a new file called pose_detector.py in your project folder (the same folder that contains config.py).
• Add the imports and class initialization by pasting this code into the file:

import cv2
import numpy as np
from ultralytics import YOLO
from config import POSE_MODEL, CONFIDENCE_THRESHOLD


class PoseDetector:
    def __init__(self):
        # Load the YOLOv8 nano pose model
        self.model = YOLO(POSE_MODEL)
        # Define which keypoints connect to form the skeleton
        self.keypoint_connections = [
            (0, 1), (0, 2), (1, 3), (2, 4),
            (5, 6), (5, 7), (7, 9), (6, 8), (8, 10),
            (5, 11), (6, 12), (11, 12),
            (11, 13), (13, 15), (12, 14), (14, 16)
        ]

What does this code do?

YOLO(POSE_MODEL) loads the yolov8n-pose.pt model. The first time you run this, Ultralytics automatically downloads the model weights (about 6 MB).

The keypoint_connections list defines 16 bone connections between the 17 keypoints. Each tuple like (5, 7) means "draw a line from keypoint 5 (left shoulder) to keypoint 7 (left elbow)." This creates the visible skeleton overlay later.

• Save pose_detector.py.

Add the detection method

The detection method takes a single video frame, passes it through the YOLO model, and returns structured data about every person it finds.

• In pose_detector.py, add the following detect method inside the PoseDetector class, below the __init__ method:

    def detect(self, frame):
        # Run pose estimation on the frame
        results = self.model(frame, conf=CONFIDENCE_THRESHOLD, verbose=False)
        detections = []

        for result in results:
            if result.keypoints is not None and result.keypoints.xy is not None:
                # Extract keypoints as numpy array with shape [N, 17, 3]
                keypoints = result.keypoints.data.cpu().numpy()
                boxes = result.boxes.xyxy.cpu().numpy() if result.boxes is not None else []

                # Build a detection dict for each person found
                for i, kpts in enumerate(keypoints):
                    bbox = boxes[i] if i < len(boxes) else None
                    detections.append({
                        "keypoints": kpts,
                        "bbox": bbox
                    })

        return detections

What does this method return?

For each person detected, you get a dictionary with two keys:

• keypoints contains a numpy array of shape (17, 3). Each row is one body joint with [x, y, confidence].
• bbox contains the bounding box coordinates [x1, y1, x2, y2] around the person.

The .cpu().numpy() call converts the model output from a PyTorch tensor to a regular NumPy array so the rest of your code can work with it easily.

• Save pose_detector.py.

Seeing an import error?

If you see ModuleNotFoundError: No module named 'ultralytics', double check that you installed the packages in Step 1. Run pip install ultralytics again to fix it.

If you see ModuleNotFoundError: No module named 'config', make sure pose_detector.py is in the same folder as config.py.

Help me fix my import error in pose_detector.py

Add the skeleton drawing method

Raw keypoint data is useful for classification, but to verify your pipeline visually you need to draw the skeleton on the video frame. This method renders green dots at each joint, yellow lines between connected joints, and a blue bounding box around each person.

• In pose_detector.py, add the draw_skeleton method below the detect method:

    def draw_skeleton(self, frame, detections):
        annotated = frame.copy()

        for det in detections:
            kpts = det["keypoints"]

            # Draw a green circle at each detected keypoint
            for idx in range(len(kpts)):
                x, y, conf = kpts[idx]
                if conf > CONFIDENCE_THRESHOLD:
                    cv2.circle(annotated, (int(x), int(y)), 4, (0, 255, 0), -1)

            # Draw yellow lines between connected keypoints
            for start, end in self.keypoint_connections:
                if start < len(kpts) and end < len(kpts):
                    x1, y1, c1 = kpts[start]
                    x2, y2, c2 = kpts[end]
                    if c1 > CONFIDENCE_THRESHOLD and c2 > CONFIDENCE_THRESHOLD:
                        cv2.line(annotated, (int(x1), int(y1)), (int(x2), int(y2)), (0, 255, 255), 2)

            # Draw a blue bounding box around the person
            if det["bbox"] is not None:
                x1, y1, x2, y2 = det["bbox"].astype(int)
                cv2.rectangle(annotated, (x1, y1), (x2, y2), (255, 0, 0), 2)

        return annotated

What does this method do?

The method creates a copy of the original frame so the raw data stays untouched. For each person, it loops through all 17 keypoints and draws a green dot wherever confidence exceeds the threshold.

It then draws yellow lines between connected joints using the keypoint_connections list you defined in __init__. The result looks like a stick figure overlaid on the video.

• Save pose_detector.py.

Seeing an indentation error?

Make sure draw_skeleton is indented at the same level as detect (one level inside the class). Python uses indentation to determine what belongs to a class.

Help me fix my indentation in pose_detector.py

✔️ Awesome, I've got everything!

Great. Double check you've saved pose_detector.py before moving on.

ⓧ I'd like to double check the full code

Here is the complete pose_detector.py file. Make sure yours matches:

import cv2
import numpy as np
from ultralytics import YOLO
from config import POSE_MODEL, CONFIDENCE_THRESHOLD


class PoseDetector:
    def __init__(self):
        self.model = YOLO(POSE_MODEL)
        self.keypoint_connections = [
            (0, 1), (0, 2), (1, 3), (2, 4),
            (5, 6), (5, 7), (7, 9), (6, 8), (8, 10),
            (5, 11), (6, 12), (11, 12),
            (11, 13), (13, 15), (12, 14), (14, 16)
        ]

    def detect(self, frame):
        results = self.model(frame, conf=CONFIDENCE_THRESHOLD, verbose=False)
        detections = []

        for result in results:
            if result.keypoints is not None and result.keypoints.xy is not None:
                keypoints = result.keypoints.data.cpu().numpy()
                boxes = result.boxes.xyxy.cpu().numpy() if result.boxes is not None else []

                for i, kpts in enumerate(keypoints):
                    bbox = boxes[i] if i < len(boxes) else None
                    detections.append({
                        "keypoints": kpts,
                        "bbox": bbox
                    })

        return detections

    def draw_skeleton(self, frame, detections):
        annotated = frame.copy()

        for det in detections:
            kpts = det["keypoints"]

            for idx in range(len(kpts)):
                x, y, conf = kpts[idx]
                if conf > CONFIDENCE_THRESHOLD:
                    cv2.circle(annotated, (int(x), int(y)), 4, (0, 255, 0), -1)

            for start, end in self.keypoint_connections:
                if start < len(kpts) and end < len(kpts):
                    x1, y1, c1 = kpts[start]
                    x2, y2, c2 = kpts[end]
                    if c1 > CONFIDENCE_THRESHOLD and c2 > CONFIDENCE_THRESHOLD:
                        cv2.line(annotated, (int(x1), int(y1)), (int(x2), int(y2)), (0, 255, 255), 2)

            if det["bbox"] is not None:
                x1, y1, x2, y2 = det["bbox"].astype(int)
                cv2.rectangle(annotated, (x1, y1), (x2, y2), (255, 0, 0), 2)

        return annotated

Test the pipeline with real-time video

Time to see your pose detector in action. You'll write a quick test script that opens your webcam (or a video file), runs each frame through your PoseDetector, and displays the annotated skeleton in a window.

• Create a new file called test_pose.py in your project folder.
• Add the following test script:

import cv2
from pose_detector import PoseDetector

# Initialize the detector (downloads model on first run)
detector = PoseDetector()

# Open webcam (use 0) or a video file path like "test_video.mp4"
cap = cv2.VideoCapture(0)

print("Press 'q' to quit the video window.")

while True:
    ret, frame = cap.read()
    if not ret:
        break

    # Detect poses and draw the skeleton overlay
    detections = detector.detect(frame)
    annotated = detector.draw_skeleton(frame, detections)

    # Display the annotated frame
    cv2.imshow("Pose Detection Test", annotated)
    if cv2.waitKey(1) & 0xFF == ord("q"):
        break

cap.release()
cv2.destroyAllWindows()

What does this test script do?

cv2.VideoCapture(0) opens your default webcam. If you don't have a webcam, replace 0 with a path to any video file (like "test_video.mp4").

Each frame goes through detect() to find people, then draw_skeleton() to add the visual overlay. The loop runs continuously until you press q.

• Save test_pose.py.
• Run the test script by executing this command in your terminal:

python test_pose.py

What should I see?

The first time you run this, the yolov8n-pose.pt model downloads automatically (about 6 MB). After that, a window titled "Pose Detection Test" opens showing your webcam feed.

When you step into the camera frame, you should see green dots on your joints, yellow lines connecting them into a stick figure, and a blue rectangle around your body. Press q to close the window when you're done testing.

Don't see a video window?

If the window appears black, your webcam might be in use by another application. Close any video calls or camera apps and try again.

If you get an error about not being able to open the video source, you may not have a webcam. Replace cv2.VideoCapture(0) with cv2.VideoCapture("test_video.mp4") and provide any video file.

If you see no skeleton even though you appear on screen, move closer to the camera. The nano model works best when the person takes up a significant portion of the frame.

Help me debug my pose detection test

Your pose estimation pipeline is working. You can see exactly where each person's body parts are in every frame. Next up, you'll train a classifier that looks at these keypoint positions and decides what activity the person is performing.

Create and Train the Activity Classifier

Your pose detector can now see people and extract their body keypoints. The next challenge is turning those raw coordinates into meaningful labels like "walking" or "fighting."

You need a machine learning classifier that takes pose features (joint angles, body proportions, limb positions) and predicts what activity a person is performing. In this step, you will build that classifier using synthetic training data and a Random Forest model from scikit-learn.

In this step, get ready to:

• Generate synthetic training data that defines pose feature vectors for six different activities.
• Build a feature extraction pipeline that computes joint angles, body ratios, and limb positions from raw keypoints.
• Train a Random Forest classifier and save the model to disk.

Set up the classifier class and synthetic data generation

Real surveillance datasets like UCF-Crime require thousands of annotated video clips. For this project, you will generate synthetic pose features instead. Each activity has a distinct "signature" in terms of body proportions and joint positions.

• Create a new file called activity_classifier.py in your project folder.
• Add the imports, class definition, and the synthetic data generation methods by pasting this code:

import numpy as np
import joblib
from sklearn.ensemble import RandomForestClassifier
from config import ACTIVITIES, SUSPICIOUS_ACTIVITIES, SUSPICIOUS_THRESHOLD


class ActivityClassifier:
    def __init__(self, model_path=None):
        if model_path:
            self.model = joblib.load(model_path)
        else:
            self.model = None

    def _generate_training_data(self):
        np.random.seed(42)
        n_samples_per_class = 200
        X = []
        y = []

        for class_idx, activity in enumerate(ACTIVITIES):
            for _ in range(n_samples_per_class):
                features = self._generate_synthetic_features(activity)
                X.append(features)
                y.append(class_idx)

        return np.array(X), np.array(y)

What does this code do?

The class can load a pre-trained model from disk (when you pass a model_path) or start with no model (for training from scratch).

The _generate_training_data method creates 200 synthetic samples per activity class (6 classes = 1,200 total samples). Each sample is a 12-element feature vector representing body proportions. The random seed ensures reproducible results every time you train.

Each activity (walking, standing, running, fighting, shoplifting, falling) has a distinct pattern of body proportions. The next method defines those patterns using statistical distributions.

• Add the _generate_synthetic_features method below _generate_training_data inside the class:

    def _generate_synthetic_features(self, activity):
        if activity == "walking":
            return np.array([
                np.random.normal(1.0, 0.1),
                np.random.normal(0.6, 0.1),
                np.random.normal(0.4, 0.1),
                np.random.normal(0.4, 0.1),
                np.random.normal(0.8, 0.05),
                np.random.normal(0.8, 0.05),
                np.random.normal(0.0, 0.05),
                np.random.normal(-1.5, 0.1),
                np.random.normal(0.3, 0.1),
                np.random.normal(0.3, 0.1),
                np.random.normal(0.1, 0.05),
                np.random.normal(3.5, 0.2),
            ])
        elif activity == "standing":
            return np.array([
                np.random.normal(1.0, 0.05),
                np.random.normal(0.6, 0.05),
                np.random.normal(0.5, 0.05),
                np.random.normal(0.5, 0.05),
                np.random.normal(0.95, 0.03),
                np.random.normal(0.95, 0.03),
                np.random.normal(0.0, 0.02),
                np.random.normal(-1.8, 0.05),
                np.random.normal(0.5, 0.05),
                np.random.normal(0.5, 0.05),
                np.random.normal(0.02, 0.02),
                np.random.normal(3.8, 0.1),
            ])
        elif activity == "running":
            return np.array([
                np.random.normal(1.1, 0.15),
                np.random.normal(0.7, 0.1),
                np.random.normal(0.3, 0.15),
                np.random.normal(0.3, 0.15),
                np.random.normal(0.6, 0.1),
                np.random.normal(0.6, 0.1),
                np.random.normal(0.1, 0.1),
                np.random.normal(-1.3, 0.15),
                np.random.normal(-0.2, 0.15),
                np.random.normal(-0.2, 0.15),
                np.random.normal(0.2, 0.1),
                np.random.normal(3.2, 0.2),
            ])
        elif activity == "fighting":
            return np.array([
                np.random.normal(1.2, 0.2),
                np.random.normal(0.7, 0.15),
                np.random.normal(0.2, 0.15),
                np.random.normal(0.2, 0.15),
                np.random.normal(0.7, 0.1),
                np.random.normal(0.7, 0.1),
                np.random.normal(0.15, 0.1),
                np.random.normal(-1.2, 0.2),
                np.random.normal(-0.5, 0.2),
                np.random.normal(-0.5, 0.2),
                np.random.normal(0.3, 0.15),
                np.random.normal(3.0, 0.3),
            ])
        elif activity == "shoplifting":
            return np.array([
                np.random.normal(0.9, 0.1),
                np.random.normal(0.5, 0.1),
                np.random.normal(0.3, 0.1),
                np.random.normal(0.3, 0.1),
                np.random.normal(0.85, 0.05),
                np.random.normal(0.85, 0.05),
                np.random.normal(0.05, 0.05),
                np.random.normal(-1.0, 0.15),
                np.random.normal(-0.3, 0.2),
                np.random.normal(-0.3, 0.2),
                np.random.normal(0.15, 0.1),
                np.random.normal(3.3, 0.2),
            ])
        elif activity == "falling":
            return np.array([
                np.random.normal(1.3, 0.2),
                np.random.normal(0.8, 0.2),
                np.random.normal(0.5, 0.2),
                np.random.normal(0.5, 0.2),
                np.random.normal(0.5, 0.15),
                np.random.normal(0.5, 0.15),
                np.random.normal(0.3, 0.2),
                np.random.normal(-0.5, 0.3),
                np.random.normal(0.5, 0.3),
                np.random.normal(0.5, 0.3),
                np.random.normal(0.5, 0.2),
                np.random.normal(2.0, 0.5),
            ])
        return np.random.randn(12)

What do these 12 features represent?

Each 12-element array models a different activity's body "signature". The 12 values correspond to: shoulder width, hip width, left arm angle, right arm angle, left knee angle, right knee angle, nose X offset, nose Y offset, left hand height, right hand height, body lean, and height ratio.

For example, "fighting" has wider shoulders (1.2), lower arm angles (0.2, meaning arms raised), and more body lean (0.3). "Standing" has nearly straight knees (0.95) and minimal lean (0.02). These statistical distributions let the classifier learn what each activity "looks like" in feature space.

• Save activity_classifier.py.

Seeing an indentation error?

Make sure _generate_synthetic_features is indented at the same level as _generate_training_data (one level inside the class). Both methods should start with exactly 4 spaces of indentation.

Help me fix my indentation in activity_classifier.py

Build the feature extraction pipeline

The synthetic data defines what each activity looks like in feature space. Now you need a method that converts raw YOLOv8 keypoints (17 points with x, y, confidence) into that same 12-element feature vector.

• In activity_classifier.py, add the extract_features and _calculate_angle methods below __init__ and above _generate_training_data:

    def extract_features(self, keypoints):
        kpts = keypoints[:, :2]
        confidences = keypoints[:, 2]

        if np.sum(confidences > 0.3) < 5:
            return None

        features = []

        left_shoulder = kpts[5]
        right_shoulder = kpts[6]
        left_hip = kpts[11]
        right_hip = kpts[12]
        left_knee = kpts[13]
        right_knee = kpts[14]
        left_ankle = kpts[15]
        right_ankle = kpts[16]
        left_wrist = kpts[9]
        right_wrist = kpts[10]
        nose = kpts[0]

        torso_height = np.linalg.norm(
            ((left_shoulder + right_shoulder) / 2) - ((left_hip + right_hip) / 2)
        )
        if torso_height < 1:
            torso_height = 1

        shoulder_width = np.linalg.norm(left_shoulder - right_shoulder) / torso_height
        features.append(shoulder_width)

        hip_width = np.linalg.norm(left_hip - right_hip) / torso_height
        features.append(hip_width)

        left_arm_angle = self._calculate_angle(left_shoulder, kpts[7], left_wrist)
        right_arm_angle = self._calculate_angle(right_shoulder, kpts[8], right_wrist)
        features.append(left_arm_angle / 180.0)
        features.append(right_arm_angle / 180.0)

        left_knee_angle = self._calculate_angle(left_hip, left_knee, left_ankle)
        right_knee_angle = self._calculate_angle(right_hip, right_knee, right_ankle)
        features.append(left_knee_angle / 180.0)
        features.append(right_knee_angle / 180.0)

        body_center = (left_hip + right_hip) / 2
        nose_offset_x = (nose[0] - body_center[0]) / torso_height
        nose_offset_y = (nose[1] - body_center[1]) / torso_height
        features.append(nose_offset_x)
        features.append(nose_offset_y)

        left_hand_height = (left_wrist[1] - left_shoulder[1]) / torso_height
        right_hand_height = (right_wrist[1] - right_shoulder[1]) / torso_height
        features.append(left_hand_height)
        features.append(right_hand_height)

        body_lean = np.abs(nose[0] - body_center[0]) / torso_height
        features.append(body_lean)

        height_ratio = np.linalg.norm(nose - ((left_ankle + right_ankle) / 2)) / torso_height
        features.append(height_ratio)

        return np.array(features)

    def _calculate_angle(self, point_a, point_b, point_c):
        ba = point_a - point_b
        bc = point_c - point_b
        cosine = np.dot(ba, bc) / (np.linalg.norm(ba) * np.linalg.norm(bc) + 1e-6)
        cosine = np.clip(cosine, -1.0, 1.0)
        angle = np.degrees(np.arccos(cosine))
        return angle

What does the feature extraction do?

• First, it checks if enough keypoints are visible (at least 5 with confidence > 0.3). If too few are visible, the pose is unreliable and it returns None.
• It computes a torso_height value (distance from shoulders to hips) and uses it to normalize all other measurements. This makes features scale-independent, so the same pose at different distances from the camera produces the same features.
• Joint angles are calculated using the cosine rule between three connected keypoints. An angle near 180 degrees means a limb is straight. A lower angle means it is bent.
• The final 12-element vector captures shoulder width, hip width, arm angles, knee angles, head position, hand heights, body lean, and overall height ratio.

• Save activity_classifier.py.

Not sure where to place these methods?

The extract_features and _calculate_angle methods go between __init__ and _generate_training_data. Make sure they are indented one level inside the class (4 spaces). The method order in the final file is: __init__, extract_features, _calculate_angle, classify, train_model, _generate_training_data, _generate_synthetic_features.

Help me understand the correct method order in my ActivityClassifier class

Train the Random Forest classifier and save the model

You have synthetic data and a feature extractor. Now you need two more methods: one that classifies a set of keypoints into an activity, and one that trains the model.

• In activity_classifier.py, add the classify and train_model methods between _calculate_angle and _generate_training_data:

    def classify(self, keypoints):
        features = self.extract_features(keypoints)
        if features is None:
            return "unknown", 0.0

        features = features.reshape(1, -1)
        prediction = self.model.predict(features)[0]
        probabilities = self.model.predict_proba(features)[0]
        confidence = np.max(probabilities)

        activity = ACTIVITIES[prediction]
        is_suspicious = activity in SUSPICIOUS_ACTIVITIES and confidence >= SUSPICIOUS_THRESHOLD

        return activity, confidence, is_suspicious

    def train_model(self, save_path="activity_model.joblib"):
        X, y = self._generate_training_data()
        self.model = RandomForestClassifier(n_estimators=100, random_state=42)
        self.model.fit(X, y)
        joblib.dump(self.model, save_path)
        print(f"Model trained and saved to {save_path}")
        return self.model

What do these methods do?

• classify() takes raw keypoints, extracts features, and returns three values: the predicted activity name, the model's confidence (0 to 1), and whether the activity is suspicious.
• An activity is flagged as suspicious only if it is in the SUSPICIOUS_ACTIVITIES list AND the confidence is at or above SUSPICIOUS_THRESHOLD (0.7). This prevents false positives from low-confidence predictions.
• train_model() generates the synthetic training data, fits a Random Forest with 100 decision trees, and saves the trained model to disk using joblib. The saved file can be loaded later without retraining.

• Save activity_classifier.py.

Now create a simple training script that uses the classifier class to train and save the model.

• Create a new file called train_model.py in your project folder and paste this code:

from activity_classifier import ActivityClassifier

if __name__ == "__main__":
    classifier = ActivityClassifier()
    classifier.train_model(save_path="activity_model.joblib")
    print("Training complete. Model saved as activity_model.joblib")

What does this script do?

This is a standalone script that creates an ActivityClassifier with no pre-trained model, runs the training process, and saves the result as activity_model.joblib. You only need to run this once. After training, the main pipeline will load the saved model instead of retraining every time.

• Save train_model.py.
• Run the training script by executing this command in your terminal:

python train_model.py

• You should see output confirming the model was saved:

Model trained and saved to activity_model.joblib
Training complete. Model saved as activity_model.joblib

• Confirm the model file exists by running:

ls activity_model.joblib

• You should see activity_model.joblib listed in the output.

Seeing a ModuleNotFoundError?

If you see "No module named 'config'" or "No module named 'activity_classifier'", make sure your terminal is in the same directory as your Python files. Run ls to verify you can see config.py, activity_classifier.py, and train_model.py listed.

If you see "No module named 'sklearn'", run pip install scikit-learn to install the package.

Help me debug my train_model.py error

Why synthetic data instead of a real dataset?

Real activity recognition datasets (like UCF-Crime or NTU RGB+D) contain thousands of annotated video clips that require significant storage and preprocessing time. Synthetic data lets you demonstrate the full classification pipeline in minutes.

In a production deployment, you would replace the synthetic generation with real labeled pose data collected from your specific environment.

✔️ Awesome, I've got everything!

Great. Double check you have saved both files and that activity_model.joblib exists in your project folder.

ⓧ I'd like to double check the full code

Here is the complete activity_classifier.py file:

import numpy as np
import joblib
from sklearn.ensemble import RandomForestClassifier
from config import ACTIVITIES, SUSPICIOUS_ACTIVITIES, SUSPICIOUS_THRESHOLD


class ActivityClassifier:
    def __init__(self, model_path=None):
        if model_path:
            self.model = joblib.load(model_path)
        else:
            self.model = None

    def extract_features(self, keypoints):
        kpts = keypoints[:, :2]
        confidences = keypoints[:, 2]

        if np.sum(confidences > 0.3) < 5:
            return None

        features = []

        left_shoulder = kpts[5]
        right_shoulder = kpts[6]
        left_hip = kpts[11]
        right_hip = kpts[12]
        left_knee = kpts[13]
        right_knee = kpts[14]
        left_ankle = kpts[15]
        right_ankle = kpts[16]
        left_wrist = kpts[9]
        right_wrist = kpts[10]
        nose = kpts[0]

        torso_height = np.linalg.norm(
            ((left_shoulder + right_shoulder) / 2) - ((left_hip + right_hip) / 2)
        )
        if torso_height < 1:
            torso_height = 1

        shoulder_width = np.linalg.norm(left_shoulder - right_shoulder) / torso_height
        features.append(shoulder_width)

        hip_width = np.linalg.norm(left_hip - right_hip) / torso_height
        features.append(hip_width)

        left_arm_angle = self._calculate_angle(left_shoulder, kpts[7], left_wrist)
        right_arm_angle = self._calculate_angle(right_shoulder, kpts[8], right_wrist)
        features.append(left_arm_angle / 180.0)
        features.append(right_arm_angle / 180.0)

        left_knee_angle = self._calculate_angle(left_hip, left_knee, left_ankle)
        right_knee_angle = self._calculate_angle(right_hip, right_knee, right_ankle)
        features.append(left_knee_angle / 180.0)
        features.append(right_knee_angle / 180.0)

        body_center = (left_hip + right_hip) / 2
        nose_offset_x = (nose[0] - body_center[0]) / torso_height
        nose_offset_y = (nose[1] - body_center[1]) / torso_height
        features.append(nose_offset_x)
        features.append(nose_offset_y)

        left_hand_height = (left_wrist[1] - left_shoulder[1]) / torso_height
        right_hand_height = (right_wrist[1] - right_shoulder[1]) / torso_height
        features.append(left_hand_height)
        features.append(right_hand_height)

        body_lean = np.abs(nose[0] - body_center[0]) / torso_height
        features.append(body_lean)

        height_ratio = np.linalg.norm(nose - ((left_ankle + right_ankle) / 2)) / torso_height
        features.append(height_ratio)

        return np.array(features)

    def _calculate_angle(self, point_a, point_b, point_c):
        ba = point_a - point_b
        bc = point_c - point_b
        cosine = np.dot(ba, bc) / (np.linalg.norm(ba) * np.linalg.norm(bc) + 1e-6)
        cosine = np.clip(cosine, -1.0, 1.0)
        angle = np.degrees(np.arccos(cosine))
        return angle

    def classify(self, keypoints):
        features = self.extract_features(keypoints)
        if features is None:
            return "unknown", 0.0

        features = features.reshape(1, -1)
        prediction = self.model.predict(features)[0]
        probabilities = self.model.predict_proba(features)[0]
        confidence = np.max(probabilities)

        activity = ACTIVITIES[prediction]
        is_suspicious = activity in SUSPICIOUS_ACTIVITIES and confidence >= SUSPICIOUS_THRESHOLD

        return activity, confidence, is_suspicious

    def train_model(self, save_path="activity_model.joblib"):
        X, y = self._generate_training_data()
        self.model = RandomForestClassifier(n_estimators=100, random_state=42)
        self.model.fit(X, y)
        joblib.dump(self.model, save_path)
        print(f"Model trained and saved to {save_path}")
        return self.model

    def _generate_training_data(self):
        np.random.seed(42)
        n_samples_per_class = 200
        X = []
        y = []

        for class_idx, activity in enumerate(ACTIVITIES):
            for _ in range(n_samples_per_class):
                features = self._generate_synthetic_features(activity)
                X.append(features)
                y.append(class_idx)

        return np.array(X), np.array(y)

    def _generate_synthetic_features(self, activity):
        if activity == "walking":
            return np.array([
                np.random.normal(1.0, 0.1),
                np.random.normal(0.6, 0.1),
                np.random.normal(0.4, 0.1),
                np.random.normal(0.4, 0.1),
                np.random.normal(0.8, 0.05),
                np.random.normal(0.8, 0.05),
                np.random.normal(0.0, 0.05),
                np.random.normal(-1.5, 0.1),
                np.random.normal(0.3, 0.1),
                np.random.normal(0.3, 0.1),
                np.random.normal(0.1, 0.05),
                np.random.normal(3.5, 0.2),
            ])
        elif activity == "standing":
            return np.array([
                np.random.normal(1.0, 0.05),
                np.random.normal(0.6, 0.05),
                np.random.normal(0.5, 0.05),
                np.random.normal(0.5, 0.05),
                np.random.normal(0.95, 0.03),
                np.random.normal(0.95, 0.03),
                np.random.normal(0.0, 0.02),
                np.random.normal(-1.8, 0.05),
                np.random.normal(0.5, 0.05),
                np.random.normal(0.5, 0.05),
                np.random.normal(0.02, 0.02),
                np.random.normal(3.8, 0.1),
            ])
        elif activity == "running":
            return np.array([
                np.random.normal(1.1, 0.15),
                np.random.normal(0.7, 0.1),
                np.random.normal(0.3, 0.15),
                np.random.normal(0.3, 0.15),
                np.random.normal(0.6, 0.1),
                np.random.normal(0.6, 0.1),
                np.random.normal(0.1, 0.1),
                np.random.normal(-1.3, 0.15),
                np.random.normal(-0.2, 0.15),
                np.random.normal(-0.2, 0.15),
                np.random.normal(0.2, 0.1),
                np.random.normal(3.2, 0.2),
            ])
        elif activity == "fighting":
            return np.array([
                np.random.normal(1.2, 0.2),
                np.random.normal(0.7, 0.15),
                np.random.normal(0.2, 0.15),
                np.random.normal(0.2, 0.15),
                np.random.normal(0.7, 0.1),
                np.random.normal(0.7, 0.1),
                np.random.normal(0.15, 0.1),
                np.random.normal(-1.2, 0.2),
                np.random.normal(-0.5, 0.2),
                np.random.normal(-0.5, 0.2),
                np.random.normal(0.3, 0.15),
                np.random.normal(3.0, 0.3),
            ])
        elif activity == "shoplifting":
            return np.array([
                np.random.normal(0.9, 0.1),
                np.random.normal(0.5, 0.1),
                np.random.normal(0.3, 0.1),
                np.random.normal(0.3, 0.1),
                np.random.normal(0.85, 0.05),
                np.random.normal(0.85, 0.05),
                np.random.normal(0.05, 0.05),
                np.random.normal(-1.0, 0.15),
                np.random.normal(-0.3, 0.2),
                np.random.normal(-0.3, 0.2),
                np.random.normal(0.15, 0.1),
                np.random.normal(3.3, 0.2),
            ])
        elif activity == "falling":
            return np.array([
                np.random.normal(1.3, 0.2),
                np.random.normal(0.8, 0.2),
                np.random.normal(0.5, 0.2),
                np.random.normal(0.5, 0.2),
                np.random.normal(0.5, 0.15),
                np.random.normal(0.5, 0.15),
                np.random.normal(0.3, 0.2),
                np.random.normal(-0.5, 0.3),
                np.random.normal(0.5, 0.3),
                np.random.normal(0.5, 0.3),
                np.random.normal(0.5, 0.2),
                np.random.normal(2.0, 0.5),
            ])
        return np.random.randn(12)

And here is the complete train_model.py file:

from activity_classifier import ActivityClassifier

if __name__ == "__main__":
    classifier = ActivityClassifier()
    classifier.train_model(save_path="activity_model.joblib")
    print("Training complete. Model saved as activity_model.joblib")

Your classifier is trained and ready to label activities. Next up, you will connect this classifier to the pose estimation pipeline so detections flow through classification in real time.

Build the Langflow Agent Flow

Your vision pipeline can now detect poses and classify activities. The next piece is a reasoning layer that can think about what those detections mean and decide what to do about them.

In this step, you will build an AI agent inside Langflow that acts as a security analyst. When your Python pipeline detects suspicious activity, it will send a message to this agent. The agent then reasons about the situation and autonomously calls tools to log, assess, or escalate the incident.

In this step, get ready to:

• Create a Langflow agent flow with Chat Input, Agent, and Chat Output components.
• Configure three custom tool components for the agent to use.
• Test the agent in the Playground and grab the flow ID for your Python code.

Create the agent flow and configure Google Gemini

Langflow exposes every flow as a REST API. You will build a flow that receives security alerts as chat messages, reasons about them using Google Gemini, and calls tools to take action.

• Open your browser and navigate to http://127.0.0.1:7860 where Langflow is running.
• Click New Flow in the top right corner.
• Select the Simple Agent template from the list of templates.

Why start from the Simple Agent template?

The Simple Agent template gives you a pre-wired flow with Chat Input, an Agent component, and Chat Output already connected. This saves you from manually wiring the basic structure. You will customize it by swapping the model provider and replacing the default tools with your own security-focused tools.

The template starts with default tools (Calculator and URL). You will remove those and add your own. First, configure the model provider so the Agent can use Google Gemini.

• Click your profile icon in the bottom left corner of Langflow.
• Select Settings, then click Model Providers.
• Find Google in the providers list and click it to expand.
• Paste your Gemini API key into the API Key field.
• Under Language Models, enable gemini-2.5-flash.
• Close the Settings panel to return to your flow.

Now configure the Agent component to use Gemini.

• Click the Agent component in the workspace.
• In the Language Model dropdown, select gemini-2.5-flash under the Google provider.
• In the Agent Instructions field, paste the following system prompt:

You are an AI security analyst monitoring a surveillance system. You receive real-time alerts about suspicious human activities detected by a computer vision system.

When you receive an alert, you must:
1. Assess the threat level (low, medium, high, critical) based on the activity type, confidence score, number of people involved, and duration.
2. Decide on an action: log the incident for records, send an alert to security personnel, or escalate to emergency services.
3. Always use your available tools to take action. Never just respond with text alone.

Guidelines:
- Fighting with high confidence and multiple people = critical threat, escalate immediately
- Shoplifting with moderate confidence = medium threat, send alert
- Falling = could be medical emergency, send alert with medical context
- Single brief detections with low confidence = low threat, log only
- Consider duration: longer incidents are more serious

What does this system prompt do?

This prompt tells the agent how to behave when it receives security alerts. It provides a decision framework that maps activity types and confidence levels to threat levels and actions. The agent uses this reasoning to decide which tool to call.

The tools you create next will be what the agent actually calls to take action. The system prompt guides the decision, and the tools execute it.

• Delete the Calculator and URL components that came with the template by clicking each one and pressing Delete or the backspace key.

Don't see gemini-2.5-flash in the dropdown?

Make sure you saved your API key in Model Providers and enabled the model under Language Models. If the model still does not appear, try refreshing the page.

help me configure Google Gemini as my model provider in Langflow

Add custom tool components

An agentic AI system needs tools to take action in the world. You will create three custom components that the agent can call: one to log incidents, one to assess threat levels, and one to send alerts.

• In the left sidebar, search for Custom Component or find it under the Helpers section.
• Drag a Custom Component onto the workspace.
• Click the Code button on the component to open the code editor.
• Replace all the code in the editor with the following for the log_incident tool:

from langflow.custom import Component
from langflow.io import MessageTextInput, Output
from langflow.schema import Data

class LogIncidentComponent(Component):
    display_name = "Log Incident"
    description = "Logs a security incident to the system records."
    icon = "file-text"
    name = "LogIncidentComponent"

    inputs = [
        MessageTextInput(
            name="incident_details",
            display_name="Incident Details",
            info="Details of the incident to log",
            value="",
            tool_mode=True,
        ),
    ]

    outputs = [
        Output(display_name="Log Result", name="output", method="log_incident"),
    ]

    def log_incident(self) -> Data:
        details = self.incident_details
        result = f"LOGGED: Incident recorded - {details}"
        self.status = result
        return Data(value={"status": "logged", "details": details})

What does this component do?

This creates a Langflow component that inherits from Component. The tool_mode=True parameter on the input is what makes this component usable as a tool by the Agent. When the agent decides to log an incident, it calls this tool and passes the incident details as input.

• Click Check & Save in the code editor to save the component.
• Enable Tool Mode by clicking the toggle in the component's header menu.
• Connect the component's Toolset output (the cyan dot) to the Agent component's Tools input.

Now create the second tool. Drag another Custom Component onto the workspace and paste this code:

from langflow.custom import Component
from langflow.io import MessageTextInput, Output
from langflow.schema import Data

class AssessThreatLevelComponent(Component):
    display_name = "Assess Threat Level"
    description = "Evaluates the threat level of a detected activity and returns a severity rating."
    icon = "shield-alert"
    name = "AssessThreatLevelComponent"

    inputs = [
        MessageTextInput(
            name="activity_report",
            display_name="Activity Report",
            info="The activity report to assess",
            value="",
            tool_mode=True,
        ),
    ]

    outputs = [
        Output(display_name="Threat Assessment", name="output", method="assess_threat"),
    ]

    def assess_threat(self) -> Data:
        report = self.activity_report
        result = f"ASSESSED: Threat evaluation complete for - {report}"
        self.status = result
        return Data(value={"status": "assessed", "report": report})

• Click Check & Save, enable Tool Mode, and connect the Toolset output to the Agent's Tools input.

Create the third and final tool. Drag one more Custom Component and paste this code:

from langflow.custom import Component
from langflow.io import MessageTextInput, Output
from langflow.schema import Data

class SendAlertComponent(Component):
    display_name = "Send Alert"
    description = "Sends a security alert to the monitoring team with urgency level and recommended actions."
    icon = "bell-ring"
    name = "SendAlertComponent"

    inputs = [
        MessageTextInput(
            name="alert_message",
            display_name="Alert Message",
            info="The alert message to send to security",
            value="",
            tool_mode=True,
        ),
    ]

    outputs = [
        Output(display_name="Alert Result", name="output", method="send_alert"),
    ]

    def send_alert(self) -> Data:
        message = self.alert_message
        result = f"ALERT SENT: {message}"
        self.status = result
        return Data(value={"status": "alert_sent", "message": message})

• Click Check & Save, enable Tool Mode, and connect the Toolset output to the Agent's Tools input.

Why three separate tools instead of one?

Each tool represents a distinct action the agent can take. By separating them, the agent can reason about which action is appropriate. For a low-confidence detection it might only call log_incident. For a high-confidence fight it might call assess_threat_level first and then send_alert. This is what makes the system agentic: the AI decides which tools to use based on the situation.

Your flow should now have five components: Chat Input, Agent (with three tools connected), and Chat Output.

Seeing a validation error on a custom component?

Make sure you clicked Check & Save after pasting the code. Langflow validates the Python syntax before saving. Check for any typos, especially in the import lines and class definition.

help me fix my custom component code in Langflow

Test the agent and grab the flow ID

Before connecting this agent to your Python pipeline, verify it works by sending a test message through the Playground.

• Click the Playground button in the top right corner of the workspace.
• In the chat input, type the following test message and press Enter:

SECURITY ALERT: Suspicious activity detected.
Activity Type: fighting
Confidence: 0.85
Number of People: 3
Timestamp: 2026-05-05T14:30:00
Duration: 4.5 seconds
Please assess the threat level and recommend an action.

What should I see?

The agent should respond with a threat assessment and call one or more of your tools. In the Playground, you will see the agent's tool calls displayed alongside its reasoning. It should identify the fighting activity as a high or critical threat and use the assess_threat_level and send_alert tools.

• Verify that the agent calls at least one tool in its response. You should see tool call indicators in the Playground output.
• Try a second test with a lower-severity message to confirm the agent adapts its response:

SECURITY ALERT: Suspicious activity detected.
Activity Type: shoplifting
Confidence: 0.55
Number of People: 1
Timestamp: 2026-05-05T14:35:00
Duration: 1.2 seconds
Please assess the threat level and recommend an action.

For this lower-confidence detection, the agent should choose a less severe action like logging the incident.

Now grab the flow ID so your Python code can call this agent programmatically.

• Look at the URL in your browser's address bar. It follows the pattern http://127.0.0.1:7860/flow/YOUR-FLOW-ID.
• Copy the flow ID portion (the long string of characters after /flow/).
• Open config.py in your project folder.
• Find this line:

LANGFLOW_FLOW_ID = "your-flow-id-here"

• Replace your-flow-id-here with the flow ID you copied from the URL, keeping the quotes. The line should now look like:

LANGFLOW_FLOW_ID = "[[LANGFLOW_FLOW_ID="your flow ID from the URL bar"]]"

• Save config.py.

Where else can I find the flow ID?

You can also find it by clicking the API button in the top right of your flow workspace. Langflow generates code snippets that include your flow ID. The URL bar method is the quickest.

Agent not calling any tools?

Check that Tool Mode is enabled on each custom component (you should see the cyan Toolset output port). Also verify that all three Toolset outputs are connected to the Agent's Tools input. If the connections look correct, try clicking Build Flow before re-opening the Playground.

help me get my Langflow agent to use my custom tools

Your AI agent is built and tested. In the next step, you will write the Python integration code that sends detection data from your vision pipeline to this Langflow agent in real time.

Connect the Vision Pipeline to Langflow

Your Langflow agent can reason about security threats and decide on actions. Right now it lives in isolation, waiting for someone to manually type a message in the Playground.

In this step, you will build the bridge between your vision pipeline and the agent. You will write a Python module that packages detection data into structured alert messages, sends them to Langflow's REST API, and stores the agent's decisions in an event log for later display on the dashboard.

In this step, get ready to:

• Write a message formatter that packages detection data into structured security alerts.
• Send suspicious activity detections to Langflow's REST API endpoint.
• Parse the agent's response and store decisions in an event log.

Update your flow ID in config.py

The Langflow agent module needs to know which flow to call. Your config.py already has a LANGFLOW_FLOW_ID placeholder from your earlier setup.

• Open config.py in your editor.
• Find this line:

LANGFLOW_FLOW_ID = "your-flow-id-here"

• Replace your-flow-id-here with the flow ID you copied from Langflow's URL bar in the previous step, so the line looks like this:

LANGFLOW_FLOW_ID = "[[LANGFLOW_FLOW_ID=\"your-flow-id-from-langflow-url\"]]"

• Save config.py.

Finding your flow ID

Your flow ID is the long string in the Langflow URL when you have your flow open. It looks something like a1b2c3d4-e5f6-7890-abcd-ef1234567890. Copy only the ID portion, not the full URL.

Format detection data into structured messages

When the vision pipeline detects suspicious activity, it produces raw data: activity type, confidence score, number of people detected, and a timestamp. The agent needs this information in a human-readable format to reason about it effectively.

• Create a new file called langflow_agent.py in your project folder.
• Add the imports, class initialization, and message formatter by copying this code:

import requests
import json
from datetime import datetime
from config import LANGFLOW_BASE_URL, LANGFLOW_FLOW_ID


class LangflowAgent:
    def __init__(self):
        # Build the full API URL from config values
        self.api_url = f"{LANGFLOW_BASE_URL}/api/v1/run/{LANGFLOW_FLOW_ID}"
        # Store all agent decisions for the dashboard
        self.event_log = []

    def _format_message(self, activity_data):
        # Package detection data into a structured alert string
        message = (
            f"SECURITY ALERT: Suspicious activity detected.\n"
            f"Activity Type: {activity_data['activity']}\n"
            f"Confidence: {activity_data['confidence']:.2f}\n"
            f"Number of People: {activity_data['num_people']}\n"
            f"Timestamp: {activity_data['timestamp']}\n"
            f"Duration: {activity_data.get('duration', 'N/A')} seconds\n"
            f"Please assess the threat level and recommend an action."
        )
        return message

What does this code do?

The LangflowAgent class is the bridge between your detection pipeline and the AI agent.

• self.api_url constructs the Langflow REST endpoint using your flow ID. This is the URL the module will POST to.
• self.event_log keeps a running list of every interaction with the agent, which the dashboard will display later.
• _format_message takes raw detection data and formats it as a structured security alert that the agent can understand and reason about.

Send detections to the Langflow REST API

Langflow exposes every flow as a REST API at /api/v1/run/{flow_id}. Your module will POST the formatted message to this endpoint and receive the agent's reasoning back as JSON.

• In langflow_agent.py, add the analyze_activity method below the _format_message method, inside the LangflowAgent class:

    def analyze_activity(self, activity_data):
        # Format the raw detection into a message the agent understands
        message = self._format_message(activity_data)

        # Langflow expects this payload structure for chat-based flows
        payload = {
            "input_value": message,
            "output_type": "chat",
            "input_type": "chat"
        }

        try:
            # POST to the Langflow REST API with a 30-second timeout
            response = requests.post(
                self.api_url,
                json=payload,
                timeout=30
            )
            response.raise_for_status()
            result = response.json()

            # Parse the nested response structure
            agent_response = self._extract_response(result)
            # Log this event for the dashboard
            event = {
                "timestamp": datetime.now().isoformat(),
                "input": activity_data,
                "agent_decision": agent_response
            }
            self.event_log.append(event)
            return agent_response

        except requests.exceptions.RequestException as e:
            error_msg = f"Langflow API error: {str(e)}"
            print(error_msg)
            return {"decision": "error", "message": error_msg}

What does this method do?

• The payload uses input_value, output_type, and input_type because Langflow's chat-based flows expect this exact structure.
• The 30-second timeout prevents the pipeline from freezing if Langflow is slow to respond.
• The try/except block catches network errors gracefully so the video pipeline keeps running even if the agent is unreachable.

Parse the agent response and store events

Langflow returns a deeply nested JSON structure. You need a parser that extracts the agent's actual text response from within multiple layers of outputs arrays.

• Add the _extract_response and get_event_log methods below analyze_activity in the LangflowAgent class:

    def _extract_response(self, result):
        # Navigate Langflow's nested JSON to find the agent's text
        try:
            outputs = result.get("outputs", [])
            if outputs:
                first_output = outputs[0]
                results = first_output.get("outputs", [])
                if results:
                    message = results[0].get("results", {}).get("message", {})
                    text = message.get("text", "No response")
                    return {"decision": "analyzed", "message": text}
            return {"decision": "no_output", "message": "Agent returned no output"}
        except (KeyError, IndexError) as e:
            return {"decision": "parse_error", "message": str(e)}

    def get_event_log(self):
        # Return all stored events for the dashboard to display
        return self.event_log

Why is the response so deeply nested?

Langflow's API wraps agent output in multiple layers because a flow can have multiple output components. The path outputs[0].outputs[0].results.message.text navigates to the first output component's first result, then extracts the text.

The fallback returns handle cases where the agent produces no output or the response structure changes unexpectedly.

• Save langflow_agent.py.
• Verify the file exists by running:

ls langflow_agent.py

• You should see langflow_agent.py listed in the output.

Test the connection to your Langflow agent by running a quick script in your terminal:

python -c "from langflow_agent import LangflowAgent; agent = LangflowAgent(); print('API URL:', agent.api_url)"

• You should see your full Langflow API URL printed, confirming the module loads correctly and your flow ID is in place.

Seeing a ModuleNotFoundError?

• Make sure you are running the command from your project folder where langflow_agent.py and config.py both live.
• Check that config.py has both LANGFLOW_BASE_URL and LANGFLOW_FLOW_ID defined.
• Verify that requests is installed by running pip show requests.

Why can't Python find my langflow_agent module?

✔️ Awesome, I've got everything!

Great. Double check you have saved both config.py and langflow_agent.py.

ⓧ I'd like to double check the full code

Here is the complete langflow_agent.py file:

import requests
import json
from datetime import datetime
from config import LANGFLOW_BASE_URL, LANGFLOW_FLOW_ID


class LangflowAgent:
    def __init__(self):
        self.api_url = f"{LANGFLOW_BASE_URL}/api/v1/run/{LANGFLOW_FLOW_ID}"
        self.event_log = []

    def analyze_activity(self, activity_data):
        message = self._format_message(activity_data)

        payload = {
            "input_value": message,
            "output_type": "chat",
            "input_type": "chat"
        }

        try:
            response = requests.post(
                self.api_url,
                json=payload,
                timeout=30
            )
            response.raise_for_status()
            result = response.json()

            agent_response = self._extract_response(result)
            event = {
                "timestamp": datetime.now().isoformat(),
                "input": activity_data,
                "agent_decision": agent_response
            }
            self.event_log.append(event)
            return agent_response

        except requests.exceptions.RequestException as e:
            error_msg = f"Langflow API error: {str(e)}"
            print(error_msg)
            return {"decision": "error", "message": error_msg}

    def _format_message(self, activity_data):
        message = (
            f"SECURITY ALERT: Suspicious activity detected.\n"
            f"Activity Type: {activity_data['activity']}\n"
            f"Confidence: {activity_data['confidence']:.2f}\n"
            f"Number of People: {activity_data['num_people']}\n"
            f"Timestamp: {activity_data['timestamp']}\n"
            f"Duration: {activity_data.get('duration', 'N/A')} seconds\n"
            f"Please assess the threat level and recommend an action."
        )
        return message

    def _extract_response(self, result):
        try:
            outputs = result.get("outputs", [])
            if outputs:
                first_output = outputs[0]
                results = first_output.get("outputs", [])
                if results:
                    message = results[0].get("results", {}).get("message", {})
                    text = message.get("text", "No response")
                    return {"decision": "analyzed", "message": text}
            return {"decision": "no_output", "message": "Agent returned no output"}
        except (KeyError, IndexError) as e:
            return {"decision": "parse_error", "message": str(e)}

    def get_event_log(self):
        return self.event_log

Your vision pipeline now has a direct line to the AI agent. When suspicious activity is detected and confirmed over multiple frames, the system will package the detection data, send it to Langflow, and store the agent's decision. Next up, you will build the Flask dashboard that displays the live video feed and these agent decisions in real time.

Build the Flask Monitoring Dashboard

In the previous step, you connected your vision pipeline to the Langflow agent so suspicious detections trigger AI-powered reasoning. The agent can now analyze threats and recommend actions.

Right now, everything runs silently in the terminal. You need a visual interface where you can watch the live video feed, see pose annotations, and read the agent's decisions in real time. In this step, you will build a Flask web dashboard that ties all the pieces together into a single monitoring screen.

In this step, get ready to:

• Create a Flask app that serves annotated video frames as an MJPEG stream.
• Build a dark-themed HTML dashboard with a live video panel and an agent decisions log.
• Run the full system and verify the dashboard displays live detections.

Create the Flask app with MJPEG streaming

MJPEG streaming works by continuously sending JPEG frames over an HTTP connection. The browser renders each frame as it arrives, producing a live video effect without needing WebSockets or complex video codecs.

• Create a new file called app.py in your project folder.
• Add the imports, Flask initialization, and global state variables by pasting this code:

import cv2
import time
import threading
import numpy as np
from datetime import datetime
from flask import Flask, Response, render_template, jsonify
from pose_detector import PoseDetector
from activity_classifier import ActivityClassifier
from langflow_agent import LangflowAgent
from config import (
    CONSECUTIVE_FRAMES_REQUIRED,
    SUSPICIOUS_ACTIVITIES,
    FLASK_HOST,
    FLASK_PORT
)

# Initialize the Flask application
app = Flask(__name__)

# Load the pose detector, activity classifier, and Langflow agent
pose_detector = PoseDetector()
classifier = ActivityClassifier(model_path="activity_model.joblib")
agent = LangflowAgent()

# Shared state variables for cross-thread communication
frame_buffer = None
activity_log = []
suspicious_frame_count = 0
last_agent_call_time = 0
AGENT_COOLDOWN = 10
lock = threading.Lock()

What does this code do?

This sets up the Flask app and loads all the components you built in previous steps: the pose detector, the activity classifier, and the Langflow agent.

• frame_buffer stores the latest annotated frame so the web stream can read it from a separate thread.
• AGENT_COOLDOWN = 10 prevents the system from flooding the agent with requests. It waits at least 10 seconds between agent calls.
• lock is a threading lock that ensures the video thread and Flask thread don't read/write shared data at the same time.

• Below the global state section, add the process_video() function that runs the detection pipeline on each video frame:

def process_video(source=0):
    global frame_buffer, suspicious_frame_count, last_agent_call_time

    cap = cv2.VideoCapture(source)
    if not cap.isOpened():
        print(f"Error: Could not open video source {source}")
        return

    while True:
        ret, frame = cap.read()
        if not ret:
            # If using a video file, loop back to the beginning
            if isinstance(source, str):
                cap.set(cv2.CAP_PROP_POS_FRAMES, 0)
                continue
            break

        # Run pose detection and draw skeleton on the frame
        detections = pose_detector.detect(frame)
        annotated_frame = pose_detector.draw_skeleton(frame, detections)

        # Classify each detected person's activity
        current_activities = []
        any_suspicious = False

        for det in detections:
            result = classifier.classify(det["keypoints"])
            if len(result) == 3:
                activity, confidence, is_suspicious = result
            else:
                activity, confidence = result
                is_suspicious = False

            current_activities.append({
                "activity": activity,
                "confidence": confidence,
                "suspicious": is_suspicious
            })

            if is_suspicious:
                any_suspicious = True

            # Draw activity label on the frame
            if det["bbox"] is not None:
                x1, y1 = int(det["bbox"][0]), int(det["bbox"][1])
                color = (0, 0, 255) if is_suspicious else (0, 255, 0)
                label = f"{activity} ({confidence:.2f})"
                cv2.putText(annotated_frame, label, (x1, y1 - 10),
                            cv2.FONT_HERSHEY_SIMPLEX, 0.6, color, 2)

        # Temporal smoothing: count consecutive suspicious frames
        if any_suspicious:
            suspicious_frame_count += 1
        else:
            suspicious_frame_count = 0

        # Only call the agent after enough suspicious frames and cooldown
        current_time = time.time()
        if (suspicious_frame_count >= CONSECUTIVE_FRAMES_REQUIRED and
                current_time - last_agent_call_time > AGENT_COOLDOWN):
            last_agent_call_time = current_time
            suspicious_acts = [a for a in current_activities if a["suspicious"]]

            activity_data = {
                "activity": suspicious_acts[0]["activity"] if suspicious_acts else "unknown",
                "confidence": suspicious_acts[0]["confidence"] if suspicious_acts else 0.0,
                "num_people": len(detections),
                "timestamp": datetime.now().isoformat(),
                "duration": suspicious_frame_count / 30.0
            }

            # Call the agent in a background thread to avoid blocking video
            threading.Thread(
                target=_call_agent,
                args=(activity_data,),
                daemon=True
            ).start()

        # Update the shared frame buffer
        with lock:
            frame_buffer = annotated_frame.copy()

        time.sleep(0.03)

    cap.release()

What does process_video() do?

This is the main loop that ties together everything you built in previous steps. It runs on a background thread so it does not block the Flask server.

• Each frame goes through pose detection, skeleton drawing, and activity classification.
• Temporal smoothing requires CONSECUTIVE_FRAMES_REQUIRED suspicious frames in a row before alerting the agent. This prevents false positives from single-frame glitches.
• The agent is called in a daemon thread so video processing continues while waiting for the Langflow API response.
• If the video source is a file (a string path), it loops back to the beginning when it reaches the end.

• Below process_video(), add the agent helper, MJPEG generator, Flask routes, and the main block:

def _call_agent(activity_data):
    """Send suspicious activity to the Langflow agent and log the response."""
    response = agent.analyze_activity(activity_data)
    with lock:
        activity_log.append({
            "timestamp": datetime.now().isoformat(),
            "activity": activity_data["activity"],
            "confidence": activity_data["confidence"],
            "agent_decision": response.get("message", "No response")
        })


def generate_frames():
    """Yield JPEG frames for the MJPEG video stream."""
    while True:
        with lock:
            if frame_buffer is None:
                time.sleep(0.1)
                continue
            ret, jpeg = cv2.imencode(".jpg", frame_buffer)
            if not ret:
                continue
            frame_bytes = jpeg.tobytes()

        yield (b"--frame\r\n"
               b"Content-Type: image/jpeg\r\n\r\n" + frame_bytes + b"\r\n")
        time.sleep(0.03)


@app.route("/")
def index():
    return render_template("index.html")


@app.route("/video_feed")
def video_feed():
    return Response(generate_frames(), mimetype="multipart/x-mixed-replace; boundary=frame")


@app.route("/api/activity_log")
def get_activity_log():
    with lock:
        return jsonify(activity_log[-50:])


if __name__ == "__main__":
    video_source = 0
    video_thread = threading.Thread(target=process_video, args=(video_source,), daemon=True)
    video_thread.start()
    app.run(host=FLASK_HOST, port=FLASK_PORT, debug=False, threaded=True)

What does this code do?

• _call_agent() sends detection data to Langflow and appends the agent's response to the shared activity log.
• generate_frames() reads from the shared frame buffer and yields JPEG-encoded bytes. The multipart/x-mixed-replace MIME type tells the browser to replace each image with the next one, creating a live video effect.
• The /video_feed route streams frames. The /api/activity_log route returns the last 50 agent decisions as JSON for the dashboard to poll.
• The main block starts the video processing thread first, then launches Flask. The video thread is set as a daemon so it shuts down automatically when the Flask server stops.

• Save app.py.

Seeing an import error?

• Make sure app.py is in the same folder as pose_detector.py, activity_classifier.py, langflow_agent.py, and config.py.
• Confirm activity_model.joblib exists in the folder. If not, run python train_model.py again.
• help me debug an import error in app.py

Build the HTML dashboard page

Flask looks for HTML templates in a folder called templates. You will create a dark-themed dashboard that shows the live video feed on the left and agent decisions on the right.

• Create a folder called templates inside your project folder by running this command:

mkdir templates

• Create a new file called index.html inside the templates folder and paste the following code:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Suspicious Activity Detection - AI Surveillance</title>
    <style>
        * { margin: 0; padding: 0; box-sizing: border-box; }
        body { font-family: Arial, sans-serif; background: #1a1a2e; color: #eee; }
        .header { background: #16213e; padding: 15px 30px; text-align: center; }
        .header h1 { color: #4ecca3; font-size: 1.5em; }
        .container { display: grid; grid-template-columns: 2fr 1fr; gap: 20px; padding: 20px; max-width: 1400px; margin: 0 auto; }
        .video-panel { background: #0f3460; border-radius: 10px; padding: 15px; }
        .video-panel img { width: 100%; border-radius: 8px; }
        .log-panel { background: #0f3460; border-radius: 10px; padding: 15px; max-height: 80vh; overflow-y: auto; }
        .log-panel h2 { color: #4ecca3; margin-bottom: 10px; font-size: 1.1em; }
        .log-entry { background: #16213e; padding: 10px; margin-bottom: 8px; border-radius: 6px; border-left: 3px solid #e94560; }
        .log-entry .time { color: #888; font-size: 0.8em; }
        .log-entry .activity { color: #e94560; font-weight: bold; }
        .log-entry .decision { color: #4ecca3; margin-top: 5px; font-size: 0.9em; }
        .status-bar { background: #16213e; padding: 10px 30px; text-align: center; color: #4ecca3; }
    </style>
</head>
<body>
    <div class="header">
        <h1>Real-Time Suspicious Activity Detection - Agentic AI</h1>
    </div>
    <div class="status-bar">
        <span id="status">System Active - Monitoring Feed</span>
    </div>
    <div class="container">
        <div class="video-panel">
            <!-- The img tag streams MJPEG frames from the /video_feed route -->
            <img src="/video_feed" alt="Live Video Feed">
        </div>
        <div class="log-panel">
            <h2>Agent Decisions</h2>
            <div id="log-entries"></div>
        </div>
    </div>
    <script>
        // Poll the activity log API every 3 seconds and update the panel
        function fetchLog() {
            fetch("/api/activity_log")
                .then(response => response.json())
                .then(data => {
                    const container = document.getElementById("log-entries");
                    container.innerHTML = "";
                    data.reverse().forEach(entry => {
                        const div = document.createElement("div");
                        div.className = "log-entry";
                        div.innerHTML = `
                            <div class="time">${entry.timestamp}</div>
                            <div class="activity">${entry.activity} (${(entry.confidence * 100).toFixed(1)}%)</div>
                            <div class="decision">${entry.agent_decision}</div>
                        `;
                        container.appendChild(div);
                    });
                });
        }
        setInterval(fetchLog, 3000);
        fetchLog();
    </script>
</body>
</html>

What does this dashboard do?

• The <img src="/video_feed"> tag connects to the Flask MJPEG stream. The browser continuously renders incoming frames as a live video.
• The JavaScript polls /api/activity_log every 3 seconds and renders each agent decision as a card with timestamp, activity type, confidence, and the agent's recommended action.
• The CSS grid layout puts the video feed (2/3 width) on the left and the agent log (1/3 width) on the right.

• Save templates/index.html.
• Confirm the file exists by running:

ls templates

You should see index.html listed in the output.

Templates folder not found?

• Make sure you created the folder with the exact name templates (all lowercase, plural). Flask will not find templates in a folder with a different name.
• Confirm your terminal is in the project folder (the same folder that contains app.py).
• help me fix my Flask templates folder structure

Run the full surveillance dashboard

With all the code in place, you can now launch the complete system. The Flask app will start the video processing thread, load the pose model, and begin serving the dashboard.

Do I need a webcam?

The default video source is 0 (your webcam). If you do not have a webcam, you can change video_source = 0 to a video file path like video_source = "test_video.mp4" in the main block of app.py.

• Make sure Langflow is still running at http://127.0.0.1:7860 in a separate terminal.
• Start the Flask dashboard by running this command:

python app.py

• Open your browser and navigate to http://localhost:5000.

You should see the dark-themed dashboard with a live video feed on the left showing pose skeletons drawn on detected people. Activity labels appear above each person in green (normal) or red (suspicious).

What should I see in the Agent Decisions panel?

The right panel stays empty until the system detects suspicious activity for at least 5 consecutive frames. If you are using a webcam, try making sudden movements (like raising both arms quickly) to trigger a detection.

When the agent responds, a card appears with the timestamp, activity type, confidence percentage, and the agent's recommended action (log, alert, or escalate).

Dashboard not loading?

• If you see Error: Could not open video source 0 in the terminal, your webcam is unavailable. Switch to a video file by changing video_source = 0 to a file path in the main block.
• If the browser shows a blank page, check the terminal for errors. A TemplateNotFound error means index.html is not inside the templates folder.
• If the video feed appears but the Agent Decisions panel never updates, confirm Langflow is running and your LANGFLOW_FLOW_ID in config.py is correct.
• help me debug my Flask surveillance dashboard

✔️ Awesome, I've got everything!

Great. Double check you have saved both app.py and templates/index.html.

ⓧ I'd like to double check the full code

Here is the complete app.py file:

import cv2
import time
import threading
import numpy as np
from datetime import datetime
from flask import Flask, Response, render_template, jsonify
from pose_detector import PoseDetector
from activity_classifier import ActivityClassifier
from langflow_agent import LangflowAgent
from config import (
    CONSECUTIVE_FRAMES_REQUIRED,
    SUSPICIOUS_ACTIVITIES,
    FLASK_HOST,
    FLASK_PORT
)

app = Flask(__name__)

pose_detector = PoseDetector()
classifier = ActivityClassifier(model_path="activity_model.joblib")
agent = LangflowAgent()

frame_buffer = None
activity_log = []
suspicious_frame_count = 0
last_agent_call_time = 0
AGENT_COOLDOWN = 10
lock = threading.Lock()


def process_video(source=0):
    global frame_buffer, suspicious_frame_count, last_agent_call_time

    cap = cv2.VideoCapture(source)
    if not cap.isOpened():
        print(f"Error: Could not open video source {source}")
        return

    while True:
        ret, frame = cap.read()
        if not ret:
            if isinstance(source, str):
                cap.set(cv2.CAP_PROP_POS_FRAMES, 0)
                continue
            break

        detections = pose_detector.detect(frame)
        annotated_frame = pose_detector.draw_skeleton(frame, detections)

        current_activities = []
        any_suspicious = False

        for det in detections:
            result = classifier.classify(det["keypoints"])
            if len(result) == 3:
                activity, confidence, is_suspicious = result
            else:
                activity, confidence = result
                is_suspicious = False

            current_activities.append({
                "activity": activity,
                "confidence": confidence,
                "suspicious": is_suspicious
            })

            if is_suspicious:
                any_suspicious = True

            if det["bbox"] is not None:
                x1, y1 = int(det["bbox"][0]), int(det["bbox"][1])
                color = (0, 0, 255) if is_suspicious else (0, 255, 0)
                label = f"{activity} ({confidence:.2f})"
                cv2.putText(annotated_frame, label, (x1, y1 - 10),
                            cv2.FONT_HERSHEY_SIMPLEX, 0.6, color, 2)

        if any_suspicious:
            suspicious_frame_count += 1
        else:
            suspicious_frame_count = 0

        current_time = time.time()
        if (suspicious_frame_count >= CONSECUTIVE_FRAMES_REQUIRED and
                current_time - last_agent_call_time > AGENT_COOLDOWN):
            last_agent_call_time = current_time
            suspicious_acts = [a for a in current_activities if a["suspicious"]]

            activity_data = {
                "activity": suspicious_acts[0]["activity"] if suspicious_acts else "unknown",
                "confidence": suspicious_acts[0]["confidence"] if suspicious_acts else 0.0,
                "num_people": len(detections),
                "timestamp": datetime.now().isoformat(),
                "duration": suspicious_frame_count / 30.0
            }

            threading.Thread(
                target=_call_agent,
                args=(activity_data,),
                daemon=True
            ).start()

        with lock:
            frame_buffer = annotated_frame.copy()

        time.sleep(0.03)

    cap.release()


def _call_agent(activity_data):
    response = agent.analyze_activity(activity_data)
    with lock:
        activity_log.append({
            "timestamp": datetime.now().isoformat(),
            "activity": activity_data["activity"],
            "confidence": activity_data["confidence"],
            "agent_decision": response.get("message", "No response")
        })


def generate_frames():
    while True:
        with lock:
            if frame_buffer is None:
                time.sleep(0.1)
                continue
            ret, jpeg = cv2.imencode(".jpg", frame_buffer)
            if not ret:
                continue
            frame_bytes = jpeg.tobytes()

        yield (b"--frame\r\n"
               b"Content-Type: image/jpeg\r\n\r\n" + frame_bytes + b"\r\n")
        time.sleep(0.03)


@app.route("/")
def index():
    return render_template("index.html")


@app.route("/video_feed")
def video_feed():
    return Response(generate_frames(), mimetype="multipart/x-mixed-replace; boundary=frame")


@app.route("/api/activity_log")
def get_activity_log():
    with lock:
        return jsonify(activity_log[-50:])


if __name__ == "__main__":
    video_source = 0
    video_thread = threading.Thread(target=process_video, args=(video_source,), daemon=True)
    video_thread.start()
    app.run(host=FLASK_HOST, port=FLASK_PORT, debug=False, threaded=True)

And the complete templates/index.html file:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Suspicious Activity Detection - AI Surveillance</title>
    <style>
        * { margin: 0; padding: 0; box-sizing: border-box; }
        body { font-family: Arial, sans-serif; background: #1a1a2e; color: #eee; }
        .header { background: #16213e; padding: 15px 30px; text-align: center; }
        .header h1 { color: #4ecca3; font-size: 1.5em; }
        .container { display: grid; grid-template-columns: 2fr 1fr; gap: 20px; padding: 20px; max-width: 1400px; margin: 0 auto; }
        .video-panel { background: #0f3460; border-radius: 10px; padding: 15px; }
        .video-panel img { width: 100%; border-radius: 8px; }
        .log-panel { background: #0f3460; border-radius: 10px; padding: 15px; max-height: 80vh; overflow-y: auto; }
        .log-panel h2 { color: #4ecca3; margin-bottom: 10px; font-size: 1.1em; }
        .log-entry { background: #16213e; padding: 10px; margin-bottom: 8px; border-radius: 6px; border-left: 3px solid #e94560; }
        .log-entry .time { color: #888; font-size: 0.8em; }
        .log-entry .activity { color: #e94560; font-weight: bold; }
        .log-entry .decision { color: #4ecca3; margin-top: 5px; font-size: 0.9em; }
        .status-bar { background: #16213e; padding: 10px 30px; text-align: center; color: #4ecca3; }
    </style>
</head>
<body>
    <div class="header">
        <h1>Real-Time Suspicious Activity Detection - Agentic AI</h1>
    </div>
    <div class="status-bar">
        <span id="status">System Active - Monitoring Feed</span>
    </div>
    <div class="container">
        <div class="video-panel">
            <img src="/video_feed" alt="Live Video Feed">
        </div>
        <div class="log-panel">
            <h2>Agent Decisions</h2>
            <div id="log-entries"></div>
        </div>
    </div>
    <script>
        function fetchLog() {
            fetch("/api/activity_log")
                .then(response => response.json())
                .then(data => {
                    const container = document.getElementById("log-entries");
                    container.innerHTML = "";
                    data.reverse().forEach(entry => {
                        const div = document.createElement("div");
                        div.className = "log-entry";
                        div.innerHTML = `
                            <div class="time">${entry.timestamp}</div>
                            <div class="activity">${entry.activity} (${(entry.confidence * 100).toFixed(1)}%)</div>
                            <div class="decision">${entry.agent_decision}</div>
                        `;
                        container.appendChild(div);
                    });
                });
        }
        setInterval(fetchLog, 3000);
        fetchLog();
    </script>
</body>
</html>

Your full surveillance system is now live. The dashboard streams annotated video in real time, and when the AI agent detects sustained suspicious behavior, its decisions appear in the log panel. You have a complete pipeline from camera input to AI-powered security reasoning, all visible in one interface.

Secret mission

Multi-Person Tracking with Threat Escalation

Your detection system currently treats every suspicious frame as an isolated event. What if the same person keeps triggering alerts? In this mission, you'll add persistent person IDs and a threat memory system so your agent can escalate its response for repeat offenders.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally and uses the free tier of the Gemini API, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

• Python packages (ultralytics, opencv-python, scikit-learn, flask, google-genai, requests, langflow).
• Langflow server running on http://127.0.0.1:7860.
• Flask monitoring dashboard running on http://127.0.0.1:5000.
• Google Gemini API key (free tier).
• Trained model file (activity_model.joblib).
• YOLOv8 pose model file (yolov8n-pose.pt, downloaded automatically).
• Project source files and templates.

Keep everything running

No action needed. Choose this if you are still actively building or experimenting with the system.

• Keep Langflow running in its terminal window to continue testing agent flows.
• Keep the Flask app running to monitor video feeds.
• Your Gemini API key remains active and usable within the free tier limits.

Pause - I'll come back to this later

Shut down running processes to free up system memory. Your project files and trained model stay on disk.

• Stop the Flask app by pressing Ctrl+C in its terminal window.
• Stop Langflow by pressing Ctrl+C in the terminal where it is running.
• Close any OpenCV video windows that are still open.
• Your project folder, trained model, and API key remain intact for next time.

Delete - I don't want to use this again

Remove all project resources and start fresh. This cannot be undone.

• Stop all running processes (Flask, Langflow) with Ctrl+C.
• Delete your project folder from the Desktop.
• Uninstall the Python packages by running pip uninstall ultralytics opencv-python scikit-learn flask google-genai requests langflow -y.
• Remove the GEMINI_API_KEY environment variable from your system.
• Optionally, revoke or delete your Gemini API key from Google AI Studio.

That's a Wrap!

That's a Wrap!

Nice work! You just built a complete AI-powered surveillance intelligence system that combines computer vision, machine learning, and agentic AI into a single cohesive application.

Here's what you accomplished:

• Built a real-time pose estimation pipeline using YOLOv8 that detects 17 body keypoints per person from a live video feed.
• Trained an activity classifier using a Random Forest model that distinguishes normal behavior from suspicious activities based on pose features.
• Created an AI agent in Langflow powered by Google Gemini that autonomously reasons about security incidents and delivers decisions to a live Flask monitoring dashboard.
• Secret Mission: Added multi-person tracking with persistent IDs and a threat escalation system that remembers individual behavior over time.

Ready to quiz yourself?