"""
Pattern 207: Temporal Reasoning Pattern
Reasons about time, temporal sequences, durations, and temporal constraints.
Handles before/after relationships, concurrency, and time-based planning.
"""
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
class TemporalRelation(Enum):
"""Allen's interval algebra relations."""
BEFORE = "before"
AFTER = "after"
MEETS = "meets"
OVERLAPS = "overlaps"
DURING = "during"
STARTS = "starts"
FINISHES = "finishes"
EQUALS = "equals"
@dataclass
class TemporalEvent:
"""Event with temporal information."""
name: str
start_time: Optional[datetime] = None
end_time: Optional[datetime] = None
duration: Optional[timedelta] = None
description: str = ""
@dataclass
class TemporalConstraint:
"""Constraint between temporal events."""
event1: str
event2: str
relation: TemporalRelation
description: str = ""
class TemporalReasoningAgent:
"""Agent that performs temporal reasoning."""
def __init__(self, name: str, model_name: str = "gpt-4"):
self.name = name
self.llm = ChatOpenAI(model=model_name, temperature=0.3)
self.events: List[TemporalEvent] = []
self.constraints: List[TemporalConstraint] = []
def add_event(self, event: TemporalEvent):
"""Add temporal event."""
self.events.append(event)
def infer_temporal_order(self, events_description: str) -> List[str]:
"""Infer temporal ordering of events from description."""
print(f"\n {self.name}: Inferring temporal order...")
prompt = ChatPromptTemplate.from_messages([
("system", "Analyze text and determine temporal order of events."),
("human", """Text: {text}
List the events in chronological order (earliest to latest).
Format: one event per line.""")
])
chain = prompt | self.llm
response = chain.invoke({"text": events_description})
ordered_events = [line.strip().lstrip("0123456789.-) ")
for line in response.content.split("\n")
if line.strip()]
print(f" Ordered {len(ordered_events)} events")
return ordered_events
def calculate_duration(self, event_description: str) -> timedelta:
"""Estimate duration of an event."""
prompt = ChatPromptTemplate.from_messages([
("system", "Estimate typical duration in minutes for the event."),
("human", "Event: {event}\n\nTypical duration in minutes (just the number):")
])
chain = prompt | self.llm
response = chain.invoke({"event": event_description})
try:
minutes = int(response.content.strip().split()[0])
return timedelta(minutes=minutes)
except:
return timedelta(hours=1) # Default
def check_temporal_consistency(self) -> Dict[str, Any]:
"""Check if temporal constraints are consistent."""
print(f"\n {self.name}: Checking temporal consistency...")
if len(self.events) List[TemporalEvent]:
"""Plan schedule for tasks."""
print(f"\n {self.name}: Planning schedule for {len(tasks)} tasks...")
schedule = []
current_time = start_time
for task in tasks:
duration = self.calculate_duration(task)
event = TemporalEvent(
name=task,
start_time=current_time,
end_time=current_time + duration,
duration=duration
)
schedule.append(event)
self.events.append(event)
current_time = event.end_time
print(f" {task}: {event.start_time.strftime('%H:%M')} - "
f"{event.end_time.strftime('%H:%M')} ({duration})")
return schedule
def reason_about_deadline(self, task: str, deadline: datetime,
current_time: datetime) -> Dict[str, Any]:
"""Reason about meeting a deadline."""
time_remaining = deadline - current_time
duration = self.calculate_duration(task)
can_meet = time_remaining >= duration
buffer = time_remaining - duration if can_meet else timedelta(0)
return {
"task": task,
"deadline": deadline,
"estimated_duration": duration,
"time_remaining": time_remaining,
"can_meet_deadline": can_meet,
"buffer_time": buffer
}
class TemporalReasoningPattern:
"""
Temporal Reasoning Pattern - Reason about time.
Handles temporal ordering, durations, constraints, scheduling,
and deadline reasoning.
"""
def __init__(self):
self.agents: Dict[str, TemporalReasoningAgent] = {}
def register_agent(self, agent: TemporalReasoningAgent):
"""Register temporal reasoning agent."""
self.agents[agent.name] = agent
def run(self, scenarios: List[Dict[str, Any]]) -> Dict[str, Any]:
"""Execute temporal reasoning scenarios."""
print("=== Temporal Reasoning Pattern ===\n")
for agent in self.agents.values():
for scenario in scenarios:
scenario_type = scenario.get("type", "order")
if scenario_type == "order":
agent.infer_temporal_order(scenario["text"])
elif scenario_type == "schedule":
agent.plan_schedule(
scenario["tasks"],
scenario.get("start_time", datetime.now())
)
elif scenario_type == "deadline":
result = agent.reason_about_deadline(
scenario["task"],
scenario["deadline"],
scenario.get("current_time", datetime.now())
)
print(f"\n Deadline analysis:")
print(f" Can meet: {result['can_meet_deadline']}")
print(f" Buffer: {result['buffer_time']}")
return {
"pattern": "TemporalReasoning",
"agents": len(self.agents),
"scenarios_processed": len(scenarios),
"events_tracked": sum(len(a.events) for a in self.agents.values())
}
# Example usage
if __name__ == "__main__":
print("=== Temporal Reasoning Pattern Demo ===\n")
# Create pattern and agent
pattern = TemporalReasoningPattern()
agent = TemporalReasoningAgent("TimeKeeper")
pattern.register_agent(agent)
# Temporal reasoning scenarios
print("1. Temporal ordering scenario...")
scenarios = [
{
"type": "order",
"text": "First, I woke up. Then I had breakfast. After that, I went to work. "
"During lunch, I met a friend. Finally, I came home in the evening."
},
{
"type": "schedule",
"tasks": ["Morning meeting", "Code review", "Lunch break", "Development work"],
"start_time": datetime.now().replace(hour=9, minute=0, second=0, microsecond=0)
},
{
"type": "deadline",
"task": "Write project report",
"deadline": datetime.now() + timedelta(hours=4),
"current_time": datetime.now()
}
]
# Run temporal reasoning
summary = pattern.run(scenarios)
print(f"\n=== Summary ===")
print(f"Scenarios processed: {summary['scenarios_processed']}")
print(f"Events tracked: {summary['events_tracked']}")
print("\n✓ Temporal reasoning: understanding time and sequences!")
Agentic Architecture Pattern: Temporal Reasoning
Pattern 207: 207 Temporal Reasoning Pattern – Line-by-Line Explanation
Overview
Pattern 207: Temporal Reasoning Pattern
Reasons about time, temporal sequences, durations, and temporal constraints. Handles before/after relationships, concurrency, and time-based planning.
Code Structure Analysis
File Information
- Pattern Number: 207
- Pattern Name: 207 Temporal Reasoning Pattern
- File Path:
patterns\advanced_reasoning\207_temporal_reasoning_pattern.py - Total Lines: 258
Detailed Code Explanation
Module Docstring
"""
Pattern 207: Temporal Reasoning Pattern
Reasons about time, temporal sequences, durations, and temporal constraints.
Handles before/after relationships, concurrency, and time-based planning.
"""
Explanation: This module docstring describes the purpose and key features of the 207 Temporal Reasoning Pattern pattern.
Imports Section
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI # First 20 imports shown
Explanation: The module imports necessary dependencies:
– from typing import Dict, Any, List, Optional: Type hints for better code documentation
– from dataclasses import dataclass, field: Data class decorators for clean data structures
– from datetime import datetime, timedelta: Date and time utilities
– from enum import Enum: Enumeration support for fixed sets of values
– from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration
– from langchain_openai import ChatOpenAI: LangChain framework integration
Enumerations
TemporalRelation (Line 16)
class TemporalRelation(Enum):
BEFORE
AFTER
MEETS
OVERLAPS
DURING
STARTS
FINISHES
EQUALS
Purpose: Allen’s interval algebra relations.
Values:
– BEFORE: Represents before
– AFTER: Represents after
– MEETS: Represents meets
– OVERLAPS: Represents overlaps
– DURING: Represents during
– STARTS: Represents starts
– FINISHES: Represents finishes
– EQUALS: Represents equals
Classes
TemporalRelation (Line 16)
Description: Allen’s interval algebra relations.
TemporalEvent (Line 29)
Description: Event with temporal information.
TemporalConstraint (Line 39)
Description: Constraint between temporal events.
TemporalReasoningAgent (Line 47)
Description: Agent that performs temporal reasoning.
Methods:
__init__ (Line 50)
def __init__(self, name, model_name):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
– name: Identifier or label
– model_name: LLM model to use (e.g., “gpt-4”)
add_event (Line 56)
def add_event(self, event):
"""Add temporal event."""
Purpose: Add temporal event.
Parameters:
– self: Instance reference
– event: Parameter for event
infer_temporal_order (Line 60)
def infer_temporal_order(self, events_description):
"""Infer temporal ordering of events from description."""
Purpose: Infer temporal ordering of events from description.
Parameters:
– self: Instance reference
– events_description: Parameter for events_description
calculate_duration (Line 82)
def calculate_duration(self, event_description):
"""Estimate duration of an event."""
Purpose: Estimate duration of an event.
Parameters:
– self: Instance reference
– event_description: Parameter for event_description
check_temporal_consistency (Line 98)
def check_temporal_consistency(self):
"""Check if temporal constraints are consistent."""
Purpose: Check if temporal constraints are consistent.
Parameters:
– self: Instance reference
plan_schedule (Line 128)
def plan_schedule(self, tasks, start_time):
"""Plan schedule for tasks."""
Purpose: Plan schedule for tasks.
Parameters:
– self: Instance reference
– tasks: Parameter for tasks
– start_time: Parameter for start_time
reason_about_deadline (Line 153)
def reason_about_deadline(self, task, deadline, current_time):
"""Reason about meeting a deadline."""
Purpose: Reason about meeting a deadline.
Parameters:
– self: Instance reference
– task: Parameter for task
– deadline: Parameter for deadline
– current_time: Parameter for current_time
TemporalReasoningPattern (Line 172)
Description: Temporal Reasoning Pattern – Reason about time.
Handles temporal ordering, durations, constraints, scheduling, and deadline reasoning.
Methods:
__init__ (Line 180)
def __init__(self):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
register_agent (Line 183)
def register_agent(self, agent):
"""Register temporal reasoning agent."""
Purpose: Register temporal reasoning agent.
Parameters:
– self: Instance reference
– agent: Parameter for agent
run (Line 187)
def run(self, scenarios):
"""Execute temporal reasoning scenarios."""
Purpose: Execute temporal reasoning scenarios.
Parameters:
– self: Instance reference
– scenarios: Parameter for scenarios
Design Patterns Used
- Strategy Pattern: Different execution strategies for various scenarios
- State Pattern: Managing different states in the workflow
- Chain of Responsibility: Sequential processing through workflow nodes
- Observer Pattern: Tracking and monitoring pattern execution
- Factory Pattern: Creating instances with proper configuration
Usage Example
# Initialize the pattern
pattern = TemporalRelation(name="example", model_name="gpt-4")
# Configure and execute
result = pattern.execute(data={"key": "value"})
# Check results
print(result)
Key Concepts
- LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
- Workflow Orchestration: LangGraph manages execution flow through nodes
- State Management: TypedDict defines workflow state structure
- Error Handling: Comprehensive error handling and fallback mechanisms
- Extensibility: Designed for easy customization and extension
Extension Points
This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities