Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Empty file.
100 changes: 100 additions & 0 deletions examples/misinformation_spread/misinformation_spread/agents.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,100 @@
import mesa
from mesa.discrete_space.cell_agent import BasicMovement, HasCell
from mesa_llm.llm_agent import LLMAgent
from mesa_llm.reasoning.react import ReActReasoning

from misinformation_spread.tools import ( # noqa: F401 — import so @tool registers them
challenge_rumor,
check_neighbors,
spread_rumor,
update_belief,
)


class CitizenAgent(LLMAgent, HasCell, BasicMovement):
def __init__(self, model, name, persona, initial_stance, initial_belief):
super().__init__(
model=model,
reasoning=ReActReasoning,
llm_model=model.llm_model,
system_prompt=(
f"You are {name}, a citizen in a small community. {persona}"
),
)
# Remove built-in movement tools that confuse the small LLM
# We only want our custom tools (check_neighbors, spread_rumor,
# challenge_rumor, update_belief) for communication
for tool_name in ["move_one_step", "teleport_to_location", "speak_to"]:
self.tool_manager.tools.pop(tool_name, None)

self.name = name
self.stance = initial_stance
self.belief_score = initial_belief

def step(self):
prompt = (
f"You are currently a {self.stance} with belief score {self.belief_score:.2f}.\n"
f'The rumor is: "{self.model.rumor}"\n\n'
f"You MUST follow these steps in order:\n"
f"Step 1: Call check_neighbors to see who is nearby. Look at the agent IDs in the result.\n"
f"Step 2: Pick ONE agent ID from the check_neighbors result. "
f"If your belief score is above 0.5, call spread_rumor with that agent's ID. "
f"If your belief score is 0.5 or below, call challenge_rumor with that agent's ID. "
f"IMPORTANT: Only use agent IDs that appeared in the check_neighbors result.\n"
f"Step 3: Call update_belief with a new score. "
f"If you spread the rumor, increase your score slightly (add 0.05). "
f"If you challenged it, decrease your score slightly (subtract 0.05).\n"
)
plan = self.reasoning.plan(prompt=prompt)
self.apply_plan(plan)

# Workaround: small LLMs (e.g. Gemma 3 1B) struggle with multi-step
# ReAct tool calling — they often skip update_belief or hallucinate
# agent IDs. We apply belief updates programmatically based on which
# communication tool the LLM actually invoked.
# The plan object stores tool calls at plan.llm_plan.tool_calls,
# where each entry is a ChatCompletionMessageToolCall with
# .function.name for the tool name.
tool_calls = getattr(getattr(plan, "llm_plan", None), "tool_calls", None)
if tool_calls:
called_tools = {tc.function.name for tc in tool_calls}
if "spread_rumor" in called_tools:
self.belief_score = min(1.0, self.belief_score + 0.05)
elif "challenge_rumor" in called_tools:
self.belief_score = max(0.0, self.belief_score - 0.05)

# Update stance based on new score
if self.belief_score > 0.7:
self.stance = "believer"
elif self.belief_score < 0.3:
self.stance = "skeptic"
else:
self.stance = "neutral"


class RuleBasedAgent(mesa.Agent, HasCell, BasicMovement):
def __init__(self, model, name, persona, initial_stance, initial_belief):
super().__init__(model=model)
self.name = name
self.stance = initial_stance
self.belief_score = initial_belief

def step(self):
neighbors = [agent for cell in self.cell.neighborhood for agent in cell.agents]

for neighbor in neighbors:
if neighbor.stance == "believer":
self.belief_score += 0.1 * (1 - self.belief_score)
elif neighbor.stance == "skeptic":
self.belief_score -= 0.1 * self.belief_score
else:
self.belief_score += self.random.uniform(-0.02, 0.02)

self.belief_score = max(0.0, min(1.0, self.belief_score))

if self.belief_score > 0.7:
self.stance = "believer"
elif self.belief_score < 0.3:
self.stance = "skeptic"
else:
self.stance = "neutral"
242 changes: 242 additions & 0 deletions examples/misinformation_spread/misinformation_spread/model.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,242 @@
import mesa
from dotenv import load_dotenv
from mesa.datacollection import DataCollector
from mesa.discrete_space import OrthogonalMooreGrid

from misinformation_spread.agents import CitizenAgent, RuleBasedAgent


class MisinformationModel(mesa.Model):
def __init__(self, width=5, height=5, llm_model="ollama/llama3.2:3b"):
super().__init__()

load_dotenv()

self.llm_model = llm_model
self.rumor = (
"The town's water supply has been contaminated with dangerous "
"chemicals from the nearby factory."
)

self.grid = OrthogonalMooreGrid(
(width, height), capacity=1, torus=True, random=self.random
)

agent_configs = [
{
"name": "Maria",
"persona": "A cautious schoolteacher who values evidence and critical thinking. You don't believe things easily.",
"initial_stance": "skeptic",
"initial_belief": 0.2,
},
{
"name": "Carlos",
"persona": "An anxious shopkeeper who worries about health risks. You tend to believe warnings about safety.",
"initial_stance": "believer",
"initial_belief": 0.8,
},
{
"name": "Priya",
"persona": "A young university student studying chemistry. You trust scientific sources over gossip.",
"initial_stance": "skeptic",
"initial_belief": 0.15,
},
{
"name": "James",
"persona": "A retired factory worker who distrusts the factory owners. You've always suspected they cut corners on safety.",
"initial_stance": "believer",
"initial_belief": 0.85,
},
{
"name": "Aisha",
"persona": "A community health worker who has seen real contamination cases before. You take such claims seriously but want proof.",
"initial_stance": "neutral",
"initial_belief": 0.5,
},
{
"name": "Tom",
"persona": "A local journalist always looking for a story. You're curious but need verification before reporting.",
"initial_stance": "neutral",
"initial_belief": 0.45,
},
{
"name": "Lin",
"persona": "A grandmother who has lived here for 50 years. You trust your instincts and community word-of-mouth.",
"initial_stance": "neutral",
"initial_belief": 0.55,
},
{
"name": "David",
"persona": "A government water inspector who knows the water is tested regularly. You're confident the water is safe.",
"initial_stance": "skeptic",
"initial_belief": 0.1,
},
{
"name": "Sofia",
"persona": "A social media enthusiast who shares things quickly. You tend to amplify information without checking.",
"initial_stance": "believer",
"initial_belief": 0.75,
},
{
"name": "Raj",
"persona": "A doctor at the local clinic. You rely on medical data and are skeptical of unverified health claims.",
"initial_stance": "skeptic",
"initial_belief": 0.2,
},
{
"name": "Emma",
"persona": "A stay-at-home parent concerned about children's health. You err on the side of caution.",
"initial_stance": "neutral",
"initial_belief": 0.6,
},
{
"name": "Mike",
"persona": "A laid-back bartender who hears all sorts of gossip. You don't take rumors seriously.",
"initial_stance": "neutral",
"initial_belief": 0.4,
},
]

for config in agent_configs:
agent = CitizenAgent(
model=self,
name=config["name"],
persona=config["persona"],
initial_stance=config["initial_stance"],
initial_belief=config["initial_belief"],
)
agent.move_to(self.grid.select_random_empty_cell())

self.datacollector = DataCollector(
model_reporters={
"believers": lambda m: sum(
1 for a in m.agents if a.stance == "believer"
),
"skeptics": lambda m: sum(1 for a in m.agents if a.stance == "skeptic"),
"neutrals": lambda m: sum(1 for a in m.agents if a.stance == "neutral"),
"avg_belief": lambda m: sum(a.belief_score for a in m.agents)
/ len(m.agents),
},
agent_reporters={
"belief_score": "belief_score",
"stance": "stance",
},
)

def step(self):
self.agents.shuffle_do("step")
self.datacollector.collect(self)


class RuleBasedModel(mesa.Model):
def __init__(self, width=5, height=5):
super().__init__()

self.grid = OrthogonalMooreGrid(
(width, height), capacity=1, torus=True, random=self.random
)

agent_configs = [
{
"name": "Maria",
"persona": "A cautious schoolteacher who values evidence and critical thinking.",
"initial_stance": "skeptic",
"initial_belief": 0.2,
},
{
"name": "Carlos",
"persona": "An anxious shopkeeper who worries about health risks.",
"initial_stance": "believer",
"initial_belief": 0.8,
},
{
"name": "Priya",
"persona": "A young university student studying chemistry.",
"initial_stance": "skeptic",
"initial_belief": 0.15,
},
{
"name": "James",
"persona": "A retired factory worker who distrusts the factory owners.",
"initial_stance": "believer",
"initial_belief": 0.85,
},
{
"name": "Aisha",
"persona": "A community health worker.",
"initial_stance": "neutral",
"initial_belief": 0.5,
},
{
"name": "Tom",
"persona": "A local journalist always looking for a story.",
"initial_stance": "neutral",
"initial_belief": 0.45,
},
{
"name": "Lin",
"persona": "A grandmother who has lived here for 50 years.",
"initial_stance": "neutral",
"initial_belief": 0.55,
},
{
"name": "David",
"persona": "A government water inspector.",
"initial_stance": "skeptic",
"initial_belief": 0.1,
},
{
"name": "Sofia",
"persona": "A social media enthusiast who shares things quickly.",
"initial_stance": "believer",
"initial_belief": 0.75,
},
{
"name": "Raj",
"persona": "A doctor at the local clinic.",
"initial_stance": "skeptic",
"initial_belief": 0.2,
},
{
"name": "Emma",
"persona": "A stay-at-home parent concerned about children's health.",
"initial_stance": "neutral",
"initial_belief": 0.6,
},
{
"name": "Mike",
"persona": "A laid-back bartender who hears all sorts of gossip.",
"initial_stance": "neutral",
"initial_belief": 0.4,
},
]

for config in agent_configs:
agent = RuleBasedAgent(
model=self,
name=config["name"],
persona=config["persona"],
initial_stance=config["initial_stance"],
initial_belief=config["initial_belief"],
)
agent.move_to(self.grid.select_random_empty_cell())

self.datacollector = DataCollector(
model_reporters={
"believers": lambda m: sum(
1 for a in m.agents if a.stance == "believer"
),
"skeptics": lambda m: sum(1 for a in m.agents if a.stance == "skeptic"),
"neutrals": lambda m: sum(1 for a in m.agents if a.stance == "neutral"),
"avg_belief": lambda m: sum(a.belief_score for a in m.agents)
/ len(m.agents),
},
agent_reporters={
"belief_score": "belief_score",
"stance": "stance",
},
)

def step(self):
self.agents.shuffle_do("step")
self.datacollector.collect(self)
Loading