LangChain日志记录与分析指南

发布时间: 2025-03-29 19:29

↑

# LangChain日志记录与分析指南

本文将介绍如何在LangChain项目中实现有效的日志记录和分析，帮助你更好地追踪问题和优化性能。

## 日志配置

### 1. 基础日志设置

```python
import logging
from langchain.callbacks import FileCallbackHandler

# 配置基础日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# 创建文件处理器
file_handler = FileCallbackHandler("langchain.log")
```

### 2. 自定义回调处理器

```python
from langchain.callbacks.base import BaseCallbackHandler
from typing import Dict, Any, List

class CustomLogger(BaseCallbackHandler):
    def __init__(self):
        self.chain_starts = 0
        self.llm_starts = 0
    
    def on_chain_start(
        self,
        serialized: Dict[str, Any],
        inputs: Dict[str, Any],
        **kwargs: Any
    ) -> None:
        """记录链启动信息"""
        self.chain_starts += 1
        logging.info(f"Chain开始执行: {serialized['name']}")
        logging.info(f"输入参数: {inputs}")
    
    def on_llm_start(
        self,
        serialized: Dict[str, Any],
        prompts: List[str],
        **kwargs: Any
    ) -> None:
        """记录LLM调用信息"""
        self.llm_starts += 1
        logging.info(f"LLM开始调用: {serialized['name']}")
        logging.info(f"提示词: {prompts}")
```

## 性能监控

### 1. Token使用统计

```python
from langchain.callbacks import get_openai_callback

def track_token_usage(chain, inputs):
    with get_openai_callback() as cb:
        result = chain.run(inputs)
        print(f"总Token数: {cb.total_tokens}")
        print(f"提示Token数: {cb.prompt_tokens}")
        print(f"补全Token数: {cb.completion_tokens}")
        print(f"总花费: ${cb.total_cost}")
    return result
```

### 2. 执行时间分析

```python
import time
from functools import wraps

def timing_decorator(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        duration = end_time - start_time
        logging.info(f"{func.__name__}执行时间: {duration:.2f}秒")
        return result
    return wrapper

# 使用装饰器
@timing_decorator
def run_chain(chain, inputs):
    return chain.run(inputs)
```

## 日志分析

### 1. 日志聚合

```python
from collections import defaultdict
import json

class LogAnalyzer:
    def __init__(self, log_file):
        self.log_file = log_file
        self.chain_stats = defaultdict(int)
        self.llm_stats = defaultdict(int)
    
    def analyze(self):
        with open(self.log_file, 'r') as f:
            for line in f:
                try:
                    record = json.loads(line)
                    if 'chain_name' in record:
                        self.chain_stats[record['chain_name']] += 1
                    if 'llm_name' in record:
                        self.llm_stats[record['llm_name']] += 1
                except:
                    continue
        
        return {
            'chain_stats': dict(self.chain_stats),
            'llm_stats': dict(self.llm_stats)
        }
```

### 2. 性能报告生成

```python
import pandas as pd
import matplotlib.pyplot as plt

def generate_performance_report(log_data):
    # 创建性能数据框
    df = pd.DataFrame(log_data)
    
    # 生成统计图表
    plt.figure(figsize=(10, 6))
    df.plot(kind='bar')
    plt.title('LangChain性能分析')
    plt.xlabel('组件')
    plt.ylabel('调用次数')
    plt.savefig('performance_report.png')
    
    # 生成统计报告
    report = {
        'summary': {
            'total_calls': df.sum().sum(),
            'avg_response_time': df['duration'].mean(),
            'error_rate': len(df[df['status'] == 'error']) / len(df)
        },
        'recommendations': get_optimization_suggestions(df)
    }
    
    return report
```

## 最佳实践

### 1. 结构化日志记录

```python
from typing import Optional

def log_chain_execution(
    chain_name: str,
    inputs: dict,
    outputs: Optional[dict] = None,
    error: Optional[Exception] = None
) -> None:
    log_entry = {
        'timestamp': time.time(),
        'chain_name': chain_name,
        'inputs': inputs,
        'outputs': outputs,
        'status': 'success' if error is None else 'error',
        'error': str(error) if error else None
    }
    
    logging.info(json.dumps(log_entry))
```

### 2. 错误追踪

```python
from traceback import format_exception

class ErrorTracker:
    def __init__(self):
        self.errors = []
    
    def track_error(self, error: Exception, context: dict):
        error_info = {
            'timestamp': time.time(),
            'error_type': type(error).__name__,
            'error_message': str(error),
            'traceback': format_exception(error),
            'context': context
        }
        
        self.errors.append(error_info)
        logging.error(json.dumps(error_info))
    
    def get_error_summary(self):
        return {
            'total_errors': len(self.errors),
            'error_types': self._count_error_types(),
            'recent_errors': self.errors[-5:]
        }
    
    def _count_error_types(self):
        counts = defaultdict(int)
        for error in self.errors:
            counts[error['error_type']] += 1
        return dict(counts)
```

## 应用示例

### 1. 完整日志系统

```python
class LangChainLogger:
    def __init__(self):
        self.file_handler = FileCallbackHandler("langchain.log")
        self.custom_logger = CustomLogger()
        self.error_tracker = ErrorTracker()
    
    def setup_chain(self, chain):
        # 添加回调处理器
        chain.callbacks = [
            self.file_handler,
            self.custom_logger
        ]
        return chain
    
    @timing_decorator
    def run_with_logging(self, chain, inputs):
        try:
            with get_openai_callback() as cb:
                result = chain.run(inputs)
                self._log_success(chain, inputs, result, cb)
                return result
        except Exception as e:
            self.error_tracker.track_error(e, {
                'chain': chain.__class__.__name__,
                'inputs': inputs
            })
            raise
    
    def _log_success(self, chain, inputs, outputs, callback):
        log_chain_execution(
            chain_name=chain.__class__.__name__,
            inputs=inputs,
            outputs=outputs,
            token_usage={
                'total': callback.total_tokens,
                'prompt': callback.prompt_tokens,
                'completion': callback.completion_tokens,
                'cost': callback.total_cost
            }
        )
```

### 2. 使用示例

```python
# 创建日志系统
logger = LangChainLogger()

# 配置链
chain = logger.setup_chain(my_chain)

# 运行并记录
try:
    result = logger.run_with_logging(chain, {
        'input': '你好，请帮我总结这段文字...'
    })
    print(result)
except Exception as e:
    print(f"执行出错: {e}")
    print(logger.error_tracker.get_error_summary())
```

## 总结

有效的日志记录和分析对于LangChain应用至关重要：
- 帮助追踪和调试问题
- 监控系统性能
- 优化资源使用
- 提供运行时洞察

通过本文介绍的方法，你可以建立一个完整的日志系统，更好地管理和优化你的LangChain应用。

元素码农