"""
Pattern 094: Trace/Lineage Tracking
Description:
Trace and Lineage Tracking captures the complete execution path of an agent, including
all decisions, actions, intermediate states, and data transformations. This pattern is
essential for debugging complex agent behaviors, understanding how decisions were made,
ensuring compliance, and optimizing performance. Comprehensive tracing enables
reproducibility, auditing, and continuous improvement of AI systems.
Trace tracking provides:
- Complete execution history
- Decision point visibility
- Data lineage and provenance
- Performance metrics per step
- Error context and stack traces
- Reproducibility information
Components:
1. Trace Capture
- Execution events (start, end, error)
- Decision points and branches
- Input/output at each step
- Timestamps and durations
- Resource consumption
- Agent state snapshots
2. Trace Storage
- Structured trace format
- Hierarchical relationships
- Searchable metadata
- Efficient retrieval
- Long-term retention
3. Trace Analysis
- Path visualization
- Performance analysis
- Bottleneck identification
- Error pattern detection
- Comparison tools
4. Trace Replay
- Reproduce executions
- Debug specific paths
- Test alternative branches
- Validate fixes
- Training data generation
Use Cases:
1. Debugging
- Understand unexpected behavior
- Reproduce production issues
- Identify error causes
- Validate fixes
- Root cause analysis
2. Performance Optimization
- Identify bottlenecks
- Measure step durations
- Analyze resource usage
- Compare implementations
- Optimize hot paths
3. Compliance & Auditing
- Document decision making
- Prove regulatory compliance
- Audit trails for sensitive operations
- Accountability tracking
- Legal requirements
4. Quality Assurance
- Validate agent behavior
- Ensure determinism
- Track test coverage
- Regression detection
- Change impact analysis
5. Agent Development
- Understand agent flow
- Optimize prompts
- Refine decision logic
- Test edge cases
- Documentation generation
LangChain Implementation:
LangChain supports tracing through:
- LangSmith for comprehensive tracing
- Custom callback handlers
- OpenTelemetry integration
- Structured logging
- Trace context propagation
Key Features:
1. Comprehensive Capture
- Every agent step traced
- Full context preservation
- Hierarchical relationships
- Performance metrics
- Error details
2. Structured Format
- Consistent trace structure
- Searchable attributes
- Machine-readable format
- Human-readable display
- Standard schemas
3. Powerful Analysis
- Path visualization
- Query capabilities
- Statistical analysis
- Comparison tools
- Pattern detection
4. Production Ready
- Low overhead ( 1 year
- Compression for old traces
- Automatic cleanup
3. Performance Impact
- Target: Dict[str, Any]:
"""Convert to dictionary"""
data = asdict(self)
data['event_type'] = self.event_type.value
data['timestamp'] = self.timestamp.isoformat()
return data
@dataclass
class Trace:
"""Complete trace of an agent execution"""
trace_id: str
start_time: datetime
end_time: Optional[datetime] = None
events: List[TraceEvent] = field(default_factory=list)
status: str = "running" # running, completed, failed
metadata: Dict[str, Any] = field(default_factory=dict)
@property
def duration_ms(self) -> float:
"""Calculate total duration"""
if self.end_time:
return (self.end_time - self.start_time).total_seconds() * 1000
return 0.0
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary"""
return {
'trace_id': self.trace_id,
'start_time': self.start_time.isoformat(),
'end_time': self.end_time.isoformat() if self.end_time else None,
'duration_ms': self.duration_ms,
'status': self.status,
'event_count': len(self.events),
'metadata': self.metadata,
'events': [e.to_dict() for e in self.events]
}
class TraceCollector:
"""
Collects and manages traces for agent executions.
Captures events, maintains relationships, and provides query capabilities.
"""
def __init__(self):
"""Initialize trace collector"""
self.traces: Dict[str, Trace] = {}
self.current_trace_id: Optional[str] = None
self.event_stack: List[str] = [] # Stack of event IDs for hierarchy
def start_trace(self, metadata: Optional[Dict[str, Any]] = None) -> str:
"""
Start a new trace.
Args:
metadata: Optional trace metadata
Returns:
Trace ID
"""
trace_id = str(uuid.uuid4())
trace = Trace(
trace_id=trace_id,
start_time=datetime.now(),
metadata=metadata or {}
)
self.traces[trace_id] = trace
self.current_trace_id = trace_id
return trace_id
def end_trace(self, trace_id: str, status: str = "completed"):
"""
End a trace.
Args:
trace_id: Trace identifier
status: Final status (completed, failed)
"""
if trace_id in self.traces:
trace = self.traces[trace_id]
trace.end_time = datetime.now()
trace.status = status
def log_event(
self,
event_type: TraceEventType,
name: str,
input_data: Optional[Dict[str, Any]] = None,
output_data: Optional[Dict[str, Any]] = None,
metadata: Optional[Dict[str, Any]] = None,
error: Optional[str] = None,
duration_ms: Optional[float] = None
) -> str:
"""
Log a trace event.
Args:
event_type: Type of event
name: Event name
input_data: Input data
output_data: Output data
metadata: Additional metadata
error: Error message if applicable
duration_ms: Event duration
Returns:
Event ID
"""
if not self.current_trace_id:
return ""
event_id = str(uuid.uuid4())
parent_id = self.event_stack[-1] if self.event_stack else None
event = TraceEvent(
event_id=event_id,
trace_id=self.current_trace_id,
parent_id=parent_id,
event_type=event_type,
timestamp=datetime.now(),
name=name,
input_data=input_data or {},
output_data=output_data or {},
metadata=metadata or {},
error=error,
duration_ms=duration_ms
)
trace = self.traces[self.current_trace_id]
trace.events.append(event)
return event_id
def start_span(self, name: str, event_type: TraceEventType) -> str:
"""
Start a span (nested event).
Args:
name: Span name
event_type: Event type
Returns:
Event ID
"""
event_id = self.log_event(
event_type=event_type,
name=name,
metadata={'span_start': True}
)
self.event_stack.append(event_id)
return event_id
def end_span(self, event_id: str, output_data: Optional[Dict[str, Any]] = None):
"""
End a span.
Args:
event_id: Event ID to end
output_data: Output data
"""
if self.event_stack and self.event_stack[-1] == event_id:
self.event_stack.pop()
# Update event with output and duration
if self.current_trace_id:
trace = self.traces[self.current_trace_id]
for event in trace.events:
if event.event_id == event_id:
if output_data:
event.output_data = output_data
# Calculate duration from event timestamp to now
event.duration_ms = (datetime.now() - event.timestamp).total_seconds() * 1000
break
def get_trace(self, trace_id: str) -> Optional[Trace]:
"""Get trace by ID"""
return self.traces.get(trace_id)
def get_all_traces(self) -> List[Trace]:
"""Get all traces"""
return list(self.traces.values())
def clear_traces(self):
"""Clear all traces"""
self.traces.clear()
self.current_trace_id = None
self.event_stack.clear()
class TraceAnalyzer:
"""
Analyzes traces to extract insights and metrics.
Provides visualization, statistics, and pattern detection.
"""
def __init__(self):
"""Initialize analyzer"""
pass
def get_execution_path(self, trace: Trace) -> List[str]:
"""
Get execution path as list of event names.
Args:
trace: Trace to analyze
Returns:
List of event names in order
"""
return [event.name for event in trace.events]
def get_performance_stats(self, trace: Trace) -> Dict[str, Any]:
"""
Calculate performance statistics.
Args:
trace: Trace to analyze
Returns:
Performance statistics
"""
events_with_duration = [e for e in trace.events if e.duration_ms]
if not events_with_duration:
return {
'total_duration_ms': trace.duration_ms,
'event_count': len(trace.events),
'avg_event_duration_ms': 0,
'max_event_duration_ms': 0,
'min_event_duration_ms': 0
}
durations = [e.duration_ms for e in events_with_duration]
return {
'total_duration_ms': trace.duration_ms,
'event_count': len(trace.events),
'events_with_timing': len(events_with_duration),
'avg_event_duration_ms': sum(durations) / len(durations),
'max_event_duration_ms': max(durations),
'min_event_duration_ms': min(durations),
'total_event_time_ms': sum(durations)
}
def get_event_type_stats(self, trace: Trace) -> Dict[str, int]:
"""
Count events by type.
Args:
trace: Trace to analyze
Returns:
Event counts by type
"""
stats = {}
for event in trace.events:
event_type = event.event_type.value
stats[event_type] = stats.get(event_type, 0) + 1
return stats
def get_error_summary(self, trace: Trace) -> List[Dict[str, Any]]:
"""
Get summary of errors in trace.
Args:
trace: Trace to analyze
Returns:
List of error details
"""
errors = []
for event in trace.events:
if event.error:
errors.append({
'event_id': event.event_id,
'event_type': event.event_type.value,
'name': event.name,
'error': event.error,
'timestamp': event.timestamp.isoformat()
})
return errors
def visualize_trace(self, trace: Trace) -> str:
"""
Create text-based visualization of trace.
Args:
trace: Trace to visualize
Returns:
Formatted string visualization
"""
lines = []
lines.append(f"Trace: {trace.trace_id}")
lines.append(f"Status: {trace.status}")
lines.append(f"Duration: {trace.duration_ms:.2f}ms")
lines.append(f"Events: {len(trace.events)}")
lines.append("")
lines.append("Execution Path:")
# Build hierarchy
event_map = {e.event_id: e for e in trace.events}
root_events = [e for e in trace.events if e.parent_id is None]
def format_event(event: TraceEvent, indent: int = 0) -> List[str]:
"""Format event with indentation"""
result = []
prefix = " " * indent
duration = f"({event.duration_ms:.1f}ms)" if event.duration_ms else ""
error_mark = " ❌" if event.error else ""
result.append(f"{prefix}├─ {event.event_type.value}: {event.name} {duration}{error_mark}")
# Find children
children = [e for e in trace.events if e.parent_id == event.event_id]
for child in children:
result.extend(format_event(child, indent + 1))
return result
for root in root_events:
lines.extend(format_event(root))
return "\n".join(lines)
class TracedAgent:
"""
Example agent with comprehensive tracing.
Demonstrates trace collection for all agent operations.
"""
def __init__(self, collector: TraceCollector):
"""
Initialize traced agent.
Args:
collector: Trace collector instance
"""
self.collector = collector
self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
def process_query(self, query: str, metadata: Optional[Dict[str, Any]] = None) -> str:
"""
Process query with full tracing.
Args:
query: User query
metadata: Optional metadata
Returns:
Response
"""
# Start trace
trace_id = self.collector.start_trace(metadata={'query': query, **(metadata or {})})
try:
# Log agent start
agent_span = self.collector.start_span("process_query", TraceEventType.AGENT_START)
# Step 1: Analyze query
analysis_span = self.collector.start_span("analyze_query", TraceEventType.DECISION)
query_type = self._analyze_query(query)
self.collector.end_span(analysis_span, {'query_type': query_type})
# Step 2: Generate response
llm_span = self.collector.start_span("llm_call", TraceEventType.LLM_START)
response = self._call_llm(query)
self.collector.end_span(llm_span, {'response': response[:100]})
# End agent span
self.collector.end_span(agent_span, {'response': response})
# End trace successfully
self.collector.end_trace(trace_id, "completed")
return response
except Exception as e:
# Log error
self.collector.log_event(
event_type=TraceEventType.ERROR,
name="agent_error",
error=str(e),
metadata={'exception_type': type(e).__name__}
)
# End trace with failure
self.collector.end_trace(trace_id, "failed")
raise
def _analyze_query(self, query: str) -> str:
"""Analyze query type"""
time.sleep(0.05) # Simulate processing
if '?' in query:
return "question"
else:
return "statement"
def _call_llm(self, query: str) -> str:
"""Call LLM"""
prompt = ChatPromptTemplate.from_template("Answer concisely: {query}")
chain = prompt | self.llm | StrOutputParser()
return chain.invoke({"query": query})
def demonstrate_trace_tracking():
"""Demonstrate trace and lineage tracking"""
print("=" * 80)
print("TRACE/LINEAGE TRACKING DEMONSTRATION")
print("=" * 80)
# Example 1: Basic Trace Collection
print("\n" + "=" * 80)
print("EXAMPLE 1: Basic Trace Collection")
print("=" * 80)
collector = TraceCollector()
# Start a trace
trace_id = collector.start_trace(metadata={'user': 'demo_user', 'session': 'session_001'})
print(f"\nStarted trace: {trace_id[:8]}...")
# Log some events
collector.log_event(
event_type=TraceEventType.AGENT_START,
name="agent_init",
metadata={'agent_type': 'demo'}
)
collector.log_event(
event_type=TraceEventType.LLM_START,
name="llm_call_1",
input_data={'prompt': 'What is AI?'},
duration_ms=1250.5
)
collector.log_event(
event_type=TraceEventType.LLM_END,
name="llm_call_1",
output_data={'response': 'AI is...'},
duration_ms=1250.5
)
collector.end_trace(trace_id, "completed")
trace = collector.get_trace(trace_id)
print(f"\nTrace completed:")
print(f" Events: {len(trace.events)}")
print(f" Duration: {trace.duration_ms:.2f}ms")
print(f" Status: {trace.status}")
# Example 2: Hierarchical Tracing
print("\n" + "=" * 80)
print("EXAMPLE 2: Hierarchical Tracing with Spans")
print("=" * 80)
collector.clear_traces()
trace_id = collector.start_trace()
# Parent span
parent_span = collector.start_span("main_task", TraceEventType.AGENT_START)
time.sleep(0.1)
# Child span 1
child_span1 = collector.start_span("subtask_1", TraceEventType.TOOL_START)
time.sleep(0.05)
collector.end_span(child_span1, {'result': 'success'})
# Child span 2
child_span2 = collector.start_span("subtask_2", TraceEventType.TOOL_START)
time.sleep(0.03)
collector.end_span(child_span2, {'result': 'success'})
collector.end_span(parent_span, {'status': 'completed'})
collector.end_trace(trace_id, "completed")
trace = collector.get_trace(trace_id)
print(f"\nHierarchical trace with {len(trace.events)} events:")
for event in trace.events:
indent = " " if event.parent_id else ""
duration = f" ({event.duration_ms:.1f}ms)" if event.duration_ms else ""
print(f"{indent}{event.event_type.value}: {event.name}{duration}")
# Example 3: Traced Agent Execution
print("\n" + "=" * 80)
print("EXAMPLE 3: Full Agent Execution with Tracing")
print("=" * 80)
collector.clear_traces()
agent = TracedAgent(collector)
print("\nExecuting query with tracing...")
query = "What is machine learning?"
response = agent.process_query(query, metadata={'priority': 'high'})
print(f"\nResponse: {response[:100]}...")
# Get the trace
traces = collector.get_all_traces()
if traces:
trace = traces[0]
print(f"\nTrace captured:")
print(f" Trace ID: {trace.trace_id[:8]}...")
print(f" Events: {len(trace.events)}")
print(f" Duration: {trace.duration_ms:.2f}ms")
print(f" Status: {trace.status}")
# Example 4: Trace Analysis
print("\n" + "=" * 80)
print("EXAMPLE 4: Trace Analysis and Insights")
print("=" * 80)
analyzer = TraceAnalyzer()
if traces:
trace = traces[0]
# Performance stats
print("\nPerformance Statistics:")
stats = analyzer.get_performance_stats(trace)
for key, value in stats.items():
if isinstance(value, float):
print(f" {key}: {value:.2f}")
else:
print(f" {key}: {value}")
# Event type stats
print("\nEvent Type Distribution:")
type_stats = analyzer.get_event_type_stats(trace)
for event_type, count in sorted(type_stats.items()):
print(f" {event_type}: {count}")
# Execution path
print("\nExecution Path:")
path = analyzer.get_execution_path(trace)
for i, step in enumerate(path, 1):
print(f" {i}. {step}")
# Example 5: Trace Visualization
print("\n" + "=" * 80)
print("EXAMPLE 5: Trace Visualization")
print("=" * 80)
if traces:
visualization = analyzer.visualize_trace(traces[0])
print("\n" + visualization)
# Example 6: Error Tracing
print("\n" + "=" * 80)
print("EXAMPLE 6: Error Tracing and Debugging")
print("=" * 80)
collector.clear_traces()
trace_id = collector.start_trace(metadata={'test': 'error_handling'})
# Log successful operation
collector.log_event(
event_type=TraceEventType.AGENT_START,
name="operation_start"
)
# Log error
collector.log_event(
event_type=TraceEventType.ERROR,
name="operation_failed",
error="Simulated error: Connection timeout",
metadata={'error_code': 'TIMEOUT', 'retry_count': 3}
)
collector.end_trace(trace_id, "failed")
trace = collector.get_trace(trace_id)
errors = analyzer.get_error_summary(trace)
print(f"\nTrace with errors:")
print(f" Status: {trace.status}")
print(f" Error count: {len(errors)}")
if errors:
print("\nError Details:")
for error in errors:
print(f" Event: {error['name']}")
print(f" Type: {error['event_type']}")
print(f" Error: {error['error']}")
print(f" Time: {error['timestamp']}")
# Example 7: Trace Export and Storage
print("\n" + "=" * 80)
print("EXAMPLE 7: Trace Export for Storage/Analysis")
print("=" * 80)
if traces:
trace = traces[0]
trace_dict = trace.to_dict()
print("\nTrace exported to dictionary:")
print(f" Trace ID: {trace_dict['trace_id'][:8]}...")
print(f" Duration: {trace_dict['duration_ms']:.2f}ms")
print(f" Event count: {trace_dict['event_count']}")
print(f" Status: {trace_dict['status']}")
# Simulate JSON export
trace_json = json.dumps(trace_dict, indent=2, default=str)
print(f"\nJSON size: {len(trace_json)} bytes")
print(f"First 200 chars:\n{trace_json[:200]}...")
# Summary
print("\n" + "=" * 80)
print("TRACE/LINEAGE TRACKING SUMMARY")
print("=" * 80)
print("""
Trace Tracking Benefits:
1. Complete Visibility: See every step of agent execution
2. Debugging Power: Reproduce and understand issues
3. Performance Insights: Identify bottlenecks
4. Compliance: Document decision-making
5. Reproducibility: Replay executions exactly
6. Optimization: Data-driven improvements
Key Components:
1. Trace Events
- Agent lifecycle (start, end)
- LLM calls (input, output, tokens, cost)
- Tool invocations (name, args, results)
- Decision points (conditions, branches)
- State changes (before, after)
- Errors (type, message, context)
2. Event Hierarchy
- Parent-child relationships
- Nested spans for complex operations
- Call stack preservation
- Context propagation
- Timing at each level
3. Trace Metadata
- User/session information
- Request context
- Environment details
- Custom attributes
- Tags for filtering
4. Analysis Capabilities
- Performance profiling
- Path visualization
- Error pattern detection
- Statistical analysis
- Comparison tools
What to Trace:
1. Essential (Always)
- All LLM calls
- Tool invocations
- Agent start/end
- Errors and exceptions
- Final results
2. Important (Most cases)
- Decision points
- State changes
- Performance metrics
- Resource usage
- Intermediate results
3. Detailed (Development/Debug)
- Full context
- Internal calculations
- Validation steps
- Cache hits/misses
- Detailed timing
Trace Structure:
```
Trace
├── trace_id (unique identifier)
├── start_time
├── end_time
├── status (running, completed, failed)
├── metadata (user, session, etc.)
└── events []
├── Event 1
│ ├── event_id
│ ├── parent_id
│ ├── event_type
│ ├── timestamp
│ ├── duration_ms
│ ├── input_data
│ ├── output_data
│ └── metadata
└── Event 2
└── ...
```
Best Practices:
1. Trace Organization
- Unique trace IDs (UUID v4)
- Clear event names
- Consistent structure
- Rich metadata
- Logical grouping
2. Performance
- Async trace writes
- Batch uploads
- Sampling in production
- Compression for storage
- Resource limits
3. Privacy
- PII redaction
- Sensitive data filtering
- Access controls
- Encryption at rest/transit
- Retention policies
4. Storage
- Hot: 7-30 days (fast access)
- Warm: 31-90 days (slower)
- Cold: 91-365 days (archive)
- Automatic cleanup
- Efficient formats
Production Deployment:
1. Sampling Strategy
- Development: 100%
- Staging: 20-50%
- Production: 1-10% baseline
- Errors: Always 100%
- Important users: Higher rate
2. Performance Impact
- Target: < 5% overhead
- Measure actual impact
- Optimize hot paths
- Circuit breaker
- Degradation handling
3. Integration
- LangSmith for LangChain
- OpenTelemetry standard
- Custom backends
- Log aggregation systems
- APM tools integration
4. Analysis Workflow
- Real-time monitoring
- Periodic analysis
- Alert on patterns
- Dashboard visualization
- Export for deep analysis
When to Use Tracing:
✓ All production AI systems
✓ Complex multi-step agents
✓ Debugging difficult issues
✓ Performance optimization
✓ Compliance requirements
✓ Quality assurance
✗ Simple single-shot calls (maybe)
✗ Privacy-critical scenarios (with redaction)
Common Use Cases:
1. Debugging
- Why did agent fail?
- What was the decision path?
- What inputs caused issue?
- How to reproduce?
2. Optimization
- Which step is slowest?
- Where are bottlenecks?
- Cost breakdown by component
- Cache effectiveness
3. Monitoring
- Error rate trends
- Performance degradation
- Usage patterns
- Resource consumption
4. Compliance
- Decision documentation
- Audit trails
- Regulatory requirements
- Accountability
Tools & Services:
- LangSmith: Native LangChain tracing
- OpenTelemetry: Standard observability
- Jaeger: Distributed tracing
- DataDog: APM and tracing
- Custom: Roll your own
ROI Analysis:
- Debug time: 50-90% reduction
- Issue resolution: 2-10x faster
- Performance gains: 20-50%
- Compliance: Critical for regulated industries
- Cost: Low overhead, high value
""")
print("\n" + "=" * 80)
print("Pattern 094 (Trace/Lineage Tracking) demonstration complete!")
print("=" * 80)
if __name__ == "__main__":
demonstrate_trace_tracking()
Agentic Design Pattern: Trace Tracking
Pattern 094: Trace Tracking – Line-by-Line Explanation
Overview
This document provides a detailed line-by-line explanation of the Trace Tracking pattern implementation.
Pattern Description
Pattern 094: Trace/Lineage Tracking
Description: Trace and Lineage Tracking captures the complete execution path of an agent, including all decisions, actions, intermediate states, and data transformations. This pattern is essential for debugging complex agent behaviors, understanding how decisions were made, ensuring compliance, and optimizing performance. Comprehensive tracing enables reproducibility, auditing, and continuous improvement of AI systems.
Trace tracking provides:
- Complete execution history
- Decision point visibility
- Data lineage and provenance
- Performance metrics per step
- Error context and stack traces
- Reproducibility information
Components: 1. Trace Capture – Execution events (start, end, error) – Decision points and branches – Input/output at each step – Timestamps and durations – Resource consumption – Agent state snapshots
2. Trace Storage
- Structured trace format
- Hierarchical relationships
- Searchable metadata
- Efficient retrieval
- Long-term retention
3. Trace Analysis
- Path visualization
- Performance analysis
- Bottleneck identification
- Error pattern detection
- Comparison tools
4. Trace Replay
- Reproduce executions
- Debug specific paths
- Test alternative branches
- Validate fixes
- Training data generation
Use Cases: 1. Debugging – Understand unexpected behavior – Reproduce production issues – Identify error causes – Validate fixes – Root cause analysis
2. Performance Optimization
- Identify bottlenecks
- Measure step durations
- Analyze resource usage
- Compare implementations
- Optimize hot paths
3. Compliance & Auditing
- Document decision making
- Prove regulatory compliance
- Audit trails for sensitive operations
- Accountability tracking
- Legal requirements
4. Quality Assurance
- Validate agent behavior
- Ensure determinism
- Track test coverage
- Regression detection
- Change impact analysis
5. Agent Development
- Understand agent flow
- Optimize prompts
- Refine decision logic
- Test edge cases
- Documentation generation
LangChain Implementation: LangChain supports tracing through: – LangSmith for comprehensive tracing – Custom callback handlers – OpenTelemetry integration – Structured logging – Trace context propagation
Key Features: 1. Comprehensive Capture – Every agent step traced – Full context preservation – Hierarchical relationships – Performance metrics – Error details
2. Structured Format
- Consistent trace structure
- Searchable attributes
- Machine-readable format
- Human-readable display
- Standard schemas
3. Powerful Analysis
- Path visualization
- Query capabilities
- Statistical analysis
- Comparison tools
- Pattern detection
4. Production Ready
- Low overhead (< 5% impact)
- Efficient storage
- Privacy controls
- Sampling strategies
- Scalable architecture
Best Practices: 1. What to Trace – All LLM calls (input, output, model, tokens) – Tool invocations (name, args, results) – Decision points (conditions, chosen branch) – State changes (before, after) – Errors (type, message, stack) – Performance (start, end, duration)
2. Trace Organization
- Clear trace IDs (unique, sortable)
- Parent-child relationships
- Logical grouping (by session, user, task)
- Consistent naming
- Rich metadata
3. Performance Considerations
- Sampling in production (1-10%)
- Async trace writes
- Compression for storage
- Automatic rotation
- Resource limits
4. Privacy & Security
- PII redaction
- Access controls
- Encryption at rest
- Retention policies
- Compliance adherence
Trade-offs: Advantages: – Complete execution visibility – Powerful debugging capabilities – Compliance documentation – Performance insights – Reproducible executions – Historical analysis
Disadvantages:
- Storage overhead
- Performance impact (small)
- Complexity in setup
- Privacy concerns
- Analysis tools needed
- Maintenance overhead
Production Considerations: 1. Sampling Strategy – 100% for development – 10-20% for staging – 1-5% for production (baseline) – 100% for errors – Adaptive based on load
2. Storage Management
- Hot storage: 7-30 days
- Cold storage: 90-365 days
- Archive: > 1 year
- Compression for old traces
- Automatic cleanup
3. Performance Impact
- Target: < 5% overhead
- Async writes preferred
- Batch uploads
- Local buffering
- Circuit breaker
4. Privacy Compliance
- Automatic PII detection
- Redaction policies
- Consent tracking
- Data retention limits
- Right to deletion
5. Analysis Tools
- Web UI for browsing
- Query interface
- Visualization tools
- Alert integration
- Export capabilities
File Statistics
- Total Lines: 1067
- Classes: 6
- Functions: 1
- Constants: 0
Module Documentation
"""
Pattern 094: Trace/Lineage Tracking
Description:
Trace and Lineage Tracking captures the complete execution path of an agent, including
all decisions, actions, intermediate states, and data transformations. This pattern is
essential for debugging complex agent behaviors, understanding how decisions were made,
ensuring compliance, and optimizing performance. Comprehensive tracing enables
reproducibility, auditing, and continuous improvement of AI systems.
Trace tracking provides:
- Complete execution history
- Decision point visibility
- Data lineage and provenance
- Performance metrics per step
- Error context and stack traces
- Reproducibility information
Components:
1. Trace Capture
- Execution events (start, end, error)
- Decision points and branches
- Input/output at each step
- Timestamps and durations
- Resource consumption
- Agent state snapshots
2. Trace Storage
- Structured trace format
- Hierarchical relationships
- Searchable metadata
- Efficient retrieval
- Long-term retention
3. Trace Analysis
- Path visualization
- Performance analysis
- Bottleneck identification
- Error pattern detection
- Comparison tools
4. Trace Replay
- Reproduce executions
- Debug specific paths
- Test alternative branches
- Validate fixes
- Training data generation
Use Cases:
1. Debugging
- Understand unexpected behavior
- Reproduce production issues
- Identify error causes
- Validate fixes
- Root cause analysis
2. Performance Optimization
- Identify bottlenecks
- Measure step durations
- Analyze resource usage
- Compare implementations
- Optimize hot paths
3. Compliance & Auditing
- Document decision making
- Prove regulatory compliance
- Audit trails for sensitive operations
- Accountability tracking
- Legal requirements
4. Quality Assurance
- Validate agent behavior
- Ensure determinism
- Track test coverage
- Regression detection
- Change impact analysis
5. Agent Development
- Understand agent flow
- Optimize prompts
- Refine decision logic
- Test edge cases
- Documentation generation
LangChain Implementation:
LangChain supports tracing through:
- LangSmith for comprehensive tracing
- Custom callback handlers
- OpenTelemetry integration
- Structured logging
- Trace context propagation
Key Features:
1. Comprehensive Capture
- Every agent step traced
- Full context preservation
- Hierarchical relationships
- Performance metrics
- Error details
2. Structured Format
- Consistent trace structure
- Searchable attributes
- Machine-readable format
- Human-readable display
- Standard schemas
3. Powerful Analysis
- Path visualization
- Query capabilities
- Statistical analysis
- Comparison tools
- Pattern detection
4. Production Ready
- Low overhead (< 5% impact)
- Efficient storage
- Privacy controls
- Sampling strategies
- Scalable architecture
Best Practices:
1. What to Trace
- All LLM calls (input, output, model, tokens)
- Tool invocations (name, args, results)
- Decision points (conditions, chosen branch)
- State changes (before, after)
- Errors (type, message, stack)
- Performance (start, end, duration)
2. Trace Organization
- Clear trace IDs (unique, sortable)
- Parent-child relationships
- Logical grouping (by session, user, task)
- Consistent naming
- Rich metadata
3. Performance Considerations
- Sampling in production (1-10%)
- Async trace writes
- Compression for storage
- Automatic rotation
- Resource limits
4. Privacy & Security
- PII redaction
- Access controls
- Encryption at rest
- Retention policies
- Compliance adherence
Trade-offs:
Advantages:
- Complete execution visibility
- Powerful debugging capabilities
- Compliance documentation
- Performance insights
- Reproducible executions
- Historical analysis
Disadvantages:
- Storage overhead
- Performance impact (small)
- Complexity in setup
- Privacy concerns
- Analysis tools needed
- Maintenance overhead
Production Considerations:
1. Sampling Strategy
- 100% for development
- 10-20% for staging
- 1-5% for production (baseline)
- 100% for errors
- Adaptive based on load
2. Storage Management
- Hot storage: 7-30 days
- Cold storage: 90-365 days
- Archive: > 1 year
- Compression for old traces
- Automatic cleanup
3. Performance Impact
- Target: < 5% overhead
- Async writes preferred
- Batch uploads
- Local buffering
- Circuit breaker
4. Privacy Compliance
- Automatic PII detection
- Redaction policies
- Consent tracking
- Data retention limits
- Right to deletion
5. Analysis Tools
- Web UI for browsing
- Query interface
- Visualization tools
- Alert integration
- Export capabilities
"""
Purpose: This module implements the Trace Tracking pattern, providing a structured approach to agent design and implementation.
Import Statements
Standard Library Imports
import dataclasses.asdict
import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import enum.Enum
import json
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import time
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
Purpose: Core Python functionality:
– dataclasses.asdict: Simplified class definitions with @dataclass
– dataclasses.dataclass: Simplified class definitions with @dataclass
– dataclasses.field: Simplified class definitions with @dataclass
– datetime.datetime: Date and time handling
– enum.Enum: Enumeration support
– json: JSON serialization/deserialization
– langchain_core.output_parsers.StrOutputParser: Regular expression operations
– langchain_core.prompts.ChatPromptTemplate: Regular expression operations
– os: Environment variable access and system operations
– time: Time-related functions
– 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_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
from uuid
Purpose: Additional dependencies for specialized functionality.
Classes
Class: TraceEventType
Lines 223-233
Inherits from: Enum
Purpose:
Types of trace events
Class: TraceEvent
Lines 237-258
Purpose:
Represents a single event in a trace
Methods:
to_dict(self)
Lines 253-258
Purpose:
Convert to dictionary
Parameters:
Returns: <ast.Subscript object at 0x000001A06D624550>
Class: Trace
Lines 262-289
Purpose:
Complete trace of an agent execution
Methods:
duration_ms(self)
Lines 272-276
Decorators: property
Purpose:
Calculate total duration
Parameters:
Returns: float
to_dict(self)
Lines 278-289
Purpose:
Convert to dictionary
Parameters:
Returns: <ast.Subscript object at 0x000001A06D645990>
Class: TraceCollector
Lines 292-441
Purpose:
Collects and manages traces for agent executions.
Captures events, maintains relationships, and provides query capabilities.
Methods:
__init__(self)
Lines 299-303
Purpose:
Initialize trace collector
Parameters:
start_trace(self, metadata)
Lines 305-323
Purpose:
Start a new trace.
Args:
metadata: Optional trace metadata
Returns:
Trace ID
Parameters:
– metadata: Metadata
Returns: str
end_trace(self, trace_id, status)
Lines 325-336
Purpose:
End a trace.
Args:
trace_id: Trace identifier
status: Final status (completed, failed)
Parameters:
– trace_id: Trace id
– status: Status
log_event(self, event_type, name, input_data, output_data, metadata, error, duration_ms)
Lines 338-386
Purpose:
Log a trace event.
Args:
event_type: Type of event
name: Event name
input_data: Input data
output_data: Output data
metadata: Additional metadata
error: Error message if applicable
duration_ms: Event duration
Returns:
Event ID
Parameters:
– event_type: Event type
– name: Name
– input_data: Input data
– output_data: Output data
– metadata: Metadata
– error: Error
– duration_ms: Duration ms
Returns: str
start_span(self, name, event_type)
Lines 388-405
Purpose:
Start a span (nested event).
Args:
name: Span name
event_type: Event type
Returns:
Event ID
Parameters:
– name: Name
– event_type: Event type
Returns: str
end_span(self, event_id, output_data)
Lines 407-427
Purpose:
End a span.
Args:
event_id: Event ID to end
output_data: Output data
Parameters:
– event_id: Event id
– output_data: Output data
get_trace(self, trace_id)
Lines 429-431
Purpose:
Get trace by ID
Parameters:
– trace_id: Trace id
Returns: <ast.Subscript object at 0x000001A06D7C9750>
get_all_traces(self)
Lines 433-435
Purpose:
Get all traces
Parameters:
Returns: <ast.Subscript object at 0x000001A06D7C85D0>
clear_traces(self)
Lines 437-441
Purpose:
Clear all traces
Parameters:
Class: TraceAnalyzer
Lines 444-579
Purpose:
Analyzes traces to extract insights and metrics.
Provides visualization, statistics, and pattern detection.
Methods:
__init__(self)
Lines 451-453
Purpose:
Initialize analyzer
Parameters:
get_execution_path(self, trace)
Lines 455-465
Purpose:
Get execution path as list of event names.
Args:
trace: Trace to analyze
Returns:
List of event names in order
Parameters:
– trace: Trace
Returns: <ast.Subscript object at 0x000001A06D660DD0>
get_performance_stats(self, trace)
Lines 467-498
Purpose:
Calculate performance statistics.
Args:
trace: Trace to analyze
Returns:
Performance statistics
Parameters:
– trace: Trace
Returns: <ast.Subscript object at 0x000001A06D888850>
get_event_type_stats(self, trace)
Lines 500-514
Purpose:
Count events by type.
Args:
trace: Trace to analyze
Returns:
Event counts by type
Parameters:
– trace: Trace
Returns: <ast.Subscript object at 0x000001A06D88B6D0>
get_error_summary(self, trace)
Lines 516-536
Purpose:
Get summary of errors in trace.
Args:
trace: Trace to analyze
Returns:
List of error details
Parameters:
– trace: Trace
Returns: <ast.Subscript object at 0x000001A06D75B110>
visualize_trace(self, trace)
Lines 538-579
Purpose:
Create text-based visualization of trace.
Args:
trace: Trace to visualize
Returns:
Formatted string visualization
Parameters:
– trace: Trace
Returns: str
Class: TracedAgent
Lines 582-661
Purpose:
Example agent with comprehensive tracing.
Demonstrates trace collection for all agent operations.
Methods:
__init__(self, collector)
Lines 589-597
Purpose:
Initialize traced agent.
Args:
collector: Trace collector instance
Parameters:
– collector: Collector
process_query(self, query, metadata)
Lines 599-647
Purpose:
Process query with full tracing.
Args:
query: User query
metadata: Optional metadata
Returns:
Response
Parameters:
– query: Input query or question
– metadata: Metadata
Returns: str
_analyze_query(self, query)
Lines 649-655
Purpose:
Analyze query type
Parameters:
– query: Input query or question
Returns: str
_call_llm(self, query)
Lines 657-661
Purpose:
Call LLM
Parameters:
– query: Input query or question
Returns: str
Functions
demonstrate_trace_tracking()
Lines 664-1062
Purpose:
Demonstrate trace and lineage tracking
Key Concepts Summary
Pattern Overview
The Trace Tracking 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 093
- Pattern 095
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 Trace Tracking 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: 094_trace_tracking.py
Total Lines: 1067
Pattern Category: AI Agent Design Pattern #094
Generated from actual implementation code
For educational and reference purposes