Session 2 - Module D: Performance & Monitoring (50 minutes)¶
Prerequisites: Session 2 Core Section Complete
Target Audience: Production-focused developers
Cognitive Load: 6 optimization concepts
Module Overview¶
This module focuses on performance benchmarking, monitoring, and cost optimization for production LangChain applications. You'll learn how to measure performance, implement comprehensive monitoring, and optimize resource usage for enterprise-scale deployments.
Learning Objectives¶
By the end of this module, you will: - Implement performance benchmarking and optimization strategies for LangChain agents - Set up comprehensive monitoring and observability systems for production applications - Design cost optimization strategies for API usage and resource management - Create real-time performance dashboards and alerting systems
Part 1: Performance Benchmarking & Optimization (20 minutes)¶
Comprehensive Performance Measurement¶
🗂️ File: src/session2/performance_benchmarking.py - Performance measurement and optimization
Production LangChain applications require systematic performance measurement across multiple dimensions to ensure optimal user experience and resource utilization. The foundation includes comprehensive imports and metrics structure:
import time
import asyncio
import statistics
import psutil
import tracemalloc
from typing import Dict, List, Any, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from concurrent.futures import ThreadPoolExecutor, as_completed
import json
import logging
@dataclass
class PerformanceMetrics:
"""Comprehensive performance metrics for LangChain agents"""
# Response Time Metrics
avg_response_time: float = 0.0
p50_response_time: float = 0.0
p90_response_time: float = 0.0
The metrics dataclass captures response time percentiles essential for understanding user experience under different load conditions.
p95_response_time: float = 0.0
p99_response_time: float = 0.0
# Throughput Metrics
requests_per_second: float = 0.0
concurrent_requests_handled: int = 0
max_concurrent_capacity: int = 0
# Resource Utilization
memory_usage_mb: float = 0.0
cpu_usage_percent: float = 0.0
memory_peak_mb: float = 0.0
# LLM-Specific Metrics
tokens_per_second: float = 0.0
avg_prompt_tokens: float = 0.0
avg_completion_tokens: float = 0.0
llm_api_latency: float = 0.0
Resource and LLM-specific metrics track memory consumption, CPU usage, and token processing rates essential for cost optimization and capacity planning.
# Tool Usage Metrics
tool_execution_time: float = 0.0
tool_success_rate: float = 0.0
cache_hit_rate: float = 0.0
# Error Metrics
error_rate: float = 0.0
timeout_rate: float = 0.0
retry_rate: float = 0.0
Tool and error metrics provide insights into system reliability and performance bottlenecks across different components.
# Quality Metrics
response_quality_score: float = 0.0
user_satisfaction_score: float = 0.0
# Operational Metrics
uptime_percentage: float = 0.0
deployment_frequency: float = 0.0
timestamp: datetime = field(default_factory=datetime.now)
Quality and operational metrics ensure the system maintains high standards for user experience and operational excellence.
The benchmark suite orchestrates comprehensive performance testing across multiple scenarios:
class PerformanceBenchmarkSuite:
"""Comprehensive performance benchmarking for LangChain agents"""
def __init__(self, agent_factory: Callable, config: Dict[str, Any]):
self.agent_factory = agent_factory
self.config = config
self.performance_history: List[PerformanceMetrics] = []
self.logger = logging.getLogger(__name__)
# Initialize monitoring
tracemalloc.start()
The suite initializes with an agent factory for creating test instances and starts memory tracing to capture detailed performance data.
async def run_comprehensive_benchmark(self, test_scenarios: List[Dict[str, Any]]) -> Dict[str, Any]:
"""Execute comprehensive performance benchmark across multiple scenarios"""
benchmark_results = {
"benchmark_id": f"benchmark_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
"start_time": datetime.now(),
"config": self.config,
"scenarios": {},
"summary": {},
"recommendations": []
}
try:
# Run each test scenario
for scenario in test_scenarios:
scenario_name = scenario["name"]
self.logger.info(f"Running benchmark scenario: {scenario_name}")
scenario_results = await self._run_scenario_benchmark(scenario)
benchmark_results["scenarios"][scenario_name] = scenario_results
Benchmark execution creates unique IDs for tracking and organizes results by scenario, enabling comprehensive performance analysis across different use cases.
# Generate summary and recommendations
benchmark_results["summary"] = self._generate_benchmark_summary(benchmark_results["scenarios"])
benchmark_results["recommendations"] = self._generate_optimization_recommendations(benchmark_results["summary"])
benchmark_results["end_time"] = datetime.now()
benchmark_results["total_duration"] = (benchmark_results["end_time"] - benchmark_results["start_time"]).total_seconds()
return benchmark_results
except Exception as e:
self.logger.error(f"Benchmark failed: {str(e)}")
benchmark_results["error"] = str(e)
return benchmark_results
Benchmark finalization includes summary generation and optimization recommendations. Duration calculation provides total execution time while error handling ensures graceful failure recovery.
async def _run_scenario_benchmark(self, scenario: Dict[str, Any]) -> Dict[str, Any]:
"""Run performance benchmark for a specific scenario with comprehensive monitoring"""
scenario_config = {
"name": scenario["name"],
"description": scenario.get("description", ""),
"concurrent_users": scenario.get("concurrent_users", 1),
"requests_per_user": scenario.get("requests_per_user", 10),
"test_duration_seconds": scenario.get("test_duration_seconds", 60),
"warmup_requests": scenario.get("warmup_requests", 5),
"test_data": scenario.get("test_data", [])
}
Scenario configuration extraction normalizes benchmark parameters with sensible defaults. Each parameter controls different aspects of load testing including concurrency, volume, and duration.
# Start resource monitoring with baseline measurements
start_memory = psutil.Process().memory_info().rss / 1024 / 1024 # MB
start_cpu = psutil.cpu_percent()
Baseline resource monitoring captures initial system state before load testing. Memory and CPU measurements provide reference points for resource usage analysis.
# Warmup phase to stabilize performance metrics
await self._run_warmup_phase(scenario_config)
# Main benchmark phase with full load testing
performance_data = await self._run_load_test(scenario_config)
Warmup phase eliminates cold-start effects that could skew performance measurements. Main benchmark phase executes the actual load test with concurrent users and comprehensive monitoring.
# Resource measurement after load testing
end_memory = psutil.Process().memory_info().rss / 1024 / 1024 # MB
end_cpu = psutil.cpu_percent()
# Calculate comprehensive performance metrics
metrics = self._calculate_performance_metrics(
performance_data,
start_memory, end_memory,
start_cpu, end_cpu
)
Post-test resource measurement captures system state changes during load testing. Metrics calculation combines timing, throughput, and resource data into comprehensive performance analysis.
return {
"config": scenario_config,
"metrics": metrics,
"raw_data": performance_data,
"resource_usage": {
"memory_delta_mb": end_memory - start_memory,
"cpu_usage_percent": end_cpu,
"peak_memory_mb": max(performance_data.get("memory_samples", [start_memory]))
}
}
Structured result packaging includes configuration, computed metrics, raw performance data, and resource usage analysis for comprehensive benchmark reporting.
Concurrent Load Testing Implementation¶
The load testing system executes comprehensive performance analysis with concurrent user simulation and real-time monitoring.
async def _run_load_test(self, config: Dict[str, Any]) -> Dict[str, Any]:
"""Execute concurrent load test with comprehensive monitoring"""
concurrent_users = config["concurrent_users"]
requests_per_user = config["requests_per_user"]
test_data = config["test_data"]
Load test parameter extraction from configuration enables flexible test scenarios. Concurrent users control parallelism while requests per user determines total load volume.
# Prepare test data for load testing
if not test_data:
test_data = self._generate_test_data(requests_per_user * concurrent_users)
Test data generation creates sufficient inputs for the complete load test. Auto-generation ensures adequate test coverage when specific test data isn't provided.
# Track all performance data during execution
all_response_times = []
all_token_counts = []
error_count = 0
timeout_count = 0
memory_samples = []
Performance tracking variables collect comprehensive metrics during load testing. Response times, token usage, errors, and memory usage provide complete performance visibility.
Semaphore implementation controls concurrent request execution to match specified user count. Proper concurrency control prevents resource exhaustion while maintaining realistic load patterns.
Individual Request Execution and Monitoring¶
Each request execution includes comprehensive monitoring and error handling to capture complete performance characteristics.
async def run_single_request(test_input: str) -> Dict[str, Any]:
"""Execute single request with comprehensive monitoring"""
async with semaphore:
start_time = time.time()
Request execution begins with semaphore acquisition and timing start. Semaphore ensures concurrency limits while timing captures complete request lifecycle.
try:
# Create agent instance for request
agent = self.agent_factory()
# Monitor memory before request execution
current_memory = psutil.Process().memory_info().rss / 1024 / 1024
memory_samples.append(current_memory)
Agent instantiation and memory monitoring prepare for request execution. Memory sampling enables resource usage tracking across concurrent requests.
# Execute request with timeout protection
response = await asyncio.wait_for(
agent.arun(test_input),
timeout=self.config.get("request_timeout", 30)
)
end_time = time.time()
response_time = end_time - start_time
Request execution with timeout protection prevents hanging requests from skewing performance measurements. Response time calculation provides core performance metrics.
# Extract token information for cost analysis
prompt_tokens = getattr(agent, 'last_prompt_tokens', 0)
completion_tokens = getattr(agent, 'last_completion_tokens', 0)
return {
"success": True,
"response_time": response_time,
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"response_length": len(response) if response else 0,
"memory_usage": current_memory
}
Successful request processing captures comprehensive metrics including timing, token usage, response characteristics, and resource consumption for detailed analysis.
except asyncio.TimeoutError:
nonlocal timeout_count
timeout_count += 1
return {
"success": False,
"error_type": "timeout",
"response_time": time.time() - start_time
}
Timeout handling tracks requests that exceed time limits while capturing partial timing data. Timeout counting enables reliability analysis.
except Exception as e:
nonlocal error_count
error_count += 1
return {
"success": False,
"error_type": type(e).__name__,
"error_message": str(e),
"response_time": time.time() - start_time
}
General exception handling captures all other errors with detailed error information. Error categorization and timing help identify failure patterns and performance impact.
Concurrent Execution and Result Processing¶
All requests execute concurrently with comprehensive result aggregation for complete performance analysis.
Now we execute all requests concurrently and process the results:
# Execute all requests concurrently with exception handling
tasks = [run_single_request(data) for data in test_data]
results = await asyncio.gather(*tasks, return_exceptions=True)
Concurrent task execution uses asyncio.gather for maximum parallelism. Exception handling ensures partial failures don't abort the entire load test.
Next, we process and aggregate the performance data from all requests:
# Process results and aggregate performance data
successful_results = []
for result in results:
if isinstance(result, dict) and result.get("success"):
successful_results.append(result)
all_response_times.append(result["response_time"])
all_token_counts.append(result.get("prompt_tokens", 0) + result.get("completion_tokens", 0))
Result processing filters successful requests and aggregates performance metrics. Response time and token count aggregation enables statistical analysis.
Finally, we return comprehensive test results for analysis:
return {
"response_times": all_response_times,
"token_counts": all_token_counts,
"successful_requests": len(successful_results),
"total_requests": len(results),
"error_count": error_count,
"timeout_count": timeout_count,
"memory_samples": memory_samples,
"raw_results": successful_results
}
Comprehensive result packaging includes all performance data, success/failure counts, and raw results for detailed analysis and reporting.
Performance Metrics Calculation¶
Comprehensive performance metrics calculation transforms raw load test data into actionable insights and statistical analysis.
def _calculate_performance_metrics(self, performance_data: Dict[str, Any],
start_memory: float, end_memory: float,
start_cpu: float, end_cpu: float) -> PerformanceMetrics:
"""Calculate comprehensive performance metrics from load test data"""
response_times = performance_data["response_times"]
total_requests = performance_data["total_requests"]
successful_requests = performance_data["successful_requests"]
token_counts = performance_data["token_counts"]
Performance data extraction prepares metrics calculation from load test results. Response times, request counts, and token usage provide comprehensive analysis foundation.
Empty response validation prevents calculation errors when all requests fail. Default empty metrics provide consistent interface for failed load tests.
Let's calculate performance metrics step by step. First, we analyze response time patterns:
# Response time metrics with percentile analysis
sorted_times = sorted(response_times)
avg_response_time = statistics.mean(response_times)
p50 = sorted_times[len(sorted_times) // 2]
p90 = sorted_times[int(len(sorted_times) * 0.9)]
p95 = sorted_times[int(len(sorted_times) * 0.95)]
p99 = sorted_times[int(len(sorted_times) * 0.99)]
Response time analysis includes mean and percentile calculations. Percentiles (p50, p90, p95, p99) reveal response time distribution and identify outliers.
Next, we calculate throughput metrics for capacity planning:
# Throughput metrics for capacity planning
total_test_time = max(response_times) if response_times else 1
requests_per_second = successful_requests / total_test_time
Throughput calculation measures system capacity under load. Requests per second provides key scaling metric for capacity planning and performance optimization.
We analyze token usage for cost estimation:
# Token metrics for cost analysis
avg_tokens = statistics.mean(token_counts) if token_counts else 0
tokens_per_second = sum(token_counts) / total_test_time if token_counts else 0
Token usage analysis enables cost estimation and resource planning. Average tokens and tokens per second help optimize AI model usage and predict operational costs.
Error metrics help assess system reliability:
# Error metrics for reliability analysis
error_rate = (total_requests - successful_requests) / total_requests if total_requests > 0 else 0
timeout_rate = performance_data["timeout_count"] / total_requests if total_requests > 0 else 0
Error rate calculation measures system reliability under load. Error and timeout rates help identify stability issues and capacity limits.
Memory analysis tracks resource utilization patterns:
# Memory metrics for resource utilization
memory_samples = performance_data.get("memory_samples", [start_memory])
peak_memory = max(memory_samples) if memory_samples else start_memory
avg_memory = statistics.mean(memory_samples) if memory_samples else start_memory
Memory analysis tracks resource utilization during load testing. Peak and average memory usage help identify memory leaks and resource requirements.
Finally, we assemble the comprehensive performance metrics:
return PerformanceMetrics(
avg_response_time=avg_response_time,
p50_response_time=p50,
p90_response_time=p90,
p95_response_time=p95,
p99_response_time=p99,
requests_per_second=requests_per_second,
concurrent_requests_handled=successful_requests,
memory_usage_mb=avg_memory,
memory_peak_mb=peak_memory,
cpu_usage_percent=end_cpu,
tokens_per_second=tokens_per_second,
avg_prompt_tokens=avg_tokens / 2, # Rough estimation
avg_completion_tokens=avg_tokens / 2,
error_rate=error_rate * 100, # Convert to percentage
timeout_rate=timeout_rate * 100
)
Comprehensive metrics packaging includes timing, throughput, resource usage, and reliability metrics. Structured format enables consistent analysis and reporting.
Optimization Recommendations Generation¶
Intelligent recommendation system analyzes performance data to suggest specific optimization strategies for improved system performance.
def _generate_optimization_recommendations(self, summary: Dict[str, Any]) -> List[str]:
"""Generate actionable optimization recommendations based on benchmark results"""
recommendations = []
Recommendation system initialization prepares actionable optimization suggestions. Analysis-driven recommendations help prioritize performance improvements.
# Response time recommendations for latency optimization
avg_response_time = summary.get("avg_response_time", 0)
if avg_response_time > 5.0:
recommendations.append("Response times are high (>5s). Consider implementing caching or optimizing prompts.")
Response time analysis identifies latency issues and suggests caching strategies. High response times indicate need for prompt optimization or caching implementation.
p95_response_time = summary.get("p95_response_time", 0)
if p95_response_time > avg_response_time * 2:
recommendations.append("High response time variance detected. Investigate outliers and implement timeout handling.")
Variance analysis detects inconsistent performance patterns. High p95 variance suggests outlier investigation and timeout handling implementation.
# Throughput recommendations for capacity optimization
requests_per_second = summary.get("requests_per_second", 0)
if requests_per_second < 1.0:
recommendations.append("Low throughput detected. Consider implementing connection pooling or async processing.")
Throughput analysis identifies capacity limitations and suggests scaling strategies. Low throughput indicates need for connection pooling or async processing.
# Error rate recommendations for reliability improvement
error_rate = summary.get("error_rate", 0)
if error_rate > 5.0:
recommendations.append("High error rate (>5%). Implement better error handling and retry mechanisms.")
Error rate analysis identifies reliability issues and suggests resilience improvements. High error rates indicate need for better error handling and retry mechanisms.
# Memory recommendations for resource optimization
memory_usage = summary.get("memory_usage_mb", 0)
if memory_usage > 1000:
recommendations.append("High memory usage detected. Consider implementing memory optimization strategies.")
Memory usage analysis identifies resource consumption issues. High memory usage suggests need for optimization strategies or increased resource allocation.
# Token optimization for cost and efficiency
tokens_per_second = summary.get("tokens_per_second", 0)
if tokens_per_second < 10:
recommendations.append("Low token throughput. Consider optimizing prompt engineering or using faster models.")
Token throughput analysis identifies efficiency opportunities. Low token throughput suggests prompt engineering improvements or faster model selection.
# CPU recommendations for compute optimization
cpu_usage = summary.get("cpu_usage_percent", 0)
if cpu_usage > 80:
recommendations.append("High CPU usage detected. Consider scaling horizontally or optimizing compute-intensive operations.")
CPU usage analysis identifies compute bottlenecks. High CPU usage suggests horizontal scaling or compute optimization needs.
if not recommendations:
recommendations.append("Performance metrics are within acceptable ranges. Continue monitoring for trends.")
return recommendations
Default recommendation ensures actionable feedback even for well-performing systems. Continued monitoring recommendation maintains performance awareness.
Part 3: Continuous Monitoring Systems (15 minutes)¶
Production Performance Monitoring¶
Real-time performance monitoring provides continuous visibility into agent performance for production systems with automated alerting and analysis.
class ContinuousPerformanceMonitor:
"""Continuous performance monitoring for production systems with real-time analysis"""
def __init__(self, monitoring_config: Dict[str, Any]):
self.config = monitoring_config
self.metrics_buffer: List[PerformanceMetrics] = []
self.alert_thresholds = monitoring_config.get("alert_thresholds", {})
self.monitoring_active = False
self.logger = logging.getLogger(__name__)
Continuous monitoring initialization establishes configuration, metrics buffering, alert thresholds, and logging infrastructure for production monitoring.
async def start_monitoring(self, agent_instance):
"""Start continuous performance monitoring with multiple concurrent tasks"""
self.monitoring_active = True
self.logger.info("Starting continuous performance monitoring")
Monitoring activation begins comprehensive real-time analysis. Multiple concurrent tasks provide complete system visibility across different performance dimensions.
# Start monitoring tasks for comprehensive system analysis
monitoring_tasks = [
asyncio.create_task(self._monitor_response_times(agent_instance)),
asyncio.create_task(self._monitor_resource_usage()),
asyncio.create_task(self._monitor_error_rates(agent_instance)),
asyncio.create_task(self._process_metrics_buffer())
]
Concurrent monitoring tasks track response times, resource usage, error rates, and metrics processing. Parallel execution ensures comprehensive monitoring without performance impact.
try:
await asyncio.gather(*monitoring_tasks)
except Exception as e:
self.logger.error(f"Monitoring error: {str(e)}")
finally:
self.monitoring_active = False
Task coordination with exception handling ensures monitoring resilience. Cleanup ensures proper monitoring state management and resource release.
Response Time Monitoring Implementation¶
Continuous response time monitoring uses method injection to track agent performance without modifying agent code.
async def _monitor_response_times(self, agent_instance):
"""Monitor agent response times continuously with method injection"""
while self.monitoring_active:
try:
# Inject monitoring into agent calls using method wrapping
original_run = agent_instance.run
Method injection preserves original agent functionality while adding monitoring capabilities. Original method reference enables seamless monitoring integration.
def monitored_run(input_text):
start_time = time.time()
try:
result = original_run(input_text)
response_time = time.time() - start_time
# Record successful request metric
self._record_response_time_metric(response_time, True)
return result
Successful request monitoring captures timing and records metrics. Response time measurement includes complete request lifecycle for accurate performance analysis.
except Exception as e:
response_time = time.time() - start_time
self._record_response_time_metric(response_time, False)
raise e
Exception handling maintains timing accuracy even for failed requests. Error metrics provide complete performance picture including failure response times.
agent_instance.run = monitored_run
await asyncio.sleep(1) # Check every second
except Exception as e:
self.logger.error(f"Response time monitoring error: {str(e)}")
await asyncio.sleep(5)
Method replacement enables monitoring activation with minimal agent modification. Error handling ensures monitoring resilience with graceful degradation.
Resource Usage Monitoring¶
Continuous system resource monitoring tracks memory and CPU usage with configurable alerting thresholds for proactive system management.
async def _monitor_resource_usage(self):
"""Monitor system resource usage with threshold-based alerting"""
while self.monitoring_active:
try:
# Get current resource usage metrics
process = psutil.Process()
memory_mb = process.memory_info().rss / 1024 / 1024
cpu_percent = process.cpu_percent()
Resource monitoring captures real-time system utilization using psutil. Memory and CPU measurements provide essential operational visibility for capacity planning.
# Check against configurable thresholds
if memory_mb > self.alert_thresholds.get("memory_mb", 1000):
await self._send_alert("HIGH_MEMORY_USAGE", f"Memory usage: {memory_mb:.1f}MB")
if cpu_percent > self.alert_thresholds.get("cpu_percent", 80):
await self._send_alert("HIGH_CPU_USAGE", f"CPU usage: {cpu_percent:.1f}%")
Threshold-based alerting enables proactive response to resource constraints. Configurable thresholds adapt monitoring to different deployment environments and requirements.
# Record metrics for trend analysis
self._record_resource_metric(memory_mb, cpu_percent)
await asyncio.sleep(self.config.get("resource_check_interval", 10))
except Exception as e:
self.logger.error(f"Resource monitoring error: {str(e)}")
await asyncio.sleep(5)
Resource metric recording enables trend analysis and historical comparison. Configurable check intervals balance monitoring accuracy with system overhead.
Metrics Recording and Real-time Alerting¶
Performance metrics recording supports both batch analysis and real-time alerting for comprehensive monitoring coverage.
def _record_response_time_metric(self, response_time: float, success: bool):
"""Record response time metric with real-time alerting"""
# Add to metrics buffer for batch processing
metric = {
"type": "response_time",
"value": response_time,
"success": success,
"timestamp": datetime.now()
}
self.metrics_buffer.append(metric)
Metric buffering enables batch processing for statistical analysis. Structured metric format includes timing, success status, and timestamp for comprehensive analysis.
# Check real-time alerts for immediate response
if response_time > self.alert_thresholds.get("response_time_seconds", 10):
asyncio.create_task(self._send_alert(
"SLOW_RESPONSE",
f"Response time: {response_time:.2f}s"
))
Real-time alerting enables immediate response to performance degradation. Async alert dispatch prevents monitoring overhead from impacting system performance.
Alert System Implementation¶
Comprehensive alert system provides structured notifications with severity classification and integration with external monitoring platforms.
async def _send_alert(self, alert_type: str, message: str):
"""Send performance alert with severity classification"""
alert = {
"type": alert_type,
"message": message,
"timestamp": datetime.now(),
"severity": self._determine_alert_severity(alert_type)
}
self.logger.warning(f"Performance Alert [{alert_type}]: {message}")
Alert structure includes type classification, severity levels, and timestamps for comprehensive incident tracking. Structured format enables integration with monitoring systems.
# In production: send to monitoring system (Slack, PagerDuty, etc.)
await self._dispatch_alert_to_monitoring_system(alert)
External alert dispatch enables integration with enterprise monitoring platforms. Async dispatch prevents alert processing from blocking system performance.
Real-time Metrics Access¶
Real-time metrics provide immediate visibility into current system performance for operational dashboards and monitoring interfaces.
def get_real_time_metrics(self) -> Dict[str, Any]:
"""Get current real-time performance metrics for dashboards"""
if not self.metrics_buffer:
return {"status": "no_data"}
Metrics availability check prevents empty data responses. No-data status enables graceful handling of monitoring startup periods.
# Calculate metrics from recent buffer (last 5 minutes)
recent_metrics = [m for m in self.metrics_buffer if
(datetime.now() - m["timestamp"]).total_seconds() < 300]
if not recent_metrics:
return {"status": "no_recent_data"}
Recent metrics filtering provides relevant performance data for current system state. Five-minute window balances real-time accuracy with statistical significance.
response_times = [m["value"] for m in recent_metrics if m["type"] == "response_time"]
if response_times:
return {
"avg_response_time": statistics.mean(response_times),
"max_response_time": max(response_times),
"min_response_time": min(response_times),
"total_requests": len(response_times),
"success_rate": sum(1 for m in recent_metrics if m.get("success")) / len(recent_metrics) * 100,
"timestamp": datetime.now()
}
return {"status": "no_response_data"}
Comprehensive metrics calculation provides statistical summary of recent performance. Average, max, min response times, request counts, and success rates give complete operational visibility.
Part 2: Advanced Monitoring & Observability (20 minutes)¶
Comprehensive Monitoring Architecture¶
🗂️ File: src/session2/monitoring_observability.py - Production monitoring systems
Production monitoring architecture integrates Prometheus metrics, OpenTelemetry tracing, and structured logging for comprehensive observability.
import prometheus_client
from prometheus_client import Counter, Histogram, Gauge, Summary
import structlog
from opentelemetry import trace, metrics
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from typing import Dict, Any, Optional
import json
from datetime import datetime
import asyncio
Enterprise monitoring imports provide comprehensive observability tools. Prometheus for metrics, OpenTelemetry for tracing, and structured logging enable complete system visibility.
Prometheus Metrics Implementation¶
Prometheus metrics provide time-series data collection for operational monitoring and alerting in production LangChain applications.
class LangChainPrometheusMetrics:
"""Prometheus metrics for LangChain applications with comprehensive coverage"""
def __init__(self, service_name: str):
self.service_name = service_name
Metrics initialization establishes service identification for multi-service monitoring environments. Service name enables proper metric attribution and aggregation.
# Request metrics for operational monitoring
self.request_count = Counter(
'langchain_requests_total',
'Total number of requests to LangChain agents',
['agent_type', 'status']
)
self.request_duration = Histogram(
'langchain_request_duration_seconds',
'Time spent processing LangChain requests',
['agent_type', 'operation'],
buckets=[0.1, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0]
)
Request metrics track volume and timing for capacity planning. Counter tracks total requests by type and status while histogram captures latency distribution with appropriate buckets.
# LLM metrics for cost and performance analysis
self.llm_tokens = Counter(
'langchain_llm_tokens_total',
'Total tokens processed by LLM',
['model', 'token_type'] # token_type: prompt, completion
)
self.llm_api_calls = Counter(
'langchain_llm_api_calls_total',
'Total API calls to LLM providers',
['provider', 'model', 'status']
)
LLM metrics enable cost optimization and performance analysis. Token tracking supports cost estimation while API call metrics monitor provider reliability and usage patterns.
self.llm_latency = Histogram(
'langchain_llm_latency_seconds',
'LLM API call latency',
['provider', 'model'],
buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]
)
LLM latency metrics track API response times by provider and model. Fine-grained buckets enable SLA monitoring and performance comparison across different LLM providers.
# Tool metrics for execution monitoring
self.tool_executions = Counter(
'langchain_tool_executions_total',
'Total tool executions',
['tool_name', 'status']
)
self.tool_duration = Histogram(
'langchain_tool_duration_seconds',
'Tool execution duration',
['tool_name'],
buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 5.0, 10.0]
)
Tool metrics monitor custom tool performance and reliability. Execution counts and duration histograms help identify bottlenecks and optimize tool implementations.
# Memory and caching metrics for resource optimization
self.cache_operations = Counter(
'langchain_cache_operations_total',
'Cache operations',
['operation', 'result'] # operation: get, set; result: hit, miss
)
self.memory_usage = Gauge(
'langchain_memory_usage_bytes',
'Current memory usage',
['component']
)
Resource metrics track cache effectiveness and memory utilization. Cache hit/miss ratios guide cache optimization while memory gauges enable capacity planning.
# Error metrics for reliability monitoring
self.error_count = Counter(
'langchain_errors_total',
'Total errors in LangChain application',
['error_type', 'component']
)
# Quality metrics for response assessment
self.response_quality = Histogram(
'langchain_response_quality_score',
'Response quality scores',
['agent_type', 'metric_type'],
buckets=[0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]
)
Error and quality metrics provide comprehensive reliability and performance assessment. Error categorization helps identify failure patterns while quality scores enable response evaluation.
Metrics Recording Methods¶
Dedicated recording methods provide convenient interfaces for capturing different types of performance data throughout the application lifecycle.
def record_request(self, agent_type: str, status: str, duration: float, operation: str = "run"):
"""Record request metrics with type and operation context"""
self.request_count.labels(agent_type=agent_type, status=status).inc()
self.request_duration.labels(agent_type=agent_type, operation=operation).observe(duration)
Request recording captures both volume and timing metrics. Labels enable filtering and aggregation by agent type, status, and operation for detailed analysis.
def record_llm_call(self, provider: str, model: str, status: str, latency: float,
prompt_tokens: int, completion_tokens: int):
"""Record comprehensive LLM metrics for cost and performance analysis"""
self.llm_api_calls.labels(provider=provider, model=model, status=status).inc()
self.llm_latency.labels(provider=provider, model=model).observe(latency)
self.llm_tokens.labels(model=model, token_type="prompt").inc(prompt_tokens)
self.llm_tokens.labels(model=model, token_type="completion").inc(completion_tokens)
LLM recording captures comprehensive metrics for cost optimization. Provider, model, and token type labels enable detailed analysis of API usage patterns and costs.
def record_tool_execution(self, tool_name: str, status: str, duration: float):
"""Record tool execution metrics for performance monitoring"""
self.tool_executions.labels(tool_name=tool_name, status=status).inc()
self.tool_duration.labels(tool_name=tool_name).observe(duration)
def record_cache_operation(self, operation: str, result: str):
"""Record cache operation metrics for optimization analysis"""
self.cache_operations.labels(operation=operation, result=result).inc()
Tool and cache metrics enable performance optimization. Tool execution tracking identifies bottlenecks while cache metrics guide optimization strategies.
def record_error(self, error_type: str, component: str):
"""Record error metrics for reliability monitoring"""
self.error_count.labels(error_type=error_type, component=component).inc()
def update_memory_usage(self, component: str, bytes_used: float):
"""Update memory usage metrics for resource monitoring"""
self.memory_usage.labels(component=component).set(bytes_used)
def record_quality_score(self, agent_type: str, metric_type: str, score: float):
"""Record response quality metrics for performance assessment"""
self.response_quality.labels(agent_type=agent_type, metric_type=metric_type).observe(score)
Error, memory, and quality recording provide comprehensive system health monitoring. Categorized tracking enables targeted optimization and reliability improvements.
Structured Logging Implementation¶
Structured logging provides machine-readable log format with consistent metadata for comprehensive application observability and troubleshooting.
class StructuredLogging:
"""Structured logging for LangChain applications with comprehensive context"""
def __init__(self, service_name: str, log_level: str = "INFO"):
self.service_name = service_name
Structured logging initialization establishes service context and configuration. Service name provides consistent identification across distributed logging systems.
# Configure structured logging with processors
structlog.configure(
processors=[
structlog.stdlib.filter_by_level,
structlog.stdlib.add_logger_name,
structlog.stdlib.add_log_level,
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.TimeStamper(fmt="iso"),
Processor configuration creates consistent log structure. Level filtering, logger names, timestamps, and formatting enable structured analysis and aggregation.
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.UnicodeDecoder(),
structlog.processors.JSONRenderer()
],
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
self.logger = structlog.get_logger(service_name)
Advanced processors handle stack traces, exception formatting, unicode encoding, and JSON rendering. Factory and wrapper configuration integrates with standard Python logging.
Structured Logging Methods¶
Specialized logging methods capture different types of events with appropriate structured context for comprehensive traceability.
def log_request(self, request_id: str, agent_type: str, input_text: str,
response: str, duration: float, status: str, **kwargs):
"""Log request with comprehensive structured context"""
self.logger.info(
"agent_request_completed",
request_id=request_id,
agent_type=agent_type,
input_length=len(input_text),
response_length=len(response) if response else 0,
duration_seconds=duration,
status=status,
service=self.service_name,
**kwargs
)
Request logging captures complete request lifecycle with structured metadata. Request ID enables tracing while metrics provide performance context.
def log_llm_call(self, request_id: str, provider: str, model: str,
prompt_tokens: int, completion_tokens: int, latency: float, **kwargs):
"""Log LLM call with cost and performance context"""
self.logger.info(
"llm_api_call",
request_id=request_id,
provider=provider,
model=model,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens,
latency_seconds=latency,
cost_estimate=self._calculate_cost_estimate(model, prompt_tokens, completion_tokens),
service=self.service_name,
**kwargs
)
LLM call logging includes cost estimation and performance metrics. Token tracking and cost calculation enable operational cost analysis and optimization.
Tool Execution and Error Logging¶
Tool execution logging captures comprehensive execution data including input parameters, result metadata, and performance timings. This enables detailed analysis of tool performance and resource utilization patterns.
def log_tool_execution(self, request_id: str, tool_name: str,
input_params: Dict[str, Any], result: Any,
duration: float, status: str, **kwargs):
"""Log tool execution with structured data"""
self.logger.info(
"tool_execution",
request_id=request_id,
tool_name=tool_name,
input_params=input_params,
result_type=type(result).__name__,
result_size=len(str(result)) if result else 0,
duration_seconds=duration,
status=status,
service=self.service_name,
**kwargs
)
Error logging provides structured error tracking with component identification and stack trace information. This enables rapid debugging and issue resolution in production environments.
def log_error(self, request_id: str, error_type: str, error_message: str,
component: str, stack_trace: str = None, **kwargs):
"""Log error with structured data"""
self.logger.error(
"application_error",
request_id=request_id,
error_type=error_type,
error_message=error_message,
component=component,
stack_trace=stack_trace,
service=self.service_name,
**kwargs
)
Performance Alert and Cost Calculation¶
Performance alert logging enables proactive monitoring by capturing threshold violations and performance degradations. Alert data includes metric values and thresholds for analysis.
def log_performance_alert(self, alert_type: str, metric_name: str,
current_value: float, threshold: float, **kwargs):
"""Log performance alert"""
self.logger.warning(
"performance_alert",
alert_type=alert_type,
metric_name=metric_name,
current_value=current_value,
threshold=threshold,
service=self.service_name,
**kwargs
)
Cost estimation provides real-time cost tracking for LLM API usage. Token-based pricing models enable accurate cost projection and budget management across different LLM providers.
def _calculate_cost_estimate(self, model: str, prompt_tokens: int, completion_tokens: int) -> float:
"""Calculate estimated cost for LLM call"""
# Simplified cost calculation - update with actual pricing
cost_per_1k_tokens = {
"gpt-4": 0.03,
"gpt-3.5-turbo": 0.002,
"claude-3-sonnet": 0.015,
"claude-3-haiku": 0.0025
}
rate = cost_per_1k_tokens.get(model, 0.01) # Default rate
total_tokens = prompt_tokens + completion_tokens
return (total_tokens / 1000) * rate
Distributed Tracing System¶
Distributed tracing enables end-to-end observability across complex LangChain applications. This system provides request correlation, performance bottleneck identification, and dependency mapping.
class DistributedTracing:
"""Distributed tracing for LangChain applications"""
def __init__(self, service_name: str, jaeger_endpoint: str = None):
self.service_name = service_name
# Initialize tracing
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)
# Configure Jaeger exporter if endpoint provided
if jaeger_endpoint:
jaeger_exporter = JaegerExporter(
agent_host_name="localhost",
agent_port=6831,
)
span_processor = BatchSpanProcessor(jaeger_exporter)
trace.get_tracer_provider().add_span_processor(span_processor)
self.tracer = tracer
Request and Component Tracing¶
Agent request tracing creates root spans for entire workflow execution. Request correlation enables tracking complex multi-agent interactions and identifying performance bottlenecks.
def trace_agent_request(self, request_id: str, agent_type: str):
"""Create trace span for agent request"""
return self.tracer.start_span(
f"agent_request:{agent_type}",
attributes={
"service.name": self.service_name,
"request.id": request_id,
"agent.type": agent_type
}
)
LLM call tracing captures provider-specific performance characteristics. Model and provider tracking enables optimization decisions and cost analysis across different LLM services.
def trace_llm_call(self, request_id: str, provider: str, model: str):
"""Create trace span for LLM call"""
return self.tracer.start_span(
f"llm_call:{provider}",
attributes={
"service.name": self.service_name,
"request.id": request_id,
"llm.provider": provider,
"llm.model": model
}
)
Tool execution tracing monitors individual tool performance and resource usage. Tool-specific spans enable detailed analysis of workflow execution patterns and optimization opportunities.
def trace_tool_execution(self, request_id: str, tool_name: str):
"""Create trace span for tool execution"""
return self.tracer.start_span(
f"tool_execution:{tool_name}",
attributes={
"service.name": self.service_name,
"request.id": request_id,
"tool.name": tool_name
}
)
Real-time Monitoring Dashboard¶
The monitoring dashboard aggregates metrics from multiple sources to provide comprehensive system visibility. Dashboard data includes health indicators, performance metrics, and operational insights.
class MonitoringDashboard:
"""Real-time monitoring dashboard for LangChain applications"""
def __init__(self, metrics: LangChainPrometheusMetrics,
logging: StructuredLogging):
self.metrics = metrics
self.logging = logging
self.dashboard_data = {}
Comprehensive Dashboard Data Generation¶
Dashboard data generation collects metrics from all monitoring systems and aggregates them into a unified view. This enables real-time system health assessment and performance analysis.
async def generate_dashboard_data(self) -> Dict[str, Any]:
"""Generate comprehensive dashboard data"""
current_time = datetime.now()
# Collect metrics from Prometheus
dashboard_data = {
"timestamp": current_time.isoformat(),
"service_health": await self._get_service_health(),
"request_metrics": await self._get_request_metrics(),
"llm_metrics": await self._get_llm_metrics(),
"tool_metrics": await self._get_tool_metrics(),
"performance_metrics": await self._get_performance_metrics(),
"error_metrics": await self._get_error_metrics(),
"resource_usage": await self._get_resource_usage(),
"alerts": await self._get_active_alerts()
}
return dashboard_data
Service Health Assessment¶
Service health assessment analyzes error rates and response times to determine overall system status. Health categorization enables rapid identification of service degradation and system issues.
async def _get_service_health(self) -> Dict[str, Any]:
"""Get overall service health status"""
# Calculate health based on error rates and response times
recent_errors = 0 # Would query from Prometheus
recent_requests = 100 # Would query from Prometheus
avg_response_time = 1.5 # Would query from Prometheus
error_rate = recent_errors / recent_requests if recent_requests > 0 else 0
if error_rate > 0.1 or avg_response_time > 10:
health_status = "unhealthy"
elif error_rate > 0.05 or avg_response_time > 5:
health_status = "degraded"
else:
health_status = "healthy"
return {
"status": health_status,
"error_rate": error_rate,
"avg_response_time": avg_response_time,
"uptime_percentage": 99.5 # Would calculate from actual data
}
Request and LLM Metrics Collection¶
Request metrics provide throughput and performance indicators for system capacity planning. Response time percentiles and success rates enable SLA monitoring and performance optimization.
async def _get_request_metrics(self) -> Dict[str, Any]:
"""Get request-related metrics"""
return {
"requests_per_minute": 45, # Would query from Prometheus
"avg_response_time": 2.1,
"p95_response_time": 4.5,
"success_rate": 98.2,
"concurrent_requests": 12
}
LLM metrics track token usage, API latency, and cost performance across different models. Model distribution analysis enables optimization decisions and cost management strategies.
async def _get_llm_metrics(self) -> Dict[str, Any]:
"""Get LLM-specific metrics"""
return {
"total_tokens_processed": 125000, # Would query from Prometheus
"tokens_per_minute": 2500,
"avg_llm_latency": 1.8,
"api_calls_per_minute": 50,
"cost_per_hour": 12.50,
"model_distribution": {
"gpt-4": 60,
"gpt-3.5-turbo": 40
}
}
async def _get_performance_metrics(self) -> Dict[str, Any]:
"""Get performance metrics"""
return {
"cpu_usage_percent": 45.2,
"memory_usage_mb": 512.8,
"cache_hit_rate": 78.5,
"throughput_requests_per_second": 0.75
}
Part 3: Cost Optimization & Resource Management (10 minutes)¶
Intelligent Cost Management¶
🗂️ File: src/session2/cost_optimization.py - Cost optimization strategies
Cost optimization requires comprehensive tracking and intelligent management of LLM and resource costs. This system provides real-time cost monitoring, budget management, and optimization recommendations.
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta
import asyncio
import json
Cost Metrics Data Structure¶
Cost metrics capture comprehensive cost data across multiple dimensions including token usage, API calls, resource consumption, and optimization savings. This enables detailed cost analysis and optimization.
@dataclass
class CostMetrics:
"""Cost tracking metrics for LangChain applications"""
# Token costs
total_prompt_tokens: int = 0
total_completion_tokens: int = 0
total_token_cost: float = 0.0
# API call costs
total_api_calls: int = 0
total_api_cost: float = 0.0
# Resource costs
compute_cost_per_hour: float = 0.0
storage_cost_per_hour: float = 0.0
CostMetrics dataclass captures token usage and API costs for comprehensive expense tracking. Token costs include both prompt and completion charges while API costs track overall usage patterns.
# Time period
period_start: datetime = None
period_end: datetime = None
# Cost breakdown by model
cost_by_model: Dict[str, float] = None
# Optimization metrics
cache_savings: float = 0.0
optimization_savings: float = 0.0
Cost Optimization Manager¶
The cost optimization manager provides intelligent cost tracking, budget management, and optimization recommendations. It monitors spending patterns and suggests cost-saving strategies.
class CostOptimizationManager:
"""Intelligent cost optimization for LangChain applications"""
def __init__(self, cost_config: Dict[str, Any]):
self.cost_config = cost_config
self.cost_tracking = {}
self.optimization_rules = []
self.cost_alerts = []
# Initialize cost rates
self.token_costs = cost_config.get("token_costs", {
"gpt-4": {"prompt": 0.03, "completion": 0.06},
"gpt-3.5-turbo": {"prompt": 0.0015, "completion": 0.002},
"claude-3-sonnet": {"prompt": 0.015, "completion": 0.015}
})
self.daily_budget = cost_config.get("daily_budget", 100.0)
self.monthly_budget = cost_config.get("monthly_budget", 3000.0)
LLM Cost Tracking and Budget Management¶
LLM cost tracking provides real-time cost calculation and budget monitoring. Cost accumulation by model enables detailed analysis and optimization opportunities.
def track_llm_cost(self, model: str, prompt_tokens: int, completion_tokens: int,
provider: str = "openai") -> float:
"""Track cost for LLM usage"""
if model not in self.token_costs:
model = "gpt-3.5-turbo" # Default fallback
rates = self.token_costs[model]
prompt_cost = (prompt_tokens / 1000) * rates["prompt"]
completion_cost = (completion_tokens / 1000) * rates["completion"]
total_cost = prompt_cost + completion_cost
Daily cost tracking organizes expenses by date for budget management and trend analysis. Model-specific cost breakdown enables optimization decisions and provider comparisons.
# Track in daily/monthly buckets
today = datetime.now().date()
if today not in self.cost_tracking:
self.cost_tracking[today] = CostMetrics(
period_start=datetime.combine(today, datetime.min.time()),
cost_by_model={}
)
daily_metrics = self.cost_tracking[today]
daily_metrics.total_prompt_tokens += prompt_tokens
daily_metrics.total_completion_tokens += completion_tokens
daily_metrics.total_token_cost += total_cost
daily_metrics.total_api_calls += 1
# Track by model
if model not in daily_metrics.cost_by_model:
daily_metrics.cost_by_model[model] = 0.0
daily_metrics.cost_by_model[model] += total_cost
# Check budget alerts
self._check_budget_alerts(daily_metrics, total_cost)
return total_cost
Budget Alert System¶
Budget alert checking monitors daily and monthly spending against configured thresholds. Proactive alerts enable cost control and prevent budget overruns.
def _check_budget_alerts(self, daily_metrics: CostMetrics, new_cost: float):
"""Check if budget thresholds are exceeded"""
daily_total = daily_metrics.total_token_cost
# Daily budget alerts
if daily_total > self.daily_budget * 0.8: # 80% threshold
self._create_budget_alert("DAILY_BUDGET_WARNING", daily_total, self.daily_budget)
if daily_total > self.daily_budget:
self._create_budget_alert("DAILY_BUDGET_EXCEEDED", daily_total, self.daily_budget)
Monthly budget tracking aggregates costs across the current month and compares against monthly limits. Early warning thresholds enable proactive budget management.
# Monthly budget calculation
month_start = datetime.now().replace(day=1).date()
monthly_total = sum(
metrics.total_token_cost
for date, metrics in self.cost_tracking.items()
if date >= month_start
)
if monthly_total > self.monthly_budget * 0.8:
self._create_budget_alert("MONTHLY_BUDGET_WARNING", monthly_total, self.monthly_budget)
Intelligent Model Selection¶
Model selection optimization chooses the most cost-effective model that meets quality requirements. Multi-criteria optimization balances cost, speed, and quality factors.
def optimize_model_selection(self, task_complexity: str, required_quality: float) -> str:
"""Suggest optimal model based on task requirements and cost"""
model_capabilities = {
"gpt-4": {"quality": 0.95, "cost_per_1k": 0.045, "speed": 0.7},
"gpt-3.5-turbo": {"quality": 0.85, "cost_per_1k": 0.002, "speed": 0.9},
"claude-3-sonnet": {"quality": 0.90, "cost_per_1k": 0.015, "speed": 0.8}
}
# Filter models that meet quality requirements
suitable_models = {
model: specs for model, specs in model_capabilities.items()
if specs["quality"] >= required_quality
}
if not suitable_models:
return "gpt-4" # Fallback to highest quality
Task complexity weighting adjusts optimization priorities based on task requirements. Simple tasks prioritize cost savings while complex tasks emphasize quality and capability.
# Simple task complexity scoring
complexity_weights = {
"simple": {"cost": 0.7, "speed": 0.3, "quality": 0.0},
"medium": {"cost": 0.5, "speed": 0.2, "quality": 0.3},
"complex": {"cost": 0.2, "speed": 0.1, "quality": 0.7}
}
weights = complexity_weights.get(task_complexity, complexity_weights["medium"])
Weighted scoring combines normalized metrics to find the optimal model. Cost normalization ensures fair comparison while quality and speed factors influence final selection.
# Calculate weighted scores
best_model = None
best_score = -1
for model, specs in suitable_models.items():
# Normalize metrics (lower cost is better, higher speed/quality is better)
cost_score = 1 - (specs["cost_per_1k"] / max(s["cost_per_1k"] for s in suitable_models.values()))
speed_score = specs["speed"]
quality_score = specs["quality"]
total_score = (
cost_score * weights["cost"] +
speed_score * weights["speed"] +
quality_score * weights["quality"]
)
if total_score > best_score:
best_score = total_score
best_model = model
return best_model or "gpt-3.5-turbo"
Cost Optimization Recommendations¶
Recommendation generation analyzes cost patterns and suggests specific optimization strategies. Prioritized recommendations enable focused cost reduction efforts with quantified savings potential.
def get_cost_optimization_recommendations(self) -> List[Dict[str, Any]]:
"""Generate cost optimization recommendations"""
recommendations = []
# Analyze recent cost patterns
recent_costs = self._get_recent_cost_analysis()
Model optimization recommendations identify opportunities to use more cost-effective models. High GPT-4 usage triggers suggestions for task complexity analysis and model selection optimization.
# High-cost model recommendations
if recent_costs.get("gpt-4_percentage", 0) > 60:
recommendations.append({
"type": "model_optimization",
"priority": "high",
"title": "Reduce GPT-4 Usage",
"description": "Consider using GPT-3.5-turbo for simpler tasks",
"potential_savings": recent_costs.get("gpt-4_cost", 0) * 0.3,
"implementation": "Implement task complexity analysis"
})
Prompt optimization recommendations address excessive token usage through length reduction and compression strategies. Token efficiency improvements directly impact operational costs.
# Token optimization recommendations
avg_tokens = recent_costs.get("avg_tokens_per_request", 0)
if avg_tokens > 2000:
recommendations.append({
"type": "prompt_optimization",
"priority": "medium",
"title": "Optimize Prompt Length",
"description": "Reduce average prompt length to decrease token costs",
"potential_savings": recent_costs.get("daily_cost", 0) * 0.15,
"implementation": "Implement prompt compression and caching"
})
Caching optimization recommendations focus on reducing redundant API calls through improved caching strategies. Higher cache hit rates directly translate to cost savings and performance improvements.
# Caching recommendations
cache_hit_rate = recent_costs.get("cache_hit_rate", 0)
if cache_hit_rate < 30:
recommendations.append({
"type": "caching_optimization",
"priority": "high",
"title": "Improve Caching Strategy",
"description": "Implement better caching to reduce redundant API calls",
"potential_savings": recent_costs.get("daily_cost", 0) * 0.25,
"implementation": "Add semantic caching and increase cache TTL"
})
return recommendations
Recent Cost Analysis¶
Cost pattern analysis aggregates recent spending data to identify trends and optimization opportunities. Seven-day rolling analysis provides balanced insights for recommendation generation.
def _get_recent_cost_analysis(self) -> Dict[str, Any]:
"""Analyze recent cost patterns"""
# Get last 7 days of data
end_date = datetime.now().date()
start_date = end_date - timedelta(days=7)
total_cost = 0
total_tokens = 0
total_calls = 0
model_costs = {}
Recent cost analysis preparation establishes seven-day analysis window. Cost aggregation variables track total expenses, token usage, API calls, and model-specific spending patterns.
for date in [start_date + timedelta(days=x) for x in range(8)]:
if date in self.cost_tracking:
metrics = self.cost_tracking[date]
total_cost += metrics.total_token_cost
total_tokens += metrics.total_prompt_tokens + metrics.total_completion_tokens
total_calls += metrics.total_api_calls
for model, cost in (metrics.cost_by_model or {}).items():
model_costs[model] = model_costs.get(model, 0) + cost
Cost aggregation loops through tracked dates and accumulates metrics. Total cost, token usage, and model-specific expenses enable comprehensive pattern analysis for optimization recommendations.
return {
"daily_cost": total_cost / 7,
"avg_tokens_per_request": total_tokens / total_calls if total_calls > 0 else 0,
"gpt-4_cost": model_costs.get("gpt-4", 0),
"gpt-4_percentage": (model_costs.get("gpt-4", 0) / total_cost * 100) if total_cost > 0 else 0,
"cache_hit_rate": 25 # Would get from actual metrics
}
Cost Dashboard Data Generation¶
Cost dashboard generation provides comprehensive cost visibility including current spending, budget utilization, usage breakdowns, and optimization metrics. Monthly projection calculations enable proactive budget management.
def get_cost_dashboard_data(self) -> Dict[str, Any]:
"""Generate cost dashboard data"""
today = datetime.now().date()
month_start = today.replace(day=1)
# Daily costs
daily_metrics = self.cost_tracking.get(today, CostMetrics())
# Monthly costs
monthly_cost = sum(
metrics.total_token_cost
for date, metrics in self.cost_tracking.items()
if date >= month_start
)
# Cost projections
days_in_month = (today.replace(month=today.month + 1) - month_start).days
days_elapsed = (today - month_start).days + 1
projected_monthly = (monthly_cost / days_elapsed) * days_in_month if days_elapsed > 0 else 0
Dashboard data structure organizes cost information into logical categories for presentation. Budget utilization percentages enable quick assessment of spending patterns and budget health.
return {
"current_costs": {
"daily": daily_metrics.total_token_cost,
"monthly": monthly_cost,
"projected_monthly": projected_monthly
},
"budget_status": {
"daily_budget": self.daily_budget,
"monthly_budget": self.monthly_budget,
"daily_utilization": (daily_metrics.total_token_cost / self.daily_budget * 100),
"monthly_utilization": (monthly_cost / self.monthly_budget * 100)
},
"usage_breakdown": {
"total_tokens": daily_metrics.total_prompt_tokens + daily_metrics.total_completion_tokens,
"api_calls": daily_metrics.total_api_calls,
"cost_by_model": daily_metrics.cost_by_model or {}
},
"optimization_metrics": {
"cache_savings": daily_metrics.cache_savings,
"total_savings": daily_metrics.optimization_savings
}
}
📝 Multiple Choice Test - Module D¶
Test your understanding of performance monitoring and optimization:
Question 1: What categories of metrics are tracked in the PerformanceMetrics dataclass?
A) Only response times
B) Response time, resource usage, LLM-specific, tool usage, error, quality, and operational metrics
C) Just error rates and memory usage
D) Only API latency measurements
Question 2: What does the benchmark suite initialization include?
A) Only basic configuration
B) Agent factory, configuration, performance history, logger, and memory tracing
C) Just test scenarios
D) Only result storage
Question 3: How are benchmark results organized for analysis?
A) Single flat structure
B) By scenario with unique benchmark IDs, timestamps, and recommendations
C) Only by execution time
D) Just error logs
Question 4: What metrics are particularly important for cost optimization?
A) Only response times
B) Token processing rates, API latency, and resource consumption
C) Just error rates
D) Only cache hit rates
Question 5: What is the purpose of tracking percentile metrics (p50, p90) rather than just averages?
A) Reduce storage requirements
B) Understand user experience under different load conditions and identify outliers
C) Simplify calculations
D) Improve execution speed
Module Summary¶
You've now mastered performance optimization and monitoring for production LangChain systems:
✅ Performance Benchmarking & Optimization: Implemented comprehensive performance measurement and optimization strategies
✅ Advanced Monitoring & Observability: Set up Prometheus metrics, structured logging, and distributed tracing
✅ Cost Optimization & Resource Management: Created intelligent cost tracking and optimization systems
✅ Real-time Dashboards: Built monitoring dashboards with alerting and performance analytics
Next Steps¶
- Return to Core: Session 2 Main
- Advance to Session 3: LangGraph Multi-Agent Workflows
- Review Production: Module B: Production Deployment Strategies
🗂️ Source Files for Module D: - src/session2/performance_benchmarking.py - Performance measurement and optimization - src/session2/monitoring_observability.py - Comprehensive monitoring systems - src/session2/cost_optimization.py - Cost tracking and optimization strategies