"""
Pattern 111: Situational Awareness
Description:
Situational Awareness enables agents to maintain comprehensive understanding
of their current context, environment, and state. The pattern continuously
monitors and updates a model of the situation, including relevant entities,
relationships, constraints, opportunities, and threats. This awareness informs
decision-making and enables context-appropriate behavior.
The pattern is crucial for adaptive agents operating in dynamic environments
where context changes frequently. It combines perception, state estimation,
context modeling, and predictive awareness to maintain an accurate and
up-to-date understanding of the situation.
Situational awareness includes environmental monitoring, state tracking,
context interpretation, threat detection, opportunity identification, and
predictive situation assessment.
Key Components:
1. Perception Module: Gathers situational data
2. State Estimator: Tracks current state
3. Context Modeler: Interprets situation meaning
4. Change Detector: Identifies state changes
5. Threat Assessor: Evaluates risks
6. Opportunity Detector: Identifies opportunities
7. Predictor: Anticipates future states
Awareness Dimensions:
- Spatial: Where things are
- Temporal: When things happen
- Causal: Why things occur
- Social: Who is involved
- Functional: What capabilities exist
- Intentional: What goals are active
Use Cases:
- Autonomous vehicles understanding traffic
- Security monitoring systems
- Personal assistants with context awareness
- Game AI with environmental awareness
- Crisis management systems
LangChain Implementation:
Uses memory systems, context tracking, state estimation chains, and
predictive models to maintain situational awareness.
"""
import os
from typing import List, Dict, Any, Optional
from datetime import datetime
from dotenv import load_dotenv
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain.memory import ConversationBufferMemory
load_dotenv()
class SituationStatus(Enum):
"""Enumeration of situation statuses."""
NORMAL = "normal"
ALERT = "alert"
WARNING = "warning"
CRITICAL = "critical"
OPPORTUNITY = "opportunity"
class SituationalAwarenessAgent:
"""
Agent that maintains comprehensive awareness of current situation.
"""
def __init__(self, agent_role: str = "general_agent"):
"""Initialize with role and context."""
self.llm = ChatOpenAI(temperature=0.3, model="gpt-4")
self.role = agent_role
self.situation_memory = ConversationBufferMemory(
memory_key="situation_history",
return_messages=True
)
# Situational state
self.current_context = {}
self.active_entities = {}
self.threats = []
self.opportunities = []
self.status = SituationStatus.NORMAL
def perceive_environment(
self,
observations: Dict[str, Any]
) -> Dict[str, Any]:
"""
Process environmental observations to update awareness.
Args:
observations: Dictionary of observed facts
Returns:
Updated situational model
"""
print(f"\n{'='*60}")
print("PERCEPTION & STATE ESTIMATION")
print(f"{'='*60}")
print(f"Processing {len(observations)} observations...")
# Update context with observations
self.current_context.update(observations)
self.current_context['last_update'] = datetime.now().isoformat()
# Analyze observations for entities
prompt = ChatPromptTemplate.from_messages([
("system", """You are a situational awareness system analyzing observations.
Extract key entities, their states, and relationships from the observations.
Provide a JSON response with:
- entities: List of identified entities with properties
- relationships: List of relationships between entities
- key_facts: Important facts about the situation"""),
("user", "Observations: {observations}")
])
chain = prompt | self.llm | JsonOutputParser()
try:
analysis = chain.invoke({"observations": str(observations)})
# Update entities
if 'entities' in analysis:
for entity in analysis['entities']:
entity_id = entity.get('id', entity.get('name', 'unknown'))
self.active_entities[entity_id] = entity
print(f"Identified {len(self.active_entities)} entities")
print(f"Key facts: {analysis.get('key_facts', [])}")
return {
"perception": analysis,
"context_updated": True,
"timestamp": self.current_context['last_update']
}
except:
# Fallback if JSON parsing fails
return {
"perception": observations,
"context_updated": True,
"timestamp": self.current_context['last_update']
}
def assess_situation(self) -> Dict[str, Any]:
"""
Assess current situation for threats and opportunities.
Returns:
Situation assessment with status and recommendations
"""
print(f"\n{'='*60}")
print("SITUATION ASSESSMENT")
print(f"{'='*60}")
prompt = ChatPromptTemplate.from_messages([
("system", """You are a situational assessment expert. Analyze the current
situation and provide:
1. Overall situation status (normal/alert/warning/critical/opportunity)
2. Identified threats or risks
3. Identified opportunities
4. Recommended actions
5. Confidence in assessment (0-1)
Format as: STATUS | Threats: [...] | Opportunities: [...] | Actions: [...] | Confidence: X.X"""),
("user", """Current Context: {context}
Active Entities: {entities}
Situation History: {history}
Assess this situation comprehensively.""")
])
chain = prompt | self.llm | StrOutputParser()
assessment_text = chain.invoke({
"context": str(self.current_context),
"entities": str(self.active_entities),
"history": self.situation_memory.load_memory_variables({}).get('situation_history', [])
})
# Parse assessment
parts = assessment_text.split('|')
status_text = parts[0].strip().lower()
# Update status
if 'critical' in status_text:
self.status = SituationStatus.CRITICAL
elif 'warning' in status_text:
self.status = SituationStatus.WARNING
elif 'alert' in status_text:
self.status = SituationStatus.ALERT
elif 'opportunity' in status_text:
self.status = SituationStatus.OPPORTUNITY
else:
self.status = SituationStatus.NORMAL
result = {
"status": self.status.value,
"assessment": assessment_text,
"context_size": len(self.current_context),
"entities_tracked": len(self.active_entities),
"timestamp": datetime.now().isoformat()
}
print(f"Status: {self.status.value.upper()}")
print(f"Assessment:\n{assessment_text}")
# Store in memory
self.situation_memory.save_context(
{"input": "situation_assessment"},
{"output": assessment_text}
)
return result
def predict_changes(
self,
time_horizon: str = "near future"
) -> Dict[str, Any]:
"""
Predict likely changes to the situation.
Args:
time_horizon: Time frame for prediction
Returns:
Predictions about situation evolution
"""
print(f"\n{'='*60}")
print("PREDICTIVE AWARENESS")
print(f"{'='*60}")
print(f"Predicting changes for: {time_horizon}")
prompt = ChatPromptTemplate.from_messages([
("system", """You are a predictive analyst. Based on current situation,
predict likely changes and developments.
Provide:
1. Most likely changes
2. Probability estimates
3. Potential triggers
4. Recommended preparations"""),
("user", """Current Situation:
Context: {context}
Entities: {entities}
Current Status: {status}
Predict changes for: {time_horizon}""")
])
chain = prompt | self.llm | StrOutputParser()
predictions = chain.invoke({
"context": str(self.current_context),
"entities": str(self.active_entities),
"status": self.status.value,
"time_horizon": time_horizon
})
result = {
"time_horizon": time_horizon,
"predictions": predictions,
"based_on_entities": len(self.active_entities),
"confidence": "medium" # Could be calculated
}
print(f"Predictions:\n{predictions}")
return result
def context_appropriate_response(
self,
query: str
) -> Dict[str, Any]:
"""
Generate response appropriate to current situational context.
Args:
query: Query or request
Returns:
Context-aware response
"""
print(f"\n{'='*60}")
print("CONTEXT-AWARE RESPONSE")
print(f"{'='*60}")
print(f"Query: {query}")
prompt = ChatPromptTemplate.from_messages([
("system", """You are an agent with full situational awareness.
Respond to queries considering the complete situational context.
Your response should:
1. Acknowledge relevant context
2. Address the query appropriately
3. Consider current status and constraints
4. Provide context-aware recommendations"""),
("user", """Query: {query}
Current Situation:
- Status: {status}
- Context: {context}
- Active Entities: {entities}
Provide a situationally appropriate response.""")
])
chain = prompt | self.llm | StrOutputParser()
response = chain.invoke({
"query": query,
"status": self.status.value,
"context": str(self.current_context)[:500], # Limit size
"entities": str(list(self.active_entities.keys()))
})
result = {
"query": query,
"response": response,
"context_considered": True,
"status_at_response": self.status.value
}
print(f"Response:\n{response}")
return result
def get_situation_summary(self) -> Dict[str, Any]:
"""Get comprehensive summary of current situation."""
return {
"status": self.status.value,
"context": self.current_context,
"entities": self.active_entities,
"role": self.role,
"last_update": self.current_context.get('last_update', 'never')
}
def demonstrate_situational_awareness():
"""Demonstrate situational awareness capabilities."""
print("=" * 60)
print("SITUATIONAL AWARENESS AGENT DEMONSTRATION")
print("=" * 60)
agent = SituationalAwarenessAgent(agent_role="security_monitor")
# Example 1: Initial Perception
print("\n" + "=" * 60)
print("Example 1: Environmental Perception")
print("=" * 60)
observations1 = {
"time": "14:30",
"location": "Building A, Floor 3",
"people_count": 15,
"noise_level": "normal",
"access_points": ["door_301", "door_302"],
"cameras_active": True
}
result1 = agent.perceive_environment(observations1)
# Example 2: Situation Assessment
print("\n" + "=" * 60)
print("Example 2: Situation Assessment")
print("=" * 60)
result2 = agent.assess_situation()
# Example 3: Changed Situation
print("\n" + "=" * 60)
print("Example 3: Detecting Changes")
print("=" * 60)
observations2 = {
"time": "14:35",
"people_count": 25, # Increased
"noise_level": "elevated",
"unauthorized_access_attempt": "door_304",
"alarm_triggered": True
}
result3 = agent.perceive_environment(observations2)
result4 = agent.assess_situation()
# Example 4: Predictive Awareness
print("\n" + "=" * 60)
print("Example 4: Predicting Changes")
print("=" * 60)
result5 = agent.predict_changes("next 15 minutes")
# Example 5: Context-Aware Response
print("\n" + "=" * 60)
print("Example 5: Context-Appropriate Response")
print("=" * 60)
result6 = agent.context_appropriate_response(
"Should I allow the scheduled maintenance crew to enter?"
)
# Summary
print("\n" + "=" * 60)
print("SITUATION SUMMARY")
print("=" * 60)
summary = agent.get_situation_summary()
print(f"Current Status: {summary['status'].upper()}")
print(f"Entities Tracked: {len(summary['entities'])}")
print(f"Context Elements: {len(summary['context'])}")
print(f"Agent Role: {summary['role']}")
print("\n" + "=" * 60)
print("DEMONSTRATION COMPLETE")
print("=" * 60)
print("✓ Processed environmental observations")
print("✓ Maintained situational context")
print("✓ Detected and assessed changes")
print("✓ Predicted future developments")
print("✓ Generated context-aware responses")
if __name__ == "__main__":
demonstrate_situational_awareness()
Agentic Design Pattern: Situational Awareness
Pattern 111: Situational Awareness – Line-by-Line Explanation
Overview
This document provides a detailed line-by-line explanation of the Situational Awareness pattern implementation.
Pattern Description
Pattern 111: Situational Awareness
Description: Situational Awareness enables agents to maintain comprehensive understanding of their current context, environment, and state. The pattern continuously monitors and updates a model of the situation, including relevant entities, relationships, constraints, opportunities, and threats. This awareness informs decision-making and enables context-appropriate behavior.
The pattern is crucial for adaptive agents operating in dynamic environments
where context changes frequently. It combines perception, state estimation,
context modeling, and predictive awareness to maintain an accurate and
up-to-date understanding of the situation.
Situational awareness includes environmental monitoring, state tracking,
context interpretation, threat detection, opportunity identification, and
predictive situation assessment.
Key Components: 1. Perception Module: Gathers situational data 2. State Estimator: Tracks current state 3. Context Modeler: Interprets situation meaning 4. Change Detector: Identifies state changes 5. Threat Assessor: Evaluates risks 6. Opportunity Detector: Identifies opportunities 7. Predictor: Anticipates future states
Awareness Dimensions: – Spatial: Where things are – Temporal: When things happen – Causal: Why things occur – Social: Who is involved – Functional: What capabilities exist – Intentional: What goals are active
Use Cases: – Autonomous vehicles understanding traffic – Security monitoring systems – Personal assistants with context awareness – Game AI with environmental awareness – Crisis management systems
LangChain Implementation: Uses memory systems, context tracking, state estimation chains, and predictive models to maintain situational awareness.
File Statistics
- Total Lines: 433
- Classes: 2
- Functions: 1
- Constants: 0
Module Documentation
"""
Pattern 111: Situational Awareness
Description:
Situational Awareness enables agents to maintain comprehensive understanding
of their current context, environment, and state. The pattern continuously
monitors and updates a model of the situation, including relevant entities,
relationships, constraints, opportunities, and threats. This awareness informs
decision-making and enables context-appropriate behavior.
The pattern is crucial for adaptive agents operating in dynamic environments
where context changes frequently. It combines perception, state estimation,
context modeling, and predictive awareness to maintain an accurate and
up-to-date understanding of the situation.
Situational awareness includes environmental monitoring, state tracking,
context interpretation, threat detection, opportunity identification, and
predictive situation assessment.
Key Components:
1. Perception Module: Gathers situational data
2. State Estimator: Tracks current state
3. Context Modeler: Interprets situation meaning
4. Change Detector: Identifies state changes
5. Threat Assessor: Evaluates risks
6. Opportunity Detector: Identifies opportunities
7. Predictor: Anticipates future states
Awareness Dimensions:
- Spatial: Where things are
- Temporal: When things happen
- Causal: Why things occur
- Social: Who is involved
- Functional: What capabilities exist
- Intentional: What goals are active
Use Cases:
- Autonomous vehicles understanding traffic
- Security monitoring systems
- Personal assistants with context awareness
- Game AI with environmental awareness
- Crisis management systems
LangChain Implementation:
Uses memory systems, context tracking, state estimation chains, and
predictive models to maintain situational awareness.
"""
Purpose: This module implements the Situational Awareness pattern, providing a structured approach to agent design and implementation.
Import Statements
Standard Library Imports
import datetime.datetime
import enum.Enum
import langchain_core.output_parsers.JsonOutputParser
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
Purpose: Core Python functionality:
– datetime.datetime: Date and time handling
– enum.Enum: Enumeration support
– langchain_core.output_parsers.JsonOutputParser: Regular expression operations
– langchain_core.output_parsers.StrOutputParser: Regular expression operations
– langchain_core.prompts.ChatPromptTemplate: Regular expression operations
– os: Environment variable access and system operations
– typing.Any: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.List: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional)
LangChain Imports
from langchain.memory.ConversationBufferMemory
from langchain_openai.ChatOpenAI
Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows – Memory: State management across interactions
Other Imports
from dotenv.load_dotenv
Purpose: Additional dependencies for specialized functionality.
Classes
Class: SituationStatus
Lines 63-69
Inherits from: Enum
Purpose:
Enumeration of situation statuses.
Class: SituationalAwarenessAgent
Lines 72-345
Purpose:
Agent that maintains comprehensive awareness of current situation.
Methods:
__init__(self, agent_role)
Lines 77-91
Purpose:
Initialize with role and context.
Parameters:
– agent_role: Agent role
perceive_environment(self, observations)
Lines 93-152
Purpose:
Process environmental observations to update awareness.
Args:
observations: Dictionary of observed facts
Returns:
Updated situational model
Parameters:
– observations: Observations
Returns: <ast.Subscript object at 0x000001A06D7FDFD0>
assess_situation(self)
Lines 154-223
Purpose:
Assess current situation for threats and opportunities.
Returns:
Situation assessment with status and recommendations
Parameters:
Returns: <ast.Subscript object at 0x000001A06D72EE50>
predict_changes(self, time_horizon)
Lines 225-278
Purpose:
Predict likely changes to the situation.
Args:
time_horizon: Time frame for prediction
Returns:
Predictions about situation evolution
Parameters:
– time_horizon: Time horizon
Returns: <ast.Subscript object at 0x000001A06D604250>
context_appropriate_response(self, query)
Lines 280-335
Purpose:
Generate response appropriate to current situational context.
Args:
query: Query or request
Returns:
Context-aware response
Parameters:
– query: Input query or question
Returns: <ast.Subscript object at 0x000001A06D17C810>
get_situation_summary(self)
Lines 337-345
Purpose:
Get comprehensive summary of current situation.
Parameters:
Returns: <ast.Subscript object at 0x000001A06D6F5AD0>
Functions
demonstrate_situational_awareness()
Lines 348-428
Purpose:
Demonstrate situational awareness capabilities.
Key Concepts Summary
Pattern Overview
The Situational Awareness pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.
Core Components
- Agent Class: Main implementation of the pattern logic
- Prompt Engineering: Carefully crafted prompts to guide LLM behavior
- Tool Integration: Optional tools for extended functionality
- Memory Management: State handling across interactions
- Output Parsing: Structured response processing
When to Use
✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability
❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)
Best Practices Demonstrated
1. Environment Configuration
- ✅ Use
.envfiles for API keys - ✅ Never hardcode credentials
- ✅ Validate environment variables on startup
2. Error Handling
- ✅ Graceful degradation on failures
- ✅ Informative error messages
- ✅ Logging for debugging
3. Prompt Engineering
- ✅ Clear, specific instructions
- ✅ Structured output formats
- ✅ Few-shot examples when beneficial
4. Code Organization
- ✅ Separation of concerns (class-based design)
- ✅ Type hints for clarity
- ✅ Docstrings for documentation
5. Performance Optimization
- ✅ Appropriate temperature settings
- ✅ Token budget management
- ✅ Caching where applicable
LangChain Components Used
- ChatOpenAI: OpenAI model interface
- Chains: Sequential component composition
- Memory: Conversation state management
Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development
Execution Flow
1. Initialize agent with configuration
↓
2. Load environment and dependencies
↓
3. Prepare prompt with input data
↓
4. Invoke LLM through LangChain
↓
5. Parse and validate response
↓
6. Return formatted result
Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation
Performance Considerations
Computational Cost
- LLM Calls: Primary cost factor (based on tokens)
- Latency: Depends on model and input size
- Memory: Minimal for most patterns
Optimization Strategies
- Prompt Optimization: Minimize tokens while maintaining clarity
- Caching: Cache repeated queries/responses
- Model Selection: Use smaller models when appropriate
- Batch Processing: Group similar requests
- Streaming: For long responses, stream tokens
Cost Estimates (Approximate)
- GPT-3.5-turbo: ~$0.002 per request
- GPT-4: ~$0.03 per request
- GPT-4-turbo: ~$0.01 per request
Actual costs vary based on input/output token counts
Common Pitfalls to Avoid
1. Prompt Design Issues
❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts
2. Error Handling
❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation
3. Token Management
❌ Exceeding model token limits
✅ Truncate inputs, chunk processing
4. Security
❌ Exposing API keys in code
✅ Use environment variables
5. Testing
❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs
6. Cost Management
❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring
Further Improvements
Production Enhancements
- Monitoring: Add logging, metrics, tracing
- Validation: Input/output schema validation
- Rate Limiting: Implement request throttling
- Caching: Redis or in-memory cache for responses
- Async Support: Non-blocking operations for scalability
Feature Extensions
- Multi-Model Support: Switch between providers
- Fallback Models: Graceful degradation strategy
- Custom Tools: Domain-specific function calling
- Advanced Memory: Vector stores for long-term memory
- Evaluation: Automated testing and benchmarking
Code Quality
- Testing: Unit tests, integration tests
- Documentation: API docs, usage examples
- Type Safety: Stricter type hints
- Configuration: Externalized config files
Related Patterns
- Pattern 110
- Pattern 112
Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…
Summary
The Situational Awareness pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:
✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns
🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability
📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/
File: 111_situational_awareness.py
Total Lines: 433
Pattern Category: AI Agent Design Pattern #111
Generated from actual implementation code
For educational and reference purposes
Leave a Reply