Agent Development¶
This document mainly introduces how to design and implement custom agents in AgentSociety.
Part 1: Agent Positioning in AgentSociety¶
Core Positioning of Agents¶
In AgentSociety, agents are autonomous execution units in urban environments. Each agent represents an entity that can operate independently and make autonomous decisions in an urban environment, forming the core components of the entire urban simulation system.
Agents are not just simple program modules, but complex systems with the following characteristics:
Autonomy: Agents can make independent decisions based on their own state and environmental information
Persistence: Agents continuously exist throughout the simulation process, maintaining their own state and history
Interactivity: Agents can interact with other agents and environmental systems
Adaptability: Agents can adjust their behavioral strategies according to environmental changes
Basic Responsibilities of Agents¶
As autonomous execution units in urban environments, agents undertake the following core responsibilities:
1. Autonomous Decision-Making: Agents need to autonomously decide on their next action based on current state, historical experience, and environmental information. This decision-making process may involve:
Requirement analysis: Identifying current problems to be solved
Plan formulation: Developing action plans to achieve goals
Action execution: Converting plans into specific actions
2. Environmental Interaction: Agents need to continuously interact with the urban environment, including:
Environmental perception: Obtaining relevant information from the environment
State update: Updating own state based on environmental changes
Behavioral feedback: Influencing environmental state through actions
3. State Maintenance: Agents need to maintain their internal state, including:
Real-time state: Current emotions, location, activities, etc.
Historical memory: Past experiences and decisions
Knowledge accumulation: Knowledge learned from experience
Part 2: Core Agent Workflow¶
The agent workflow is the core mechanism of agent operation, defining how agents respond to and execute tasks. Understanding the workflow is crucial for developing effective agents.
Active Workflow: run() Method¶
The agent’s main workflow is implemented through the run() method, which serves as a unified entry point coordinating the entire execution process.
Agent Workflow Controlled by run() Method¶
async def run(self) -> Any:
"""
Unified entry point for executing the agent's logic.
- **Description**:
- It calls the forward method to execute the agent's behavior logic.
- Acts as the main control flow for the agent, coordinating when and how the agent performs its actions.
"""
start_time = time.time()
# run required methods before agent forward
await self.before_forward()
await self.before_blocks()
# run agent forward
await self.forward()
# run required methods after agent forward
await self.after_blocks()
await self.after_forward()
await self.status_summary()
end_time = time.time()
return end_time - start_time
Important Note: The run() method is the unified entry point for agents and should not be modified. It calls various lifecycle methods in a fixed order.
Core Methods That Must Be Implemented¶
forward() Method
@abstractmethod
async def forward(self) -> Any:
"""
Define the behavior logic of the agent.
- **Description**:
- This abstract method should contain the core logic for what the agent does at each step of its operation.
- It is intended to be overridden by subclasses to define specific behaviors.
"""
raise NotImplementedError
The forward() method is the core behavioral logic of the agent and must be implemented by subclasses. It contains the agent’s main decision-making and behavioral logic.
before_forward() and after_forward() Methods
async def before_forward(self):
"""
Before forward - prepare context and environment
"""
pass
async def after_forward(self):
"""
After forward - cleanup and save state
"""
pass
These two methods are optional and are used for preparation before execution and cleanup after execution.
Lifecycle Execution Methods Related to Blocks¶
async def before_blocks(self):
"""
Before blocks - prepare all blocks
"""
if self.blocks is None:
return
for block in self.blocks:
await block.before_forward()
async def after_blocks(self):
"""
After blocks - cleanup all blocks
"""
if self.blocks is None:
return
for block in self.blocks:
await block.after_forward()
These two methods automatically call the before_forward() and after_forward() methods of all registered Blocks to ensure Block lifecycle management. Do not modify. For Block-related content, please refer to the following sections.
Passive Response Workflow: react_to_intervention()¶
Agents need to respond to external interventions, which is implemented through the react_to_intervention() method.
async def react_to_intervention(self, intervention_message: str):
"""
React to an intervention.
- **Args**:
- `intervention_message` (`str`): The message of the intervention.
- **Description**:
- React to an intervention from external sources.
"""
# Parse intervention message
intervention_data = json.loads(intervention_message)
# Update agent behavior based on intervention
if intervention_data.get("type") == "policy_change":
await self.memory.status.update("policy_awareness", True)
await self.memory.stream.add(
topic="intervention",
description=f"Received policy intervention: {intervention_data.get('content')}"
)
# Adjust behavior accordingly
await self.adjust_behavior_for_intervention(intervention_data)
Important Note: The react_to_intervention() method is mandatory to implement and is used to handle external interventions.
Passive Response Workflow of CitizenAgentBase¶
In addition to the standard run() workflow, CitizenAgentBase provides several core passive response methods, all of which have default implementations and are optional implementation items.
do_chat() Method¶
Used to respond to social messages from other agents
async def do_chat(self, message: Message) -> str:
"""
Process a chat message received from another agent.
- **Args**:
- `message` (`Message`): The chat message data received from another agent.
- **Returns**:
- `str`: Response to the chat message
"""
# Default implementation
resp = f"Agent {self.id} received agent chat response: {message.payload}"
get_logger().debug(resp)
return resp
Features:
Has default implementation and can be used directly
Can be overridden by subclasses to provide custom chat response logic
Automatically handles message storage and logging
do_survey() Method¶
Used to respond to questionnaire surveys
async def do_survey(self, survey: Survey) -> str:
"""
Process a survey questionnaire.
- **Args**:
- `survey` (`Survey`): The survey questionnaire to respond to.
- **Returns**:
- `str`: Survey response based on agent's memory and background
"""
# Get survey questions
questions = survey.to_prompt()
# Generate response based on agent's memory
response = await self.llm.atext_request([
{"role": "system", "content": "You are a citizen, please answer based on your background"},
{"role": "user", "content": questions[0]}
])
return response
Features:
Generates survey responses based on agent memory
Automatically handles storage of survey data
Can be overridden to provide more complex response logic
do_interview() Method¶
Used to respond to interviews
async def do_interview(self, question: str) -> str:
"""
Process an interview question.
- **Args**:
- `question` (`str`): The interview question.
- **Returns**:
- `str`: Interview response based on agent's background
"""
# Get agent background
background = await self.memory.status.get("background_story")
# Generate interview response
response = await self.llm.atext_request([
{"role": "system", "content": f"You are {background}, please answer the interview question"},
{"role": "user", "content": question}
])
return response
Features:
Generates interview responses based on agent background stories
Supports deep communication, focusing more on personal experiences than questionnaire surveys
Can be overridden to provide more personalized responses
Workflow Execution Order¶
The agent workflow execution follows the following order:
Active Workflow (triggered by run()):
before_forward()- Preparation workbefore_blocks()- Block preparation workforward()- Core behavioral logicafter_blocks()- Block cleanup workafter_forward()- Cleanup work
Passive Response Workflow (triggered by external events):
react_to_intervention()- Respond to interventionsdo_chat()- Respond to chat messagesdo_survey()- Respond to questionnaire surveysdo_interview()- Respond to interview questions
5. 工作流设计原则¶
Active Workflow Design Principles¶
Single Responsibility: Each lifecycle method is only responsible for specific functions
Extensibility: New preparation or cleanup work can be easily added
Error Handling: Exceptions must be properly handled at each stage
State Management: Ensure consistency of state across all stages
Passive Response Workflow Design Principles¶
Responsiveness: Quickly respond to external events
Consistency: Maintain state consistency with the active workflow
Customizability: Allow subclasses to override to provide specific behaviors
Data Integrity: Ensure complete recording of response data
Part 3: Core Agent Subsystems¶
The agent design in AgentSociety is based on four core elements, each with its unique design principles and value. Understanding the design concepts of these core elements is crucial for developing high-quality agents.
Memory System Design Principles¶
The agent’s memory system is key to an agent’s ability to maintain continuity and learning capabilities. AgentSociety designs two different types of memory, each with specific uses and advantages.
Status Memory: Real-time State Storage¶
Status Memory stores the agent’s real-time state in key-value pairs with the following characteristics:
Fast Access: Direct access via key names with fast response times
Structured Storage: Each state has a clear type and default value
Real-time Updates: States can be updated at any time to reflect the agent’s current condition
Embeddability: Supports vectorized storage for semantic retrieval
Status Memory is primarily used to store:
Basic attributes of agents (name, age, occupation, etc.)
Current state (mood, location, activities, etc.)
Real-time data (energy level, satisfaction, etc.)
Status Memory Function Description¶
Defining State Attributes
Define the state attributes contained in the agent through the StatusAttributes class variable in the agent class:
class MyAgent(Agent):
StatusAttributes = [
MemoryAttribute(
name="mood",
type=str,
default_or_value="happy",
description="Agent's current mood",
whether_embedding=True,
),
MemoryAttribute(
name="energy",
type=float,
default_or_value=0.8,
description="Agent's energy level, 0-1",
),
MemoryAttribute(
name="current_activity",
type=str,
default_or_value="idle",
description="Agent's current activity",
),
]
Getting State Values
Get state values in the agent through the memory.status.get() method:
async def forward(self):
# Get current mood and energy
mood = await self.memory.status.get("mood")
energy = await self.memory.status.get("energy")
Updating State Values
Update state values in the agent through the memory.status.update() method:
async def update_status(self):
# Update mood based on recent events
await self.memory.status.update("mood", "excited")
Stream Memory: Streaming Memory¶
Stream Memory records agents’ experiences over time through streaming storage with the following characteristics:
Sequentiality: Records events and experiences in chronological order
Richness: Can store complex textual descriptions and cognitive processes
Searchability: Supports semantic retrieval to find relevant historical experiences
Learnability: Guides future decisions through historical experiences
Stream Memory is primarily used to store:
Agent experiences and memories
Decision-making and thinking processes
Interaction history with other agents
Event information obtained from the environment
Stream Memory Function Description¶
Adding Memory Entries
Add new memory entries in the agent through the memory.stream.add() method:
async def record_experience(self, event: str, thought: str):
# Record a new experience
await self.memory.stream.add(
topic=f"Event",
description="I met my friend"
)
Retrieving Relevant Memories
Find relevant historical memories through semantic retrieval:
async def recall_related_memories(self, query: str, limit: int = 5):
# Search for memories related to the query
memories = await self.memory.stream.search(
query=query,
topic:Optional[str]='Event',
top_k=limit
)
return memories
Block System Design Principles¶
The Block system is a key design in AgentSociety for implementing complex agent behaviors. Blocks are similar to layers in neural networks, with each Block responsible for specific functional modules.
Relationship Between Block and Agent¶
Agent is a container: Agent is responsible for coordinating and managing multiple Blocks
Block is a functional module: Each Block focuses on specific functions
Compositional design: Complex agent behaviors are built by combining different Blocks
Reusability: Blocks can be reused across different agents
The Significance of Blocks¶
The Block system design solves the following problems:
Modular development: Decompose complex behaviors into independent functional modules
Code reuse: The same functions can be reused in different agents
Easy testing: Each Block can be tested independently
Flexible combination: Different Blocks can be combined as needed
Block Design Principles¶
Single responsibility: Each Block is only responsible for one specific function
Composability: Blocks can be flexibly combined with each other
Testability: Each Block can be tested independently
Extensibility: New Blocks can be easily added
Block Function Description¶
Creating Custom Blocks
Create custom Blocks by inheriting from the Block base class:
class MyBlockParams(BlockParams):
threshold: float = 0.5
max_iterations: int = 10
class MyBlockOutput(BlockOutput):
success: bool = True
result: str = ""
confidence: float = 0.0
class MyCustomBlock(Block):
ParamsType = MyBlockParams
OutputType = MyBlockOutput
name = "my_custom_block"
description = "A custom block for specific functionality"
async def forward(self, agent_context):
# Get parameters
threshold = self.params.threshold
max_iterations = self.params.max_iterations
# Access agent memory
current_mood = await self.memory.status.get("mood")
# Perform block-specific logic
result = await self.process_logic(agent_context, threshold)
# Return output
return MyBlockOutput(
success=True,
result=result,
confidence=0.8
)
async def process_logic(self, context, threshold):
# Implement specific logic here
return "Processed result"
Block Lifecycle
Each Block has a complete lifecycle, including before_forward, forward, and after_forward parts, of which forward must be implemented:
class LifecycleBlock(Block):
async def before_forward(self): # Optional
"""Called before forward execution"""
# Prepare context, validate inputs
pass
async def forward(self, agent_context): # You have to rewrite the forward function
"""Main execution logic"""
# Core block functionality
return result
async def after_forward(self): # Optional
"""Called after forward execution"""
# Cleanup, update state
pass
Integration of Block and Agent
Option 1: Using Blocks directly in Agents:
class MyAgent(Agent):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Create and add blocks
self.analysis_block = MyCustomBlock(
toolbox=self.toolbox,
agent_memory=self.memory,
block_params=MyBlockParams(threshold=0.7)
)
# Set agent reference for blocks that need it
self.analysis_block.set_agent(self)
async def forward(self):
# Use blocks in agent logic
context = self.context
# Execute analysis block
analysis_result = await self.analysis_block.forward(context)
# Use block output for decision making
if analysis_result.success and analysis_result.confidence > 0.8:
# High confidence result, proceed with action
pass
Option 2: Registering multiple Blocks with the dispatcher:
class DispatcherAgent(Agent):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Create multiple blocks
self.news_collector = NewsCollectorBlock(
toolbox=self.toolbox,
agent_memory=self.memory,
block_params=NewsCollectorParams(sources=["rss", "api"])
)
self.news_analyzer = NewsAnalyzerBlock(
toolbox=self.toolbox,
agent_memory=self.memory,
block_params=NewsAnalyzerParams(importance_threshold=0.7)
)
self.news_distributor = NewsDistributorBlock(
toolbox=self.toolbox,
agent_memory=self.memory,
block_params=NewsDistributorParams(distribution_channels=["social"])
)
# Set agent reference for blocks that need it
for block in [self.news_collector, self.news_analyzer, self.news_distributor]:
if block.NeedAgent:
block.set_agent(self)
# Register blocks to dispatcher
self.dispatcher.register_blocks([
self.news_collector,
self.news_analyzer,
self.news_distributor
])
async def forward(self):
# Update context with current intention
self.context.current_intention = "collect and analyze news"
# Let dispatcher automatically select the most appropriate block
selected_block = await self.dispatcher.dispatch(self.context)
if selected_block:
# Execute the selected block
result = await selected_block.forward(self.context)
# Process the result
if result.success:
# Update context with block results
self.context.last_block_result = result
# Record the activity
await self.memory.stream.add(
topic="activity",
description=f"Executed {selected_block.name}: {result.evaluation}"
)
else:
# Handle block failure
await self.memory.stream.add(
topic="error",
description=f"Block {selected_block.name} failed: {result.error}"
)
else:
# No suitable block found
await self.memory.stream.add(
topic="activity",
description=f"No suitable block found for: {self.context.current_intention}"
)
return "Agent behavior completed"
Dispatcher Working Principle
The Dispatcher intelligently selects the most suitable Block to execute tasks based on the Block’s name and description through LLM. The Dispatcher uses customizable prompt templates and automatically retrieves formatted variables from the context:
Block Registration: Each Block needs to provide a clear
nameanddescriptionwhen registeringPrompt Template: Uses customizable prompt templates that support retrieving variables from the context
Intelligent Selection: LLM performs semantic matching based on the prompt and Block description to select the most suitable Block
Customizing Dispatcher Prompt
# the default prompt template
DEFAULT_DISPATCHER_PROMPT = """
Based on the task information (which describes the needs of the user), select the most appropriate block to handle the task.
Each block has its specific functionality as described in the function schema.
Task information:
${context.current_intention}
"""
# define your dispatcher prompt
CUSTOM_DISPATCHER_PROMPT = """
Based on the current situation and agent state, select the most appropriate block to handle the task.
Current situation:
- Agent mood: ${context.current_mood}
- Current activity: ${context.current_activity}
- Task priority: ${context.task_priority}
- Available time: ${context.available_time}
Task information:
${context.current_intention}
Select the block that best matches the current situation and task requirements.
"""
# register your prompt
class CustomDispatcherAgent(Agent):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# register prompt
self.dispatcher.register_dispatcher_prompt(CUSTOM_DISPATCHER_PROMPT)
# register blocks
self.dispatcher.register_blocks([...])
Importance of Block Description
class WellDescribedBlock(Block):
name = "news_collector" # A clear name
description = "Collects news from RSS feeds and APIs, filters content based on keywords, and returns structured news data" # A clear description helps the LLM to make decisions
async def forward(self, agent_context):
# Block implementation
pass
Advantages of the Dispatcher
Flexible Configuration: Supports customizable prompt templates
Rich Context: Can retrieve any variables from the context
Semantic Understanding: LLM can understand complex contextual information
Intelligent Matching: Selects the most suitable Block based on the complete context
Automatic Expansion: No need to modify selection logic when adding new Blocks
Intelligent Fallback: Returns None instead of an error when no suitable Block is found
Differences Between the Two Options
Features |
Option 1: Direct Usage |
Option 2: Dispatcher |
|---|---|---|
Control Method |
Manually control Block execution order |
Automatically select the most suitable Block |
Flexibility |
High, can precisely control execution logic |
Medium, depends on LLM selection |
Complexity |
Requires manual management of Block calls |
Automatically managed, cleaner code |
Applicable Scenarios |
Clear execution flow |
Dynamic task assignment |
Debugging Difficulty |
Easy to debug, clear process |
Need to understand dispatcher logic |
Block Parameter Configuration
Blocks support flexible parameter configuration:
# Configure block with specific parameters
analysis_block = MyCustomBlock(
toolbox=toolbox,
agent_memory=memory,
block_params=MyBlockParams(
threshold=0.8,
max_iterations=15
)
)
# Access parameters in block
threshold = self.params.threshold
max_iterations = self.params.max_iterations
Tool Collection Design Principles¶
AgentToolbox provides agents with a unified core tool collection, each tool having its specific role and value.
LLM Tool: The Agent’s Brain¶
The LLM tool is the core of agent reasoning and decision-making:
Natural Language Understanding: Understanding input natural language
Reasoning Ability: Logical reasoning based on context
Generation Ability: Generating natural language responses
Knowledge Application: Applying existing knowledge to solve problems
Environment Tool: Environmental Perception and Interaction¶
The Environment tool allows agents to perceive and influence the environment:
Environmental Perception: Obtaining information from the environment (weather, location, other agents, etc.)
State Query: Querying various states in the environment
Action Execution: Executing specific actions in the environment
Feedback Acquisition: Obtaining results and feedback from actions
Accessing Core Tools¶
Agents can directly access core tools in the toolbox:
class MyAgent(Agent):
async def forward(self):
# Access LLM for reasoning
response = await self.llm.atext_request([
{"role": "system", "content": "You are a helpful assistant"},
{"role": "user", "content": "What should I do next?"}
])
# Access environment for information
current_time = self.environment.get_datetime()
weather = self.environment.sense("weather")
Context System Design Principles¶
The context system not only provides a unified context entry for agents’ long-term execution, but also offers a flexible information transfer mechanism between agents and Blocks.
AgentContext: Agent-level Context¶
AgentContext is used to maintain agent-level context information:
Global State: Agent’s global state information
Environmental Information: Information obtained from the environment
Configuration Parameters: Agent’s configuration parameters
Execution Results: Results of agent execution
BlockContext: Module-level Context¶
BlockContext is used to pass information between Blocks:
Input Data: Block’s input data
Processing Results: Block’s processing results
Intermediate State: Intermediate states during Block processing
Error Information: Error information during Block execution
Role of Context¶
The context system solves the following problems:
Information Transfer: Passing information between different components
State Management: Managing the state of agents and Blocks
Parameter Configuration: Configuring parameters for agents and Blocks
Result Return: Returning execution results and error information
DotDict Design¶
The context system is implemented based on DotDict, providing the convenience of dot notation access:
Attribute-style Access: Accessing dictionary elements using dot notation
Nested Support: Supporting nested dictionary structures
Merge Operation: Supporting dictionary merge operations
Deep Copy: Automatically performing deep copy to avoid side effects
Context System Function Description¶
Defining AgentContext
Creating a custom AgentContext class:
class MyAgentContext(AgentContext):
current_time: str = ""
current_location: str = ""
current_mood: str = ""
recent_events: list[str] = []
decision_history: list[dict] = []
class Config:
arbitrary_types_allowed = True
Defining BlockContext
Creating a custom BlockContext class:
class MyBlockContext(BlockContext):
input_data: str = ""
processing_stage: str = "initial"
intermediate_results: list = []
error_message: str = ""
confidence_score: float = 0.0
Using Context in Agents
class MyAgent(Agent):
Context = MyAgentContext
async def before_forward(self):
# Update context with current information
self.context.current_time = self.environment.get_datetime()
self.context.current_location = self.environment.get_location()
self.context.current_mood = await self.memory.status.get("mood")
# Add recent events to context
recent_memories = await self.memory.stream.search("", limit=5)
self.context.recent_events = [mem.content for mem in recent_memories]
async def forward(self):
# Use context information for decision making
if self.context.current_mood == "happy":
# Agent is happy, can work efficiently
pass
elif len(self.context.recent_events) > 0:
# Process recent events
await self.process_recent_events(self.context.recent_events)
Using Context in Blocks
class MyBlock(Block):
Context = MyBlockContext
async def forward(self, agent_context):
# Initialize block context
self.context.input_data = agent_context.get("input", "")
self.context.processing_stage = "started"
try:
# Process input data
result = await self.process_data(self.context.input_data)
self.context.intermediate_results.append(result)
# Update processing stage
self.context.processing_stage = "completed"
self.context.confidence_score = 0.9
return result
except Exception as e:
# Handle errors
self.context.error_message = str(e)
self.context.processing_stage = "error"
self.context.confidence_score = 0.0
raise
Using DotDict
DotDict provides convenient dot notation access:
# Create DotDict
context = DotDict({
"user": {
"name": "Alice",
"preferences": {
"color": "blue",
"food": "pizza"
}
},
"session": {
"start_time": "2024-01-01 10:00:00"
}
})
# Access using dot notation
user_name = context.user.name # "Alice"
user_color = context.user.preferences.color # "blue"
session_time = context.session.start_time # "2024-01-01 10:00:00"
# Update values
context.user.preferences.food = "sushi"
# Merge with another DotDict
additional_context = DotDict({
"user": {
"age": 25
},
"system": {
"version": "1.0"
}
})
# Merge contexts
merged_context = context | additional_context
# Now merged_context.user.age = 25, merged_context.system.version = "1.0"
Context Passing Example
class ContextAwareAgent(Agent):
Context = MyAgentContext
async def forward(self):
# Prepare agent context
await self.prepare_context()
# Pass context to blocks
for block in self.blocks:
try:
result = await block.forward(self.context)
# Update context with block results
self.context.decision_history.append({
"block": block.name,
"result": result,
"timestamp": self.context.current_time
})
except Exception as e:
# Handle block errors
self.context.decision_history.append({
"block": block.name,
"error": str(e),
"timestamp": self.context.current_time
})
async def prepare_context(self):
# Gather all necessary information
self.context.current_time = self.environment.get_datetime()
self.context.current_location = self.environment.get_location()
self.context.current_mood = await self.memory.status.get("mood")
# Get recent memories for context
recent_memories = await self.memory.stream.search("", limit=3)
self.context.recent_events = [mem.content for mem in recent_memories]
Part 4: Development Examples¶
This section demonstrates how to build different types of agents in AgentSociety through specific development examples. Each example includes complete requirement analysis, design concepts, and implementation code.
Agent Types and Base Class Selection¶
Before starting development, it’s necessary to select an appropriate base class based on the agent’s functions and responsibilities:
Entity Type |
Corresponding Base Class |
Main Functions |
Applicable Scenarios |
|---|---|---|---|
City Residents |
CitizenAgentBase |
Traffic simulation, economic system binding, daily behaviors |
Simulating daily life behaviors of ordinary citizens |
Enterprise Institutions |
FirmAgentBase |
Production and operation, market interaction, decision making |
Simulating enterprise business decisions |
Banking Institutions |
BankAgentBase |
Financial services, fund management, risk assessment |
Simulating bank financial services |
Government Institutions |
GovernmentAgentBase |
Policy making, public services, regulatory functions |
Simulating government policy making |
Central Bank Institutions |
NBSAgentBase |
Monetary policy, financial regulation, macro control |
Simulating central bank monetary policy |
Other Institutions |
InstitutionAgentBase |
General institutional functions, organizational management |
Simulating other types of institutions |
Development Process Overview¶
Agent development follows the following basic process:
Requirement Analysis: Clarify the agent’s functional requirements and behavioral characteristics
Base Class Selection: Select an appropriate agent base class based on functionality
Memory Design: Define the content in Status Memory and clarify the role of Stream Memory in agents
Block Design: Decompose complex functions into independent Block modules
Logic Implementation: Implement the agent’s core behavioral logic
Testing and Validation: Verify the agent’s functionality and performance
Complete Development Case: News Dissemination Agent¶
Requirement Analysis¶
Build an agent capable of collecting, analyzing, and disseminating news with the following functions:
Collecting news information from multiple sources
Analyzing the importance and relevance of news content
Deciding whether to disseminate news and how to disseminate it
Tracking the effectiveness of news dissemination
Function Decomposition and Block Design¶
Through requirement analysis, the functions are decomposed into the following Blocks:
NewsCollectorBlock: News collection module
Obtaining news from different sources
Filtering and preprocessing news content
Extracting key information from news
NewsAnalyzerBlock: News analysis module
Analyzing the importance of news
Evaluating the relevance of news
Generating news summaries
NewsDistributorBlock: News dissemination module
Deciding dissemination strategies
Selecting dissemination channels
Tracking dissemination effectiveness
Complete Implementation¶
Step 1: Define Agent Parameters and Context
class NewsAgentParams(AgentParams):
collection_interval: int = 300 # collection interval (s)
analysis_threshold: float = 0.7 # analysis threshold
distribution_radius: int = 1000 # broadcast radius
class NewsAgentContext(AgentContext):
current_news_count: int = 0
last_collection_time: str = ""
distribution_stats: dict = {}
class NewsBlockOutput(BlockOutput):
success: bool = True
news_items: list = []
analysis_results: dict = {}
distribution_results: dict = {}
Step 2: Implement News Collection Block
class NewsCollectorParams(BlockParams):
sources: list[str] = ["rss", "api", "social"]
max_items_per_source: int = 10
filter_keywords: list[str] = []
class NewsCollectorOutput(BlockOutput):
collected_news: list[dict] = []
source_stats: dict = {}
class NewsCollectorBlock(Block):
ParamsType = NewsCollectorParams
OutputType = NewsCollectorOutput
name = "news_collector"
description = "Collects news from various sources"
async def forward(self, agent_context):
collected_news = []
source_stats = {}
# Collect from RSS sources
if "rss" in self.params.sources:
rss_news = await self.collect_from_rss()
collected_news.extend(rss_news)
source_stats["rss"] = len(rss_news)
# Collect from API sources
if "api" in self.params.sources:
api_news = await self.collect_from_api()
collected_news.extend(api_news)
source_stats["api"] = len(api_news)
# Filter news based on keywords
filtered_news = await self.filter_news(collected_news)
return NewsCollectorOutput(
collected_news=filtered_news,
source_stats=source_stats
)
async def collect_from_rss(self):
# Simulate RSS collection
return [
{"title": "Breaking News", "content": "Important event", "source": "rss"},
{"title": "Local Update", "content": "Community news", "source": "rss"}
]
async def collect_from_api(self):
# Simulate API collection
return [
{"title": "Tech News", "content": "Technology update", "source": "api"}
]
async def filter_news(self, news_list):
# Filter based on keywords
filtered = []
for news in news_list:
if any(keyword in news["title"].lower() for keyword in self.params.filter_keywords):
filtered.append(news)
return filtered
Step 3: Implement News Analysis Block
class NewsAnalyzerParams(BlockParams):
importance_threshold: float = 0.6
relevance_keywords: list[str] = []
class NewsAnalyzerOutput(BlockOutput):
analyzed_news: list[dict] = []
importance_scores: dict = {}
class NewsAnalyzerBlock(Block):
ParamsType = NewsAnalyzerParams
OutputType = NewsAnalyzerOutput
name = "news_analyzer"
description = "Analyzes news content for importance and relevance"
async def forward(self, agent_context):
# Get news from previous block
news_items = agent_context.get("collected_news", [])
analyzed_news = []
importance_scores = {}
for news in news_items:
# Analyze importance using LLM
importance_prompt = f"""
Analyze the importance of this news:
Title: {news['title']}
Content: {news['content']}
Rate importance from 0-1 and explain why.
"""
importance_response = await self.llm.atext_request([
{"role": "system", "content": "You are a news analyst"},
{"role": "user", "content": importance_prompt}
])
# Extract importance score (simplified)
importance_score = 0.7 # In real implementation, parse from response
# Check if news meets importance threshold
if importance_score >= self.params.importance_threshold:
analyzed_news.append({
**news,
"importance_score": importance_score,
"should_distribute": True
})
importance_scores[news["title"]] = importance_score
return NewsAnalyzerOutput(
analyzed_news=analyzed_news,
importance_scores=importance_scores
)
Step 4: Implement News Dissemination Block
class NewsDistributorParams(BlockParams):
distribution_channels: list[str] = ["social", "email", "broadcast"]
target_audience: list[int] = []
class NewsDistributorOutput(BlockOutput):
distribution_results: dict = {}
audience_reached: int = 0
class NewsDistributorBlock(Block):
ParamsType = NewsDistributorParams
OutputType = NewsDistributorOutput
name = "news_distributor"
description = "Distributes news to target audience"
async def forward(self, agent_context):
# Get analyzed news
analyzed_news = agent_context.get("analyzed_news", [])
distribution_results = {}
audience_reached = 0
for news in analyzed_news:
if news.get("should_distribute", False):
# Distribute through different channels
for channel in self.params.distribution_channels:
result = await self.distribute_through_channel(news, channel)
distribution_results[f"{news['title']}_{channel}"] = result
audience_reached += result.get("audience_reached", 0)
return NewsDistributorOutput(
distribution_results=distribution_results,
audience_reached=audience_reached
)
async def distribute_through_channel(self, news, channel):
# Simulate distribution
if channel == "social":
# Send to nearby agents
nearby_agents = self.environment.get_nearby_agents(radius=100)
await self.messager.send_message_to_multiple(
agent_ids=nearby_agents,
content=f"News: {news['title']} - {news['content']}",
message_type="news"
)
return {"audience_reached": len(nearby_agents), "channel": channel}
return {"audience_reached": 0, "channel": channel}
Step 5: Integrate into a Complete News Dissemination Agent
class NewsAgent(CitizenAgentBase):
ParamsType = NewsAgentParams
Context = NewsAgentContext
BlockOutputType = NewsBlockOutput
# Define status attributes
StatusAttributes = [
MemoryAttribute(
name="news_collection_count",
type=int,
default_or_value=0,
description="Total number of news items collected",
),
MemoryAttribute(
name="distribution_success_rate",
type=float,
default_or_value=0.0,
description="Success rate of news distribution",
),
MemoryAttribute(
name="last_news_collection",
type=str,
default_or_value="",
description="Timestamp of last news collection",
),
]
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Initialize blocks
self.collector_block = NewsCollectorBlock(
toolbox=self.toolbox,
agent_memory=self.memory,
block_params=NewsCollectorParams(
sources=["rss", "api"],
filter_keywords=["breaking", "important", "urgent"]
)
)
self.analyzer_block = NewsAnalyzerBlock(
toolbox=self.toolbox,
agent_memory=self.memory,
block_params=NewsAnalyzerParams(
importance_threshold=0.7
)
)
self.distributor_block = NewsDistributorBlock(
toolbox=self.toolbox,
agent_memory=self.memory,
block_params=NewsDistributorParams(
distribution_channels=["social", "broadcast"]
)
)
# Set agent reference for blocks
self.collector_block.set_agent(self)
self.analyzer_block.set_agent(self)
self.distributor_block.set_agent(self)
async def forward(self):
# Update context
self.context.current_news_count = await self.memory.status.get("news_collection_count")
self.context.last_collection_time = self.environment.get_datetime()
# Execute news collection
collection_result = await self.collector_block.forward(self.context)
# Update context with collection results
self.context.collected_news = collection_result.collected_news
# Execute news analysis
analysis_result = await self.analyzer_block.forward(self.context)
# Update context with analysis results
self.context.analyzed_news = analysis_result.analyzed_news
# Execute news distribution
distribution_result = await self.distributor_block.forward(self.context)
# Update memory with results
await self.memory.status.update("news_collection_count",
self.context.current_news_count + len(collection_result.collected_news))
# Calculate success rate
total_distributed = len(distribution_result.distribution_results)
if total_distributed > 0:
success_rate = distribution_result.audience_reached / total_distributed
await self.memory.status.update("distribution_success_rate", success_rate)
await self.memory.status.update("last_news_collection", self.context.last_collection_time)
# Record experience in stream memory
await self.memory.stream.add(
content=f"Collected {len(collection_result.collected_news)} news items, "
f"distributed to {distribution_result.audience_reached} audience",
metadata={"type": "news_cycle", "timestamp": self.context.last_collection_time}
)
return NewsBlockOutput(
news_items=collection_result.collected_news,
analysis_results=analysis_result.importance_scores,
distribution_results=distribution_result.distribution_results
)