A Simple Agentic System For Messaging

Introduction

The goal of this project was to build an messaging system for a car customization shop, that can take in an incoming customer message and engage in conversation with them in the business owner's tone, with the goal of guiding them to book a service

General Overview of System

The main architecture for this system is a decentralized network of LangChain agents. A central Router agent analyzes each incoming message and assigns it a classification code based on predefined categories. The message is then passed to a corresponding sub-agent that specializes in that category. Then for the categories whose information about it is dependent upon the particular business, the sub-agent will Analyze the message to further determine what exactly it pertains to, Generates a draft reply as a response to the message, and finally Enforces rules created by the business on the draft reply for the final output message.

Thus, the incoming messages go through a series of 4 steps(when applicable), Route, Analyze, Generator, and Enforce, in order to create a response. Giving rise to the shorthand name for the architecture, RAGE.

A high level overview of the system is shown below:

  

  

graph TD;

  

  

Input-Message --> Router-Agent;

  

  

Router-Agent --> Classification-Code;

  

  

Classification-Code --> Sub-Agent-1;

  

  

Classification-Code --> Sub-Agent-2;

  

  

Classification-Code --> Sub-Agent-N;

  

  

Sub-Agent-1 --> Output-Message;

  

  

Sub-Agent-2 --> Output-Message;

  
  
  

Sub-Agent-N --> Output-Message;

  

  

Loading

Where each sub-agent, is an agent created with Langchain, that will take an incoming message, and have it go through the remaining 3 steps of, Analyze, Generate, and Enforce, to produce the output message.

Predefined Categories for This Business

For the business I made this for, the relevant categories for potential conversation topics with customers were the following,

1. Services & Pricing
2. Business Operations
3. Bookings
4. Customer gives a closing response

Recall that specialized sub-agents, that follow the full RAGE steps, are only created for those categories that rely upon providing information that is dependent upon the business, which for this case is only 1&2.

The classification codes used for each category is the following:

• Service & Pricing -> A
• Bookings -> B
• Business Operations -> C
• Closing Response -> E

Architecture of a Specialized Sub-Agent

After the router classifies the input message(the R step), it's handed to the relevant sub-agent, where if applicable, it goes through the remaining, AGE steps. A breakdown of each step is listed below:

• The analyze step will process a input message, relevant to the category it exists within, i.e for the service sub-agent the analyze step will process a service related query. The result of this processing is one or multiple classifications codes, that pertain to specific details the message has queried about. These codes are then used in a dictionary lookup to return details that the generator will need in order to draft its reply

• The generator step will process the inputted message along with the details, and generate a draft response to the users query

• Finally the enforce step is there to ensure that the draft reply has followed all critical rules given by the business for the category, such as pricing rules for the service & pricing sub-agent

For the sub-agents whose categories don't correspond to information that depends upon the business, the sub-agents simply are just Language Models that will generate a response given some rules. Therefore they are not really "agents" but in the sense of how they are created in python with LangChain they are in that sense

A diagram showing what a specialized sub-agent will perform is shown here:

  

  

graph TD;

  

  

Input-Message --> Analyzer;

  

Analyzer --> Codes;

  

Codes --> Detail-Lookup;

  

  

Detail-Lookup --> Details;

Details --> Generator;

  

Input-Message --> Generator;

  

  

Generator --> Draft-Reply;

  

Draft-Reply --> Enforcer;

  

Enforcer --> Final-Reply;

  

  

Loading

Other Categories

If the incoming message falls into one of these categories,

• Seeking to speak on the phone with owner
• Not related to business
• Is a message that is in conversation with the owner

Then the messages are not handed off to any sub-agent at all, rather they will either be ignored or notify the owner

In this code these categories correspond to the following codes

• Seeking to speak on the phone with owner -> F
• Is a message that is in conversation with the owner -> D
• Not related to business -> Z

For the D & Z cases they are ignored, though the logging process is different for each, that's why gave different codes, and for the F case a message is sent to the business notifying them

Other Parts of The System

The agents are the core of this system, but there is a good amount of machinery around them that takes a raw social media message in and gets the final reply out. The entire path a message takes through the system is shown below:

graph TD;

Meta-Webhook --> Flask-Server;

Flask-Server --> Message-Batcher;

Message-Batcher --> Media-Processing;

Media-Processing --> RAGE-Agents;

RAGE-Agents --> Owner-Review;

Owner-Review --> Reply-Sent;

Loading

Webhooks

The messages come in from Meta webhooks. mams_server.py runs a single flask app on port 7070 that mounts a webhook route for each channel, /webhook/instagram and /webhook/messenger, reusing the view functions that live in mams_instagram.py and mams_facebook.py. Every incoming POST gets printed raw and also appended to a daily log file in webhook-logs/, so whether a message was ever delivered can be proven later. After that the event is filtered, where it's dropped if it is older than 24 hours, a duplicate delivery, an echo of the system's own send, or has no text/media at all. Whatever survives gets handed to the batcher.

Message Batching & Media

Customers rarely send one clean message, rather they send several quick ones in a row, sometimes with media. So instead of answering each one separately, the MessageBatcher in mams_batching.py buffers each user's messages, and every new message restarts a 150 second countdown. When the countdown finally finishes the batch is flushed: any attached media (images, videos, shared posts/reels/stories, even the ad the customer clicked to start the conversation) gets turned into a text description by a vision model in mams_media_processing.py, and everything is merged into the single numbered message that the agents read.

SQL Lite DBs

There are two sqlite databases that sit at the project root,

• mams_state.sqlite -> handled by mams_store.py, this holds all the per user state the system needs to survive restarts, in buckets like the last customer/agent messages, whether pricing was sent, whether the booking link was sent, the detected car model, restricted users, and the pending owner-ask flag. It is all one state table of (bucket, user, value) rows that gets loaded at boot and saved after every turn
• agent_memory.sqlite -> a LangGraph checkpointer that gives the text generation agents per user conversation memory, using the user's id as the thread id, so the agents still remember their conversations after a restart

Dashboard

The dashboard is a separate web app that the system reports to, and it is completely optional, it only turns on if DASHBOARD_URL and DASH_KEY are set. All the reporting lives in mams_dashboard.py and fires on background threads so the message path never waits on it. It does 3 jobs,

1. Activity reporting -> every incoming message, every outgoing reply (and whether the booking link was in it), and even replies the owner typed by hand in the inbox, get mirrored into a per customer conversation view
2. Restriction syncing -> the owner can toggle the agents off for any customer from their phone. When the dashboard is reachable its toggle wins, with the local sqlite state as the fallback
3. Flagged conversations -> conversations the owner reports on the dashboard get pulled down every minute and saved as JSON files in flagged-conversations/, which can later be turned into new examples for the agents

Owner Review

Currently in the system for development purposes, before any reply actually goes out, there is a 10 second countdown in the terminal where you can press n to cancel and hand the conversation off to a human, r to redo the reply with feedback (the feedback gets folded into what the agents generate from), or j to just cancel silently. If nothing is pressed the reply sends. If the process is not attached to a terminal, then the message simply sends

Important Functions

• mams() in mams.py -> the main function of the whole system. Takes (unique_id, user_message, feedback, channel), runs the restriction check, the returning customer check, the owner-ask follow up, and the car model analyzer, then routes the classification code to the right sub-agent and saves all state after
• classifier_agent() -> the R step, reads the message plus the previous turn and returns the single classification character
• service_pricing_agent() & business_ops_agent() -> the two specialized sub-agents that perform the full AGE steps
• book_agent() -> sends the booking link, or references it instead if it was already sent to that customer. The booking branch only runs once pricing has been sent and the car model is known, otherwise it falls back to asking for what is missing
• car_model_analyzer() & car_model_interrogator() -> pulls the customer's vehicle out of a message & asks for it when booking needs it
• on_fence_handler() -> the simple sub-agent for the closing response category
• run_with_timeout() -> runs any agent call on a side thread under a hard 90 second timeout, so one stuck call can never freeze processing
• notify_human_in_loop() -> texts the owner over iMessage when a conversation needs a human, fires at most once per user
• send_ig_message() & send_messenger_message() -> the Graph API senders for each channel
• return_category_details() -> the dictionary lookup that turns the analyzer's codes into the details the generator drafts from
• extract_reply() -> pulls the final message out of the enforcer's reply tags

There are also two special tokens an agent can output: ESCALATE, for requests the agents cannot handle (the owner is notified and the system goes silent for that user), and OWNER_ASK, for when the agent offers to ask the owner something it has no info on (the customer's next message is then treated as the yes/no answer to that offer).

Usage

This system was built around one specific shop, but everything business dependent lives in just a few places, so using it for your own business is mostly swapping out content rather than code. The predefined categories also generalize to most service businesses, though they can be changed in the router's system prompt if needed.

What Information Is Needed

1. Service details -> detail-jsons/service_pricing_details.json holds the dictionary of service codes (s1, s2, ...) where each maps to everything about that service, pricing, materials, turnaround, etc. Replace these with your own services, and update the service analyzer's codes in sys_prompts.py to match
2. Business details -> detail-jsons/business_operations_details.json is the same idea for the business info fields (hours, address, contact, website, ...) under codes b1, b2, ...
3. Rules -> sys_prompts.py holds the system prompts for every agent, including the business context, the special pricing rules, and the hard rules the enforcers apply. This is where the business' critical rules get written
4. Examples -> sys_prompt_examples.py is the most important piece for tone. Every agent treats its examples as ground truth, so they should be real customer messages with real replies written exactly how the owner texts
5. Constants -> the booking link lives in mams_agents.py (BOOKING_LINK) and the owner's phone number for notifications lives in mams_functions.py (OWNER_PHONE)

What Is Needed To Install

1. Python 3.12 with a venv, then pip install langchain, langchain-anthropic, langchain-tavily, langgraph, langgraph-checkpoint-sqlite, flask, requests, rich, python-dotenv, opencv-python, and pillow
2. A Meta developer app with the messaging permissions, linked to the business' Instagram account and Facebook page, with its webhooks subscribed to the /webhook/instagram and /webhook/messenger routes
3. A src/.env file holding the keys: ANTHROPIC_API_KEY (every agent runs on Claude), TAVILY_API_KEY (the web search tool), ACCESS_TOKEN, IG_ACCESS_TOKEN, ACCESS_TOKEN_P (the Instagram/page access tokens for sending and lookups), VERIFY_TOKEN, VERIFY_TOKEN_P (the webhook verification handshakes), BLOOIO_API_KEY (the iMessage notifications to the owner), and optionally DASHBOARD_URL and DASH_KEY for the dashboard
4. Run python src/mams_server.py to start the flask app on port 7070, then expose that port publicly (ngrok, a cloudflare tunnel, etc) so Meta can deliver the webhooks

Once thats up, incoming DMs flow through the whole pipeline on their own