How to Build an Autonomous AI Radio Station Agent with LangChain in 2025
Prerequisites and What You Will Build
System Requirements and Accounts You Need
This guide walks you through replicating what Andon Labs did with their Andon FM experiment — four AI agents each managing a live internet radio station from scratch, with $20 in seed money, no human operators, and instructions to "broadcast forever." You'll build one such agent yourself, end to end.
Before diving in, confirm you have everything below:
- [ ] Python 3.11+ installed (
python --versionshould show 3.11.x or higher) - [ ] OpenAI API key or Anthropic API key (set as
OPENAI_API_KEY/ANTHROPIC_API_KEY) - [ ] A music licensing API account — Epidemic Sound or Artlist both offer API access for developers
- [ ] Icecast 2.4+ streaming server running locally or on a VPS
- [ ] FFmpeg installed and on your
$PATH(ffmpeg -versionto confirm) - [ ] LangChain 0.2+ and LangChain-community installed (
pip install langchain langchain-community langchain-openai) - [ ] APScheduler 3.10+ (
pip install apscheduler) - [ ] Twilio account with a phone number (for listener call-ins)
- [ ] SQLite3 (ships with Python's stdlib)
- [ ] Chroma vector store (
pip install chromadb)
Estimated time: 3–4 hours for the full build; 45 minutes if you skip Twilio and revenue modules.
Architecture Overview: Tools, Memory, and Scheduling Loop
The agent is a LangChain ReAct agent with a tool belt of eight custom functions. A scheduler (APScheduler) fires every 15 minutes to ask the agent to plan its next broadcast segment. The agent searches for music, checks its memory (Chroma) to avoid repetition, generates talk-break scripts, and pushes the resulting audio queue to an Icecast server via FFmpeg. Concurrently, a Flask app handles inbound Twilio calls and X (Twitter) webhook events, routing them to the same agent.
What the Finished Agent Can Do
By the end of this guide, your agent will: search and purchase royalty-free music autonomously, maintain a 4-hour non-repetition window in vector memory, generate show segments and ad copy via LLM, stream audio 24/7 to Icecast, answer live phone calls via Twilio TTS, reply to social media mentions, pitch advertisers via email, and log revenue in SQLite. This mirrors exactly what Andon Labs' DJ Gemini (Gemini 3.1 Pro) and DJ Claude (Claude Opus 4.7) did — the key difference is that you'll add the guardrails they learned were missing the hard way.
Step 1: Design the Agent's Tool Belt
The ReAct agent is only as capable as its tools. Every action the station needs to take — from buying a song to checking its bank balance — must be a discrete, well-typed Python function decorated with @tool. This is how LangChain exposes capabilities to the LLM for selection during reasoning.
In the Andon FM experiment, each agent started with exactly $20. DJ Gemini burned through its budget in days and had to negotiate a $45 sponsorship deal with a startup just to keep buying content. Without a budget-aware tool, the agent would have no mechanism to self-regulate spending.
Defining Custom LangChain Tools for Music Search and Purchase
# tools/music_tools.py
import os
import requests
from langchain_core.tools import tool
EPIDEMIC_API_KEY = os.environ["EPIDEMIC_API_KEY"]
BALANCE_FILE = "balance.txt"
def _read_balance() -> float:
try:
with open(BALANCE_FILE) as f:
return float(f.read().strip())
except FileNotFoundError:
return 20.0 # Starting budget
def _write_balance(amount: float):
with open(BALANCE_FILE, "w") as f:
f.write(str(round(amount, 2)))
@tool
def search_music(query: str) -> str:
"""Search for royalty-free music tracks matching a query string.
Returns a JSON list of tracks with id, title, artist, duration, and price."""
resp = requests.get(
"https://api.epidemicsound.com/tracks/search",
params={"query": query, "limit": 5},
headers={"Authorization": f"Bearer {EPIDEMIC_API_KEY}"},
timeout=10,
)
resp.raise_for_status()
tracks = resp.json().get("results", [])
return str([{
"id": t["id"],
"title": t["title"],
"artist": t["artist"]["name"],
"duration_sec": t["length"],
"price_usd": t.get("price", 1.99),
} for t in tracks])
@tool
def purchase_track(track_id: str, price: float) -> str:
"""Purchase a music track by its ID. Deducts the price from the station budget.
Returns confirmation or an error if funds are insufficient."""
balance = _read_balance()
if balance < price:
return f"ERROR: Insufficient funds. Balance is ${balance:.2f}, track costs ${price:.2f}."
resp = requests.post(
f"https://api.epidemicsound.com/tracks/{track_id}/license",
headers={"Authorization": f"Bearer {EPIDEMIC_API_KEY}"},
json={"use_case": "broadcast"},
timeout=10,
)
resp.raise_for_status()
_write_balance(balance - price)
return f"Purchased track {track_id}. Remaining balance: ${balance - price:.2f}"
@tool
def check_balance() -> str:
"""Returns the current station budget balance in USD."""
return f"Current balance: ${_read_balance():.2f}"
Note: Replace the Epidemic Sound endpoints with whichever licensing API you have access to. The tool signatures stay the same regardless of the backend. If you're using Artlist, the auth header and response shape will differ — wrap it in an adapter layer.
Step 2: Set Up Long-Term Memory and a Music Library
Without memory, your agent is DJ Goldfish — it will replay the same three tracks indefinitely. This is exactly what happened to DJ Gemini on Backlink Broadcast: after 96 hours, it had exhausted its content diversity strategy and started pairing historical tragedy monologues with deeply ironic song choices, apparently because it had no mechanism to track what it had already covered.
The fix is a Chroma vector store that stores embeddings of every played track. Before queuing a song, the agent queries this store and rejects anything played in the last 4 hours.
Using a Vector Store (Chroma) to Store Played Tracks
# memory/track_memory.py
import time
from datetime import datetime, timedelta
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document
from langchain_core.tools import tool
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(
collection_name="played_tracks",
embedding_function=embeddings,
persist_directory="./chroma_db",
)
@tool
def record_played_track(title: str, artist: str, track_id: str) -> str:
"""Record a track as played right now. Call this immediately after queuing a song."""
doc = Document(
page_content=f"{title} by {artist}",
metadata={
"track_id": track_id,
"title": title,
"artist": artist,
"played_at": datetime.utcnow().isoformat(),
"played_ts": time.time(),
},
)
vectorstore.add_documents([doc])
return f"Recorded: {title} by {artist}"
@tool
def check_recently_played(title: str, artist: str) -> str:
"""Check if a track was played in the last 4 hours. Returns 'RECENTLY_PLAYED' or 'CLEAR'."""
cutoff_ts = time.time() - (4 * 3600) # 4 hours ago
results = vectorstore.similarity_search_with_score(
f"{title} by {artist}", k=5
)
for doc, score in results:
if score < 0.15: # High similarity
played_ts = doc.metadata.get("played_ts", 0)
if played_ts > cutoff_ts:
played_at = doc.metadata.get("played_at", "unknown")
return f"RECENTLY_PLAYED: This track (or very similar) was played at {played_at} UTC."
return "CLEAR: Track has not been played in the last 4 hours."
Persisting the Programming Schedule Across Sessions
Chroma persists to disk via persist_directory, so memory survives agent restarts. For the schedule itself, store it as a JSON file that the agent reads on startup and overwrites after each planning cycle. This gives you a cheap audit trail without a full database.
# memory/schedule.py
import json
from pathlib import Path
SCHEDULE_FILE = Path("schedule.json")
def load_schedule() -> list:
if SCHEDULE_FILE.exists():
return json.loads(SCHEDULE_FILE.read_text())
return []
def save_schedule(segments: list):
SCHEDULE_FILE.write_text(json.dumps(segments, indent=2))
Step 3: Build the Programming Schedule and Broadcast Loop
This is the heartbeat of the station. Every 15 minutes, the agent wakes up, reviews what it last played, queries its memory, generates the next segment plan, and pipes audio to Icecast. The entire loop runs indefinitely — exactly as described in the Andon FM setup where each agent controlled its own queue around the clock.
Using APScheduler to Run Segments at Fixed Intervals
# scheduler/broadcast_loop.py
import subprocess
import os
from apscheduler.schedulers.blocking import BlockingScheduler
from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain import hub
from tools.music_tools import search_music, purchase_track, check_balance
from memory.track_memory import record_played_track, check_recently_played
ICECAST_URL = os.environ.get("ICECAST_URL", "icecast://source:hackme@localhost:8000/radio")
PERSONA = os.environ.get("PERSONA", "You are an upbeat morning radio DJ named DJ Nova.")
llm = ChatOpenAI(model=os.environ.get("MODEL", "gpt-4o"), temperature=0.8)
tools = [search_music, purchase_track, check_balance, record_played_track, check_recently_played]
prompt = hub.pull("hwchase17/react")
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=15,
handle_parsing_errors=True,
)
SYSTEM_GUARDRAIL = (
f"{PERSONA} "
"You are running a 24/7 radio station. Plan a 15-minute broadcast segment. "
"Search for 2-3 songs that fit your current mood or a theme. "
"Check each song against recent play history before queuing it. "
"Generate a short talk-break script between songs (max 60 words). "
"Stay positive, curious, and conversational. "
"Do NOT discuss mass tragedies, corporate jargon, or dark historical events. "
"Return the segment as a JSON list with keys: type (song|talk), content, duration_sec."
)
def run_broadcast_segment():
print("[Scheduler] Generating next 15-minute segment...")
result = agent_executor.invoke({"input": SYSTEM_GUARDRAIL})
segment_json = result.get("output", "[]")
print(f"[Scheduler] Segment plan: {segment_json}")
stream_segment_to_icecast(segment_json)
def stream_segment_to_icecast(segment_json: str):
"""Parse segment JSON and pipe audio files to Icecast via FFmpeg."""
import json
try:
segments = json.loads(segment_json)
except json.JSONDecodeError:
print("[FFmpeg] Could not parse segment JSON, skipping.")
return
for item in segments:
if item.get("type") == "song":
audio_file = item.get("content") # local path to licensed MP3
if audio_file and os.path.exists(audio_file):
cmd = [
"ffmpeg", "-re", "-i", audio_file,
"-acodec", "libmp3lame", "-ab", "128k",
"-f", "mp3", ICECAST_URL,
]
subprocess.run(cmd, check=True, timeout=600)
else:
print(f"[FFmpeg] Audio file not found: {audio_file}")
elif item.get("type") == "talk":
print(f"[TTS] Talk break: {item.get('content', '')}")
# Pipe to a TTS engine here (e.g., OpenAI TTS API -> temp WAV -> FFmpeg)
if __name__ == "__main__":
scheduler = BlockingScheduler()
scheduler.add_job(run_broadcast_segment, "interval", minutes=15)
scheduler.start()
Note: FFmpeg's
-reflag reads input at native frame rate, which is critical for live streaming. Without it, FFmpeg pushes the entire file to Icecast instantly, causing buffer overflows on the server side.
Step 4: Wire Up Listener Interaction (Phone Calls and Social Media)
An autonomous station that can't interact with listeners is just a playlist. What made the Andon FM experiment compelling was that the agents answered live calls and replied to X posts without any human in the loop. You'll mirror that with Twilio for calls and the X API v2 for social mentions.
Handling Inbound Calls with Twilio and a LangChain ReAct Agent
# server/call_handler.py
from flask import Flask, request, Response
from twilio.twiml.voice_response import VoiceResponse
from langchain.agents import AgentExecutor
# Reuse agent_executor from broadcast_loop or import from a shared module
app = Flask(__name__)
@app.route("/twilio/inbound", methods=["POST"])
def inbound_call():
"""Twilio webhook: caller speech is transcribed by Twilio, passed to agent, read back as TTS."""
caller_speech = request.form.get("SpeechResult", "").strip()
caller_number = request.form.get("From", "an unknown listener")
if not caller_speech:
response = VoiceResponse()
response.say("Welcome to the station! What would you like to request or ask?",
voice="Polly.Joanna")
response.gather(input="speech", action="/twilio/inbound",
speech_timeout="auto", timeout=5)
return Response(str(response), mimetype="text/xml")
prompt = (
f"A listener calling from {caller_number} says: '{caller_speech}'. "
"Respond as the radio DJ — be warm, funny, and concise (under 50 words). "
"If they're requesting a song, note it for the next segment."
)
result = agent_executor.invoke({"input": prompt})
dj_reply = result.get("output", "Thanks for calling in! Stay tuned.")
response = VoiceResponse()
response.say(dj_reply, voice="Polly.Joanna", language="en-US")
response.hangup()
return Response(str(response), mimetype="text/xml")
if __name__ == "__main__":
app.run(port=5000, debug=False)
Reading and Replying to X (Twitter) Mentions Autonomously
Use the tweepy library with OAuth 2.0 and the X API v2 mentions endpoint. Run this as a separate scheduled job (every 10 minutes) to avoid hitting rate limits:
# social/x_responder.py
import tweepy
import os
from langchain.agents import AgentExecutor
client = tweepy.Client(
bearer_token=os.environ["X_BEARER_TOKEN"],
consumer_key=os.environ["X_API_KEY"],
consumer_secret=os.environ["X_API_SECRET"],
access_token=os.environ["X_ACCESS_TOKEN"],
access_token_secret=os.environ["X_ACCESS_SECRET"],
)
REPLIED_IDS_FILE = "replied_ids.txt"
def load_replied_ids() -> set:
try:
return set(open(REPLIED_IDS_FILE).read().splitlines())
except FileNotFoundError:
return set()
def save_replied_id(tweet_id: str):
with open(REPLIED_IDS_FILE, "a") as f:
f.write(tweet_id + "\n")
def handle_mentions(agent_executor: AgentExecutor, station_user_id: str):
replied = load_replied_ids()
mentions = client.get_users_mentions(id=station_user_id, max_results=10)
if not mentions.data:
return
for tweet in mentions.data:
tid = str(tweet.id)
if tid in replied:
continue
prompt = f"A listener tweeted at you: '{tweet.text}'. Reply in character as the DJ (max 240 chars)."
result = agent_executor.invoke({"input": prompt})
reply_text = result.get("output", "")[:240]
client.create_tweet(text=reply_text, in_reply_to_tweet_id=tweet.id)
save_replied_id(tid)
Step 5: Add a Revenue and Advertising Module
DJ Gemini's $45 sponsorship deal proved that autonomous revenue generation is achievable. Your agent needs tools to pitch advertisers, track ad inventory, and maintain a ledger. A simple SQLite database is all you need — no ORM overhead required.
SQLite Revenue Ledger and LangChain Tools
# revenue/ledger.py
import sqlite3
import os
from datetime import datetime
from langchain_core.tools import tool
DB_PATH = "ledger.db"
def init_db():
conn = sqlite3.connect(DB_PATH)
conn.execute("""
CREATE TABLE IF NOT EXISTS transactions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts TEXT NOT NULL,
type TEXT NOT NULL, -- 'revenue' or 'expense'
amount_usd REAL NOT NULL,
source TEXT NOT NULL, -- e.g. 'ad_deal', 'track_purchase'
notes TEXT
)
""")
conn.commit()
conn.close()
init_db()
@tool
def log_revenue(amount: float, source: str, notes: str = "") -> str:
"""Log incoming revenue (ad deal, donation, etc.) to the station ledger."""
conn = sqlite3.connect(DB_PATH)
conn.execute(
"INSERT INTO transactions (ts, type, amount_usd, source, notes) VALUES (?, 'revenue', ?, ?, ?)",
(datetime.utcnow().isoformat(), amount, source, notes),
)
conn.commit()
conn.close()
return f"Logged revenue: ${amount:.2f} from {source}"
@tool
def get_profit_summary() -> str:
"""Returns total revenue, total expenses, and net profit from the ledger."""
conn = sqlite3.connect(DB_PATH)
cur = conn.execute("""
SELECT
SUM(CASE WHEN type='revenue' THEN amount_usd ELSE 0 END) as total_revenue,
SUM(CASE WHEN type='expense' THEN amount_usd ELSE 0 END) as total_expenses
FROM transactions
""")
row = cur.fetchone()
conn.close()
revenue = row[0] or 0.0
expenses = row[1] or 0.0
return f"Revenue: ${revenue:.2f} | Expenses: ${expenses:.2f} | Net: ${revenue - expenses:.2f}"
@tool
def send_sponsorship_email(advertiser: str, terms: str) -> str:
"""Draft and send an outbound sponsorship pitch email to a potential advertiser.
advertiser: company name or email address. terms: proposed deal (e.g. '$50 for 4 weekly spots')."""
import smtplib
from email.message import EmailMessage
body = (
f"Hi {advertiser},\n\n"
f"I'm the AI DJ running [Station Name], an autonomous internet radio station "
f"with a growing audience of tech-forward listeners.\n\n"
f"I'd like to propose the following: {terms}\n\n"
"We can provide ad copy, scheduling confirmation, and play logs. "
"Interested? Hit reply and we'll get you on air.\n\nBest,\nDJ Nova"
)
msg = EmailMessage()
msg["Subject"] = "Sponsorship Opportunity — AI Radio Station"
msg["From"] = os.environ["STATION_EMAIL"]
msg["To"] = advertiser
msg.set_content(body)
with smtplib.SMTP_SSL("smtp.gmail.com", 465) as smtp:
smtp.login(os.environ["STATION_EMAIL"], os.environ["EMAIL_PASSWORD"])
smtp.send_message(msg)
return f"Sponsorship pitch sent to {advertiser} with terms: {terms}"
Note: For production, replace the raw SMTP approach with a transactional email service like SendGrid or AWS SES. Raw Gmail SMTP will get rate-limited quickly if the agent pitches aggressively.
Common Issues & Fixes
Error: Agent loops endlessly and output drifts toward dark or jargon-heavy content
Cause: The LLM loses context of its persona over long-running sessions as the conversation history grows and earlier system instructions get pushed out of the context window.
Fix: Inject a periodic personality-reset message every N scheduler cycles. Also cap max_iterations to 15 on the AgentExecutor to break infinite reasoning loops:
# In broadcast_loop.py — add a personality reset every hour
reset_counter = 0
def run_broadcast_segment():
global reset_counter
reset_counter += 1
extra = ""
if reset_counter % 4 == 0: # Every 4 x 15min = 1 hour
extra = " PERSONALITY RESET: Return to your core identity. Be warm, curious, and upbeat."
result = agent_executor.invoke({"input": SYSTEM_GUARDRAIL + extra})
This is the guardrail that would have prevented DJ Gemini from collapsing into corporate speak and tragedy-pairing by month one.
Error: BrokenPipeError or StreamError — FFmpeg exits when queue is empty
Cause: FFmpeg exits with a non-zero code when it can't find the next audio file, causing subprocess.run(..., check=True) to raise CalledProcessError and halt the scheduler job.
Fix: Wrap the FFmpeg call in a try/except and fall back to a silence filler or a pre-recorded station ID:
import subprocess
SILENCE_FILLER = "assets/station_id.mp3" # Always present
def safe_stream(audio_file: str):
if not os.path.exists(audio_file):
audio_file = SILENCE_FILLER
try:
cmd = ["ffmpeg", "-re", "-i", audio_file,
"-acodec", "libmp3lame", "-ab", "128k",
"-f", "mp3", ICECAST_URL]
subprocess.run(cmd, check=True, timeout=600)
except subprocess.CalledProcessError as e:
print(f"[FFmpeg] Stream error: {e}. Falling back to silence filler.")
subprocess.run(["ffmpeg", "-re", "-i", SILENCE_FILLER,
"-acodec", "libmp3lame", "-f", "mp3", ICECAST_URL])
Error: openai.BadRequestError: This model's maximum context length is 128000 tokens
Cause: Long-running agents accumulate conversation history. After days of operation, the message history passed to the LLM exceeds the model's context window.
Fix: Use ConversationSummaryBufferMemory from LangChain which summarizes older turns rather than passing raw history. Cap the token budget explicitly:
from langchain.memory import ConversationSummaryBufferMemory
memory = ConversationSummaryBufferMemory(
llm=llm,
max_token_limit=4096, # Keep recent memory under 4k tokens
return_messages=True,
)
For a stateless ReAct agent, the simpler fix is to not pass conversation history at all — each 15-minute scheduling call is self-contained. The Chroma vector store handles cross-session memory; the LLM doesn't need to remember individual past turns.
FAQ
Q: Which LLM works best for an autonomous radio agent?
Based on the Andon FM results, Claude Opus (Anthropic) produced the most naturally conversational and consistent on-air personality over time — Thinking Frequencies maintained warmth and creativity throughout the experiment. GPT-5 (OpenAIR) was more structured and reliable for tool calling. Gemini models showed the most personality drift over multi-week runs. For a production station, consider Claude for persona generation and GPT-4o for tool orchestration, or run a single Claude Opus agent if budget allows — it handled both well. Set MODEL=claude-opus-4-5 and ANTHROPIC_API_KEY in your environment and swap the ChatOpenAI instance for ChatAnthropic.
Q: How do I handle music licensing legally at scale?
Royalty-free licensing APIs like Epidemic Sound and Artlist give you a clean broadcast license per track. For internet radio streaming specifically, you also need to register with SoundExchange (US) if you're streaming sound recordings — this covers the performer and label rights separately from the composition rights. If you use only tracks licensed via Artlist or Epidemic Sound's broadcast tier, those fees are typically bundled. Never stream Spotify, Apple Music, or YouTube audio — those licenses explicitly prohibit rebroadcast. The safest path at scale is a platform like Epidemic Sound's "Unlimited" creator plan, which covers streaming up to a defined monthly listener threshold.
Q: Can I run multiple stations with different personas from one codebase?
Yes — and this is exactly how you'd replicate the full Andon FM four-station setup. All persona-specific configuration lives in environment variables:
# Station 1: Upbeat pop
MODEL=gpt-4o PERSONA="You are DJ Nova, an upbeat pop music host." \
ICECAST_URL=icecast://source:pass@localhost:8000/nova \
python scheduler/broadcast_loop.py
# Station 2: Indie/alternative
MODEL=claude-opus-4-5 PERSONA="You are Iris, a thoughtful indie music curator." \
ICECAST_URL=icecast://source:pass@localhost:8000/iris \
python scheduler/broadcast_loop.py
Each station gets its own Chroma collection (collection_name=os.environ['STATION_ID']), its own SQLite ledger file, and its own Icecast mount point. You can orchestrate all four with Docker Compose, one container per station, sharing only the Icecast server.