"""
Pattern 164: Temporal Reasoning
Description:
The Temporal Reasoning pattern enables agents to understand and reason about time-based
relationships, sequences, durations, and temporal logic. This pattern helps agents solve
problems involving timelines, schedules, causality, and temporal constraints.
Components:
1. Timeline Manager: Maintains ordered sequences of events
2. Temporal Relation Analyzer: Identifies before/after/during/overlapping relationships
3. Duration Calculator: Computes time intervals and durations
4. Temporal Constraint Checker: Validates temporal constraints
5. Sequence Analyzer: Analyzes event sequences and patterns
6. Temporal Query Engine: Answers time-based questions
Use Cases:
- Schedule planning and optimization
- Event sequence analysis
- Causal reasoning
- Historical timeline construction
- Project timeline management
- Deadline tracking
- Temporal pattern detection
Benefits:
- Enables time-aware reasoning
- Supports scheduling and planning
- Detects temporal patterns
- Validates temporal constraints
- Better understanding of causality
Trade-offs:
- Complex time representation
- Timezone handling challenges
- Requires precise temporal data
- May need calendar logic
- Limited by data accuracy
LangChain Implementation:
Uses LLM for temporal relationship interpretation with structured datetime
management and interval calculations. Combines symbolic temporal logic with
natural language understanding for temporal queries.
"""
import os
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
load_dotenv()
class TemporalRelation(Enum):
"""Types of temporal relationships (Allen's interval algebra)"""
BEFORE = "before"
AFTER = "after"
DURING = "during"
CONTAINS = "contains"
OVERLAPS = "overlaps"
OVERLAPPED_BY = "overlapped_by"
MEETS = "meets"
MET_BY = "met_by"
STARTS = "starts"
STARTED_BY = "started_by"
FINISHES = "finishes"
FINISHED_BY = "finished_by"
EQUALS = "equals"
class TimeUnit(Enum):
"""Units of time measurement"""
SECOND = "second"
MINUTE = "minute"
HOUR = "hour"
DAY = "day"
WEEK = "week"
MONTH = "month"
YEAR = "year"
@dataclass
class TemporalEvent:
"""Represents an event with temporal properties"""
name: str
start_time: datetime
end_time: Optional[datetime] = None
description: str = ""
metadata: Dict[str, Any] = field(default_factory=dict)
@property
def duration(self) -> Optional[timedelta]:
"""Calculate duration of the event"""
if self.end_time:
return self.end_time - self.start_time
return None
@property
def is_point_event(self) -> bool:
"""Check if this is a point event (no duration)"""
return self.end_time is None or self.start_time == self.end_time
@dataclass
class Timeline:
"""Represents a timeline with multiple events"""
name: str
events: List[TemporalEvent] = field(default_factory=list)
def add_event(self, event: TemporalEvent):
"""Add an event to the timeline"""
self.events.append(event)
self.events.sort(key=lambda e: e.start_time)
def get_events_in_range(self, start: datetime, end: datetime) -> List[TemporalEvent]:
"""Get all events within a time range"""
return [e for e in self.events
if e.start_time = start)]
@dataclass
class TemporalQuery:
"""A temporal reasoning query"""
question: str
events: List[TemporalEvent]
context: str = ""
@dataclass
class TemporalAnalysis:
"""Result of temporal reasoning"""
answer: str
relationships: List[str]
timeline: str
sequences: List[List[str]]
reasoning: str
class TemporalReasoningAgent:
"""Agent that performs temporal reasoning tasks"""
def __init__(self, model_name: str = "gpt-3.5-turbo"):
"""Initialize the temporal reasoning agent"""
self.llm = ChatOpenAI(model=model_name, temperature=0)
# Prompts for temporal reasoning
self.temporal_prompt = ChatPromptTemplate.from_messages([
("system", """You are a temporal reasoning expert. Analyze events, timelines,
and temporal relationships. Consider durations, sequences, and causal relationships.
Be precise about time-based logic."""),
("user", """Events and their times:
{event_descriptions}
Question: {question}
Provide a clear answer based on temporal analysis.""")
])
self.sequence_prompt = ChatPromptTemplate.from_messages([
("system", """You are an expert at analyzing event sequences and patterns.
Identify patterns, detect anomalies, and reason about causality."""),
("user", """Event sequence:
{sequence}
Analysis task: {task}
Describe patterns, relationships, and insights.""")
])
self.temporal_chain = self.temporal_prompt | self.llm | StrOutputParser()
self.sequence_chain = self.sequence_prompt | self.llm | StrOutputParser()
def analyze_temporal_query(self, query: TemporalQuery) -> TemporalAnalysis:
"""Analyze a temporal query and return comprehensive analysis"""
# Identify temporal relationships
relationships = self._identify_temporal_relationships(query.events)
# Create timeline visualization
timeline_viz = self._create_timeline_visualization(query.events)
# Identify sequences
sequences = self._identify_sequences(query.events)
# Create event descriptions
event_descriptions = self._create_event_descriptions(query.events)
# Get LLM reasoning
reasoning = self.temporal_chain.invoke({
"event_descriptions": event_descriptions,
"question": query.question
})
return TemporalAnalysis(
answer=reasoning,
relationships=relationships,
timeline=timeline_viz,
sequences=sequences,
reasoning=reasoning
)
def check_temporal_relation(self, event1: TemporalEvent,
event2: TemporalEvent,
relation: TemporalRelation) -> bool:
"""Check if a specific temporal relationship holds between two events"""
e1_start = event1.start_time
e1_end = event1.end_time or event1.start_time
e2_start = event2.start_time
e2_end = event2.end_time or event2.start_time
if relation == TemporalRelation.BEFORE:
return e1_end e2_end
elif relation == TemporalRelation.DURING:
return e1_start > e2_start and e1_end < e2_end
elif relation == TemporalRelation.CONTAINS:
return e1_start e2_end
elif relation == TemporalRelation.OVERLAPS:
return e1_start e2_start and e1_end < e2_end
elif relation == TemporalRelation.OVERLAPPED_BY:
return e2_start e1_start and e2_end < e1_end
elif relation == TemporalRelation.MEETS:
return e1_end == e2_start
elif relation == TemporalRelation.MET_BY:
return e2_end == e1_start
elif relation == TemporalRelation.STARTS:
return e1_start == e2_start and e1_end e2_end
elif relation == TemporalRelation.FINISHES:
return e1_start > e2_start and e1_end == e2_end
elif relation == TemporalRelation.FINISHED_BY:
return e1_start List[Tuple[str, str]]:
"""Find all pairs of overlapping events"""
overlaps = []
for i, event1 in enumerate(events):
for event2 in events[i+1:]:
if self._events_overlap(event1, event2):
overlaps.append((event1.name, event2.name))
return overlaps
def find_gaps(self, events: List[TemporalEvent]) -> List[Tuple[datetime, datetime]]:
"""Find time gaps between consecutive events"""
if not events:
return []
sorted_events = sorted(events, key=lambda e: e.start_time)
gaps = []
for i in range(len(sorted_events) - 1):
current_end = sorted_events[i].end_time or sorted_events[i].start_time
next_start = sorted_events[i + 1].start_time
if current_end timedelta:
"""Calculate total duration of events"""
if not events:
return timedelta(0)
if not merge_overlaps:
# Simple sum of all durations
total = timedelta(0)
for event in events:
if event.duration:
total += event.duration
return total
else:
# Merge overlapping intervals first
intervals = []
for event in events:
start = event.start_time
end = event.end_time or event.start_time
intervals.append((start, end))
# Sort and merge
intervals.sort()
merged = []
current_start, current_end = intervals[0]
for start, end in intervals[1:]:
if start Dict[str, Any]:
"""Detect patterns in temporal data"""
if not events:
return {"patterns": []}
sorted_events = sorted(events, key=lambda e: e.start_time)
patterns = {
"total_events": len(events),
"time_span": {
"start": sorted_events[0].start_time,
"end": sorted_events[-1].end_time or sorted_events[-1].start_time
},
"regular_intervals": self._detect_regular_intervals(sorted_events),
"clusters": self._detect_time_clusters(sorted_events),
"gaps": len(self.find_gaps(events)),
"overlaps": len(self.find_overlapping_events(events))
}
return patterns
def _events_overlap(self, event1: TemporalEvent, event2: TemporalEvent) -> bool:
"""Check if two events overlap in time"""
e1_start = event1.start_time
e1_end = event1.end_time or event1.start_time
e2_start = event2.start_time
e2_end = event2.end_time or event2.start_time
return not (e1_end < e2_start or e2_end List[str]:
"""Identify temporal relationships between events"""
relationships = []
for i, event1 in enumerate(events):
for event2 in events[i+1:]:
# Check key relationships
if self.check_temporal_relation(event1, event2, TemporalRelation.BEFORE):
relationships.append(f"{event1.name} before {event2.name}")
elif self.check_temporal_relation(event1, event2, TemporalRelation.DURING):
relationships.append(f"{event1.name} during {event2.name}")
elif self.check_temporal_relation(event1, event2, TemporalRelation.OVERLAPS):
relationships.append(f"{event1.name} overlaps {event2.name}")
elif self.check_temporal_relation(event1, event2, TemporalRelation.MEETS):
relationships.append(f"{event1.name} meets {event2.name}")
return relationships
def _identify_sequences(self, events: List[TemporalEvent]) -> List[List[str]]:
"""Identify sequences of consecutive events"""
if not events:
return []
sorted_events = sorted(events, key=lambda e: e.start_time)
sequences = []
current_sequence = [sorted_events[0].name]
for i in range(1, len(sorted_events)):
prev_end = sorted_events[i-1].end_time or sorted_events[i-1].start_time
curr_start = sorted_events[i].start_time
# Check if events are consecutive (within 1 hour)
if (curr_start - prev_end) 1:
sequences.append(current_sequence)
current_sequence = [sorted_events[i].name]
if len(current_sequence) > 1:
sequences.append(current_sequence)
return sequences
def _detect_regular_intervals(self, sorted_events: List[TemporalEvent]) -> Optional[timedelta]:
"""Detect if events occur at regular intervals"""
if len(sorted_events) < 3:
return None
intervals = []
for i in range(1, len(sorted_events)):
interval = sorted_events[i].start_time - sorted_events[i-1].start_time
intervals.append(interval)
# Check if intervals are similar (within 10% variance)
avg_interval = sum(intervals, timedelta(0)) / len(intervals)
variance = sum((i - avg_interval).total_seconds()**2 for i in intervals) / len(intervals)
if variance int:
"""Detect clusters of events happening close together"""
if len(sorted_events) timedelta(hours=2): # New cluster if gap > 2 hours
clusters += 1
return clusters
def _create_event_descriptions(self, events: List[TemporalEvent]) -> str:
"""Create textual descriptions of events"""
descriptions = []
for event in sorted(events, key=lambda e: e.start_time):
desc = f"{event.name}: {event.start_time.strftime('%Y-%m-%d %H:%M')}"
if event.end_time:
desc += f" to {event.end_time.strftime('%Y-%m-%d %H:%M')}"
desc += f" (duration: {event.duration})"
if event.description:
desc += f" - {event.description}"
descriptions.append(desc)
return "\n".join(descriptions)
def _create_timeline_visualization(self, events: List[TemporalEvent]) -> str:
"""Create ASCII timeline visualization"""
if not events:
return "No events to visualize"
sorted_events = sorted(events, key=lambda e: e.start_time)
lines = ["Timeline:", "=" * 60]
for event in sorted_events:
time_str = event.start_time.strftime('%Y-%m-%d %H:%M')
if event.end_time:
duration_str = f" [{event.duration}]"
else:
duration_str = " [instant]"
lines.append(f" {time_str} | {event.name}{duration_str}")
return "\n".join(lines)
def demonstrate_temporal_reasoning():
"""Demonstrate temporal reasoning capabilities"""
print("=" * 80)
print("TEMPORAL REASONING PATTERN DEMONSTRATION")
print("=" * 80)
agent = TemporalReasoningAgent()
# Example 1: Project timeline analysis
print("\n" + "=" * 80)
print("Example 1: Project Timeline Analysis")
print("=" * 80)
base_time = datetime(2024, 1, 1, 9, 0)
project_events = [
TemporalEvent("Planning", base_time, base_time + timedelta(days=7), "Initial planning phase"),
TemporalEvent("Design", base_time + timedelta(days=7), base_time + timedelta(days=14), "Design phase"),
TemporalEvent("Development", base_time + timedelta(days=14), base_time + timedelta(days=45), "Development phase"),
TemporalEvent("Testing", base_time + timedelta(days=40), base_time + timedelta(days=52), "Testing phase overlaps with dev"),
TemporalEvent("Deployment", base_time + timedelta(days=52), base_time + timedelta(days=53), "Final deployment")
]
query = TemporalQuery(
question="What is the timeline of the project and are there any overlapping phases?",
events=project_events,
context="Software project timeline"
)
analysis = agent.analyze_temporal_query(query)
print(f"\nQuery: {query.question}")
print(f"\n{analysis.timeline}")
print(f"\nTemporal Relationships:")
for rel in analysis.relationships[:5]:
print(f" - {rel}")
print(f"\nAnalysis:\n{analysis.answer}")
# Example 2: Overlapping events detection
print("\n" + "=" * 80)
print("Example 2: Overlapping Events Detection")
print("=" * 80)
meeting_events = [
TemporalEvent("Team Meeting", datetime(2024, 1, 15, 10, 0), datetime(2024, 1, 15, 11, 0)),
TemporalEvent("Client Call", datetime(2024, 1, 15, 10, 30), datetime(2024, 1, 15, 11, 30)),
TemporalEvent("Lunch Break", datetime(2024, 1, 15, 12, 0), datetime(2024, 1, 15, 13, 0)),
TemporalEvent("Code Review", datetime(2024, 1, 15, 14, 0), datetime(2024, 1, 15, 15, 0))
]
overlaps = agent.find_overlapping_events(meeting_events)
print("\nSchedule for January 15, 2024:")
for event in sorted(meeting_events, key=lambda e: e.start_time):
print(f" {event.start_time.strftime('%H:%M')}-{event.end_time.strftime('%H:%M')}: {event.name}")
if overlaps:
print(f"\nConflicts detected:")
for event1, event2 in overlaps:
print(f" ⚠ {event1} overlaps with {event2}")
else:
print("\n✓ No scheduling conflicts detected")
# Example 3: Temporal relationship checking
print("\n" + "=" * 80)
print("Example 3: Temporal Relationship Checking")
print("=" * 80)
event_a = TemporalEvent("Event A", datetime(2024, 1, 1, 10, 0), datetime(2024, 1, 1, 12, 0))
event_b = TemporalEvent("Event B", datetime(2024, 1, 1, 11, 0), datetime(2024, 1, 1, 11, 30))
event_c = TemporalEvent("Event C", datetime(2024, 1, 1, 12, 0), datetime(2024, 1, 1, 13, 0))
relations_to_check = [
(event_a, event_b, TemporalRelation.CONTAINS),
(event_b, event_a, TemporalRelation.DURING),
(event_a, event_c, TemporalRelation.MEETS),
(event_c, event_a, TemporalRelation.MET_BY),
(event_a, event_b, TemporalRelation.OVERLAPS)
]
print("\nChecking temporal relationships:")
for evt1, evt2, relation in relations_to_check:
holds = agent.check_temporal_relation(evt1, evt2, relation)
symbol = "✓" if holds else "✗"
print(f" {symbol} {evt1.name} {relation.value} {evt2.name}: {holds}")
# Example 4: Duration calculations
print("\n" + "=" * 80)
print("Example 4: Duration Calculations")
print("=" * 80)
work_events = [
TemporalEvent("Task 1", datetime(2024, 1, 1, 9, 0), datetime(2024, 1, 1, 11, 0)),
TemporalEvent("Task 2", datetime(2024, 1, 1, 10, 30), datetime(2024, 1, 1, 12, 0)),
TemporalEvent("Task 3", datetime(2024, 1, 1, 13, 0), datetime(2024, 1, 1, 15, 0)),
]
total_simple = agent.calculate_total_duration(work_events, merge_overlaps=False)
total_merged = agent.calculate_total_duration(work_events, merge_overlaps=True)
print("\nWork events:")
for event in work_events:
print(f" {event.name}: {event.duration}")
print(f"\nTotal duration (simple sum): {total_simple}")
print(f"Total duration (merged overlaps): {total_merged}")
print(f"Overlap time: {total_simple - total_merged}")
# Example 5: Gap detection
print("\n" + "=" * 80)
print("Example 5: Gap Detection")
print("=" * 80)
sparse_events = [
TemporalEvent("Morning Session", datetime(2024, 1, 1, 9, 0), datetime(2024, 1, 1, 10, 0)),
TemporalEvent("Afternoon Session", datetime(2024, 1, 1, 14, 0), datetime(2024, 1, 1, 15, 0)),
TemporalEvent("Evening Session", datetime(2024, 1, 1, 19, 0), datetime(2024, 1, 1, 20, 0))
]
gaps = agent.find_gaps(sparse_events)
print("\nEvents:")
for event in sparse_events:
print(f" {event.start_time.strftime('%H:%M')}-{event.end_time.strftime('%H:%M')}: {event.name}")
print(f"\nGaps detected: {len(gaps)}")
for start, end in gaps:
duration = end - start
print(f" Gap from {start.strftime('%H:%M')} to {end.strftime('%H:%M')} ({duration})")
# Example 6: Pattern detection
print("\n" + "=" * 80)
print("Example 6: Pattern Detection")
print("=" * 80)
regular_events = [
TemporalEvent(f"Daily Standup {i}", datetime(2024, 1, i, 9, 0), datetime(2024, 1, i, 9, 15))
for i in range(1, 6)
]
patterns = agent.detect_temporal_patterns(regular_events)
print("\nEvents:")
for event in regular_events:
print(f" {event.start_time.strftime('%Y-%m-%d %H:%M')}: {event.name}")
print(f"\nPattern Analysis:")
print(f" Total events: {patterns['total_events']}")
print(f" Time span: {patterns['time_span']['start'].strftime('%Y-%m-%d')} to {patterns['time_span']['end'].strftime('%Y-%m-%d')}")
if patterns['regular_intervals']:
print(f" Regular interval detected: {patterns['regular_intervals']}")
print(f" Number of clusters: {patterns['clusters']}")
print(f" Gaps: {patterns['gaps']}")
print(f" Overlaps: {patterns['overlaps']}")
# Summary
print("\n" + "=" * 80)
print("SUMMARY")
print("=" * 80)
print("""
The Temporal Reasoning pattern enables agents to:
✓ Understand temporal relationships (before, after, during, overlaps)
✓ Detect scheduling conflicts and overlaps
✓ Calculate durations and time gaps
✓ Identify temporal patterns and sequences
✓ Validate temporal constraints
✓ Reason about causality and event sequences
✓ Analyze timelines and schedules
This pattern is valuable for:
- Project and task scheduling
- Calendar and meeting management
- Event sequence analysis
- Timeline construction and validation
- Causal reasoning
- Temporal pattern detection
- Deadline and milestone tracking
""")
if __name__ == "__main__":
demonstrate_temporal_reasoning()
Agentic Design Pattern: Temporal Reasoning
Pattern 164: Temporal Reasoning – Line-by-Line Explanation
Overview
This document provides a detailed line-by-line explanation of the Temporal Reasoning pattern implementation.
Pattern Description
Pattern 164: Temporal Reasoning
Description: The Temporal Reasoning pattern enables agents to understand and reason about time-based relationships, sequences, durations, and temporal logic. This pattern helps agents solve problems involving timelines, schedules, causality, and temporal constraints.
Components: 1. Timeline Manager: Maintains ordered sequences of events 2. Temporal Relation Analyzer: Identifies before/after/during/overlapping relationships 3. Duration Calculator: Computes time intervals and durations 4. Temporal Constraint Checker: Validates temporal constraints 5. Sequence Analyzer: Analyzes event sequences and patterns 6. Temporal Query Engine: Answers time-based questions
Use Cases: – Schedule planning and optimization – Event sequence analysis – Causal reasoning – Historical timeline construction – Project timeline management – Deadline tracking – Temporal pattern detection
Benefits: – Enables time-aware reasoning – Supports scheduling and planning – Detects temporal patterns – Validates temporal constraints – Better understanding of causality
Trade-offs: – Complex time representation – Timezone handling challenges – Requires precise temporal data – May need calendar logic – Limited by data accuracy
LangChain Implementation: Uses LLM for temporal relationship interpretation with structured datetime management and interval calculations. Combines symbolic temporal logic with natural language understanding for temporal queries.
File Statistics
- Total Lines: 637
- Classes: 7
- Functions: 1
- Constants: 0
Module Documentation
"""
Pattern 164: Temporal Reasoning
Description:
The Temporal Reasoning pattern enables agents to understand and reason about time-based
relationships, sequences, durations, and temporal logic. This pattern helps agents solve
problems involving timelines, schedules, causality, and temporal constraints.
Components:
1. Timeline Manager: Maintains ordered sequences of events
2. Temporal Relation Analyzer: Identifies before/after/during/overlapping relationships
3. Duration Calculator: Computes time intervals and durations
4. Temporal Constraint Checker: Validates temporal constraints
5. Sequence Analyzer: Analyzes event sequences and patterns
6. Temporal Query Engine: Answers time-based questions
Use Cases:
- Schedule planning and optimization
- Event sequence analysis
- Causal reasoning
- Historical timeline construction
- Project timeline management
- Deadline tracking
- Temporal pattern detection
Benefits:
- Enables time-aware reasoning
- Supports scheduling and planning
- Detects temporal patterns
- Validates temporal constraints
- Better understanding of causality
Trade-offs:
- Complex time representation
- Timezone handling challenges
- Requires precise temporal data
- May need calendar logic
- Limited by data accuracy
LangChain Implementation:
Uses LLM for temporal relationship interpretation with structured datetime
management and interval calculations. Combines symbolic temporal logic with
natural language understanding for temporal queries.
"""
Purpose: This module implements the Temporal Reasoning pattern, providing a structured approach to agent design and implementation.
Import Statements
Standard Library Imports
import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import datetime.timedelta
import enum.Enum
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
import typing.Tuple
Purpose: Core Python functionality:
– dataclasses.dataclass: Simplified class definitions with @dataclass
– dataclasses.field: Simplified class definitions with @dataclass
– datetime.datetime: Date and time handling
– datetime.timedelta: Date and time handling
– enum.Enum: Enumeration support
– 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)
– typing.Tuple: Type hints for better code documentation (List, Dict, Any, Optional)
LangChain Imports
from langchain_openai.ChatOpenAI
Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows
Other Imports
from dotenv.load_dotenv
Purpose: Additional dependencies for specialized functionality.
Classes
Class: TemporalRelation
Lines 60-74
Inherits from: Enum
Purpose:
Types of temporal relationships (Allen's interval algebra)
Class: TimeUnit
Lines 77-85
Inherits from: Enum
Purpose:
Units of time measurement
Class: TemporalEvent
Lines 89-107
Purpose:
Represents an event with temporal properties
Methods:
duration(self)
Lines 98-102
Decorators: property
Purpose:
Calculate duration of the event
Parameters:
Returns: <ast.Subscript object at 0x000001A06D7B3590>
is_point_event(self)
Lines 105-107
Decorators: property
Purpose:
Check if this is a point event (no duration)
Parameters:
Returns: bool
Class: Timeline
Lines 111-124
Purpose:
Represents a timeline with multiple events
Methods:
add_event(self, event)
Lines 116-119
Purpose:
Add an event to the timeline
Parameters:
– event: Event
get_events_in_range(self, start, end)
Lines 121-124
Purpose:
Get all events within a time range
Parameters:
– start: Start
– end: End
Returns: <ast.Subscript object at 0x000001A06D826710>
Class: TemporalQuery
Lines 128-132
Purpose:
A temporal reasoning query
Class: TemporalAnalysis
Lines 136-142
Purpose:
Result of temporal reasoning
Class: TemporalReasoningAgent
Lines 145-452
Purpose:
Agent that performs temporal reasoning tasks
Methods:
__init__(self, model_name)
Lines 148-177
Purpose:
Initialize the temporal reasoning agent
Parameters:
– model_name: Language model instance
analyze_temporal_query(self, query)
Lines 179-205
Purpose:
Analyze a temporal query and return comprehensive analysis
Parameters:
– query: Input query or question
Returns: TemporalAnalysis
check_temporal_relation(self, event1, event2, relation)
Lines 207-243
Purpose:
Check if a specific temporal relationship holds between two events
Parameters:
– event1: Event1
– event2: Event2
– relation: Relation
Returns: bool
find_overlapping_events(self, events)
Lines 245-254
Purpose:
Find all pairs of overlapping events
Parameters:
– events: Events
Returns: <ast.Subscript object at 0x000001A06D759190>
find_gaps(self, events)
Lines 256-271
Purpose:
Find time gaps between consecutive events
Parameters:
– events: Events
Returns: <ast.Subscript object at 0x000001A06D7A0E10>
calculate_total_duration(self, events, merge_overlaps)
Lines 273-313
Purpose:
Calculate total duration of events
Parameters:
– events: Events
– merge_overlaps: Merge overlaps
Returns: timedelta
detect_temporal_patterns(self, events)
Lines 315-334
Purpose:
Detect patterns in temporal data
Parameters:
– events: Events
Returns: <ast.Subscript object at 0x000001A06D624BD0>
_events_overlap(self, event1, event2)
Lines 336-343
Purpose:
Check if two events overlap in time
Parameters:
– event1: Event1
– event2: Event2
Returns: bool
_identify_temporal_relationships(self, events)
Lines 345-361
Purpose:
Identify temporal relationships between events
Parameters:
– events: Events
Returns: <ast.Subscript object at 0x000001A06D867290>
_identify_sequences(self, events)
Lines 363-387
Purpose:
Identify sequences of consecutive events
Parameters:
– events: Events
Returns: <ast.Subscript object at 0x000001A06D77F010>
_detect_regular_intervals(self, sorted_events)
Lines 389-406
Purpose:
Detect if events occur at regular intervals
Parameters:
– sorted_events: Sorted events
Returns: <ast.Subscript object at 0x000001A06D72AD90>
_detect_time_clusters(self, sorted_events)
Lines 408-419
Purpose:
Detect clusters of events happening close together
Parameters:
– sorted_events: Sorted events
Returns: int
_create_event_descriptions(self, events)
Lines 421-432
Purpose:
Create textual descriptions of events
Parameters:
– events: Events
Returns: str
_create_timeline_visualization(self, events)
Lines 434-452
Purpose:
Create ASCII timeline visualization
Parameters:
– events: Events
Returns: str
Functions
demonstrate_temporal_reasoning()
Lines 455-632
Purpose:
Demonstrate temporal reasoning capabilities
Key Concepts Summary
Pattern Overview
The Temporal Reasoning 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
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 163
- Pattern 165
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 Temporal Reasoning 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: 164_temporal_reasoning.py
Total Lines: 637
Pattern Category: AI Agent Design Pattern #164
Generated from actual implementation code
For educational and reference purposes