log-sanitization¶
第2章 · 上下文工程 · 配套项目
chapter2/log-sanitization
项目说明¶
Log Sanitization / 日志脱敏¶
本实验演示如何从 Agent 的日志与工具输出中检测并脱敏敏感信息。它提供两种互补的脱敏引擎：
- 离线规则引擎（regex，默认） —— 纯正则表达式 + 校验算法（Luhn、身份证校验码）， 无需 Ollama、无需网络、无需外部框架，结果确定、速度快，适合作为日志落盘前的第一道防线。 它同时覆盖 Agent 场景中最常泄露的密钥类敏感信息（API Key、云厂商令牌、私钥、连接串口令） 与传统 PII（身份证、手机号、信用卡、邮箱等）。
- 本地 LLM 引擎（llm） —— 通过 Ollama 调用一个本地小模型（默认
qwen3:0.6b）语义识别 Level 3 PII。呼应本章“小模型也能胜任结构化任务”的论点，同时也暴露小模型的局限 （例如可能返回带描述前缀的值而非原始字符串，导致回填失败）。
想快速看效果，直接运行
python main.py --demo（离线，无需任何依赖）即可看到多个代表性样本的 before/after 对比与脱敏类别汇总。
离线规则引擎覆盖的敏感信息类别¶
regex_sanitizer.py 按优先级处理以下类别（重叠时高优先级规则胜出），每类替换为带标签的占位符：
| 类别 | 占位符 | 说明 |
|---|---|---|
| 私钥 / 证书 | [REDACTED_PRIVATE_KEY] |
PEM 私钥块 |
| JWT | [REDACTED_JWT] |
eyJ... 三段式令牌 |
| 连接串凭据 | [REDACTED_URL_CRED] |
scheme://user:PASSWORD@host |
| AWS 访问密钥 | [REDACTED_AWS_KEY] |
AKIA... |
| GitHub / Slack / Google / OpenAI 密钥 | [REDACTED_*_TOKEN] / [REDACTED_API_KEY] |
ghp_、xoxb-、AIza、sk- |
| Bearer 令牌 | [REDACTED_BEARER_TOKEN] |
Authorization: Bearer ... |
| 口令 / 密钥赋值 | [REDACTED_SECRET] |
password=...、token: ... 等 |
| 邮箱 | [REDACTED_EMAIL] |
|
| 信用卡号 | [REDACTED_CREDIT_CARD] |
通过 Luhn 校验，降低误报 |
| IBAN | [REDACTED_IBAN] |
国际银行账号 |
| 美国社保号 | [REDACTED_SSN] |
|
| 身份证号 | [REDACTED_ID_CARD] |
中国大陆 18 位，含校验码验证 |
| 手机号 | [REDACTED_PHONE] |
中国大陆 |
| IP 地址 | [REDACTED_IP] |
IPv4 |
Level 3 PII Categories（LLM 引擎）¶
Based on the privacy protection architecture, Level 3 PII includes highly sensitive information: - Social Security Numbers (SSN) - Credit Card Numbers - Bank Account Numbers - Medical Record Numbers - Medical Diagnoses and Treatment Information - Prescription Information - Driver's License Numbers - Passport Numbers - Financial PINs - Tax ID Numbers - Health Insurance IDs - Biometric Data
Features¶
- Offline Rule Engine: Regex + Luhn/ID-checksum based sanitizer that needs no model, no network, and covers API keys/secrets in addition to PII
- Local LLM Processing: Uses Ollama with a local small model (default
qwen3:0.6b) for privacy-preserving PII detection - Internal Reasoning: Shows the model's thinking process using
<think>tags for transparency - Streaming Output: Real-time display of thinking and PII detection progress
- Performance Metrics: Measures TTFT (Time to First Token), token counts, and processing speeds
- Batch Processing: Can process multiple test cases from user-memory-evaluation framework
- Detailed Metrics: Tracks prefill time, output time, tokens per second for both phases
Installation¶
1. Install Ollama¶
通用回退（OpenRouter）：本实验默认用本地 Ollama 小模型。若 Ollama 不可用 （未运行 / 不可达）且设置了
OPENROUTER_API_KEY，Agent 会自动改走 OpenRouter （默认托管模型openai/gpt-5.6-luna）。想强制走回退做验证，可把 Ollama 指到一个 不可达端口：export OLLAMA_HOST=http://127.0.0.1:1。
macOS:¶
Linux:¶
Windows:¶
Download from ollama.com
说明：以下 Ollama 相关步骤仅在使用
--mode llm（本地 LLM 引擎）或运行 LLM 批量评测路径时才需要。 离线规则引擎（--demo、--input）只依赖 Python 标准库，无需安装 Ollama。
2. Pull the Qwen3 Model¶
Note: The 0.6B model requires approximately 500MB of disk space（可按需换用 qwen3:1.7b、qwen3:4b 提升准确率）。
3. Install Python Dependencies¶
Usage¶
完整参数说明见 python main.py --help（中文）。
离线规则演示（推荐，无需 Ollama）¶
对多个内置代表性样本展示 before/after 与脱敏类别汇总：
脱敏任意日志文件（离线）¶
python main.py --input app.log # 结果写到 app.log.sanitized
python main.py --input app.log -o cleaned.log # 指定输出文件
也可以直接运行规则引擎模块，仅对内置样本做演示：
使用本地 LLM 引擎¶
上述演示 / 文件脱敏加 --mode llm 即改用本地 Ollama 模型：
Process All Layer 3 Test Cases（LLM 批量评测路径）¶
Process all complex test cases from user-memory-evaluation（该路径固定使用 LLM，需要 Ollama 与 chapter3 评测框架）：
Process Specific Test Case¶
Limit Number of Test Cases¶
Process only the first N test cases:
选择模型¶
Output Structure¶
The sanitized logs and metrics are saved in the output/ directory:
output/
├── <test_id>_sanitized.txt # Sanitized conversation text
├── <test_id>_summary.json # Summary of PII found and replaced
├── performance_metrics.json # Detailed performance metrics
└── performance_summary.json # Aggregated performance statistics
Performance Metrics¶
The system tracks the following metrics for each conversation:
Timing Metrics¶
- Prefill Time (TTFT): Time to first token in milliseconds
- Output Time: Time to generate all output tokens
- Total Time: End-to-end processing time
Token Metrics¶
- Input Tokens: Number of tokens in the prompt
- Output Tokens: Number of tokens generated
- Prefill Speed: Tokens per second during prefill phase
- Output Speed: Tokens per second during generation
Sanitization Metrics¶
- PII Items Found: Number of Level 3 PII values detected
- Replacements Made: Number of replacements with [REDACTED]
Example Output¶
🚀 Starting Log Sanitization with Local LLM
============================================================
📦 Loading test cases from user-memory-evaluation...
🤖 Initializing Ollama agent...
✅ Using model: qwen3:0.6b
[1/1] Test Case: layer3_13_emergency_medical_cascade
Title: Emergency Medical Crisis - Multi-System Coordination Response
Conversations: 8
🔍 Processing conversation: emergency_room_001
Found 3 PII items
- 123-45-6789
- 4532 1234 5678 9012
- MRN-789456
============================================================
PERFORMANCE SUMMARY
============================================================
📊 Total Conversations Processed: 8
⏱️ Timing Metrics (milliseconds):
Prefill (TTFT): 125.34 ms (median: 118.50)
Output Time: 234.67 ms (median: 220.00)
Total Time: 360.01 ms (median: 338.50)
📝 Token Metrics:
Average Input Tokens: 450.5
Average Output Tokens: 25.3
Total Tokens Processed: 4206
⚡ Speed Metrics (tokens/second):
Prefill Speed: 3592.8 tok/s
Output Speed: 107.8 tok/s
🔒 Sanitization Results:
Total PII Items Found: 24
Total Replacements Made: 48
Average PII per Conversation: 3.0
Architecture¶
The project consists of several modules:
- regex_sanitizer.py: Offline rule-based sanitizer (regex + Luhn/ID checksums), covers keys/secrets and PII
- samples.py: Representative agent-log samples used by the offline demo
- config.py: Configuration for Ollama model and PII categories
- test_loader.py: Loads test cases from user-memory-evaluation framework
- agent.py: Core LLM sanitization logic using Ollama
- metrics.py: Performance metrics collection and reporting
- main.py: Main entry point and orchestration
How It Works¶
- Test Case Loading: The system loads conversation histories from the user-memory-evaluation framework
- PII Detection: Each conversation is sent to the local Qwen3 0.6B model with a specialized prompt to detect Level 3 PII
- Sanitization: Detected PII values are replaced with [REDACTED] in the original text
- Metrics Collection: Performance metrics are collected for each operation
- Output Generation: Sanitized logs and performance summaries are saved to the output directory
Privacy Considerations¶
- All processing happens locally using Ollama - no data is sent to external APIs
- The Qwen3 0.6B model runs entirely on your local machine
- Sanitized logs replace sensitive information with [REDACTED] placeholders
- Original PII values are logged for verification but should be handled securely
Troubleshooting¶
"Ollama not found"¶
Make sure Ollama is installed and running:
"Model qwen3:0.6b not found"¶
Pull the model:
"Evaluation framework not found"¶
Ensure the user-memory-evaluation project exists at:
源代码¶
agent.py¶
"""
Log Sanitization Agent using Local Ollama LLM
"""
import os
import time
import re
import json
from typing import List, Tuple, Dict, Optional
from pathlib import Path
import ollama
from config import (
OLLAMA_MODEL,
OLLAMA_TEMPERATURE,
SYSTEM_PROMPT,
USER_PROMPT_TEMPLATE,
PII_DETECTION_SCHEMA,
OUTPUT_DIR
)
from metrics import PerformanceMetrics, MetricsCollector
class LogSanitizationAgent:
"""Agent for sanitizing logs using local Qwen3 0.6B model via Ollama"""
def __init__(self, model: str = OLLAMA_MODEL):
"""Initialize the sanitization agent.
Primary backend is the local Ollama model. If Ollama is unavailable
(not running / not reachable) and OPENROUTER_API_KEY is set, the agent
falls back to OpenRouter (default hosted model: openai/gpt-5.6-luna),
so the experiment still runs without a local model.
"""
self.model = model
self.backend = "ollama"
self.metrics_collector = MetricsCollector(OUTPUT_DIR)
# Try the local Ollama backend first.
try:
self.client = ollama.Client()
models = self.client.list()
# models is a dict with 'models' key containing a list
if isinstance(models, dict) and 'models' in models:
available_models = [m.get('name', '') for m in models['models']]
else:
# If it's a direct list (older API versions)
available_models = [m.get('name', '') for m in models] if isinstance(models, list) else []
if not any(self.model in m for m in available_models):
print(f"⚠️ Model {self.model} not found. Pulling it now...")
self.client.pull(self.model)
print(f"✅ Model {self.model} pulled successfully")
else:
print(f"✅ Using model: {self.model}")
except Exception as e:
# Universal fallback: route through OpenRouter when Ollama is down.
openrouter_key = os.getenv("OPENROUTER_API_KEY")
if openrouter_key:
from openai import OpenAI
from openrouter_fallback import (
OPENROUTER_BASE_URL,
map_model_to_openrouter,
)
self.backend = "openrouter"
self.client = OpenAI(api_key=openrouter_key,
base_url=OPENROUTER_BASE_URL)
# 本地小模型（qwen3:0.6b 等）在 OpenRouter 上未必可用，
# 默认改用 openai/gpt-5.6-luna；带 "/" 的 id 原样透传。
self.model = (map_model_to_openrouter(self.model)
if "/" in self.model else "openai/gpt-5.6-luna")
print(f"⚠️ Ollama unavailable ({e}); "
f"falling back to OpenRouter model: {self.model}")
else:
print(f"❌ Failed to connect to Ollama: {e}")
print("Please ensure Ollama is running: ollama serve, "
"or set OPENROUTER_API_KEY as a fallback")
raise
def _chat_stream(self, messages):
"""Yield content chunks from the active backend (Ollama or OpenRouter)."""
if self.backend == "ollama":
stream = self.client.chat(
model=self.model,
messages=messages,
stream=True,
format=PII_DETECTION_SCHEMA, # Use structured output format
options={
"temperature": OLLAMA_TEMPERATURE,
"num_predict": 1000,
}
)
for chunk in stream:
yield chunk.get('message', {}).get('content', '')
else:
# 用与 Ollama 相同的 JSON Schema 强约束输出结构（pii_values 数组），
# 避免模型自行发明字段名。strict 模式要求 additionalProperties=false。
strict_schema = dict(PII_DETECTION_SCHEMA)
strict_schema["additionalProperties"] = False
stream = self.client.chat.completions.create(
model=self.model,
messages=messages,
stream=True,
temperature=OLLAMA_TEMPERATURE,
response_format={
"type": "json_schema",
"json_schema": {
"name": "pii_detection",
"strict": True,
"schema": strict_schema,
},
},
max_tokens=1000,
)
for chunk in stream:
if not chunk.choices:
continue
yield chunk.choices[0].delta.content or ""
def count_tokens(self, text: str) -> int:
"""Estimate token count (rough approximation)"""
# Rough estimate: 1 token ≈ 4 characters for English text
# For more accurate counting, we'd need the actual tokenizer
return len(text) // 4
def detect_pii(self, conversation_text: str) -> Tuple[List[str], Dict]:
"""
Detect Level 3 PII in conversation text using local LLM
Args:
conversation_text: Text to analyze
Returns:
- List of detected PII values
- Performance metrics dictionary
"""
# Prepare the prompt
user_prompt = USER_PROMPT_TEMPLATE.format(conversation_text=conversation_text)
# Count input tokens
input_tokens = self.count_tokens(SYSTEM_PROMPT + user_prompt)
# Measure prefill time (time to first token)
start_time = time.perf_counter()
# Create messages for Ollama
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_prompt}
]
# Track first token time
first_token_time = None
output_tokens_count = 0
full_response = ""
try:
# Use structured output with JSON schema (backend-agnostic stream)
print("\n 🧠 Analyzing (JSON): \033[90m", end="", flush=True) # Gray color for JSON
for content in self._chat_stream(messages):
if first_token_time is None and content:
first_token_time = time.perf_counter()
full_response += content
output_tokens_count += len(content) // 4 # Rough token estimate
# Stream the actual content
if content:
print(content, end="", flush=True)
print("\033[0m") # Reset color and new line
end_time = time.perf_counter()
except Exception as e:
print(f"\n❌ Error during PII detection: {e}")
return [], {}
# Calculate performance metrics
prefill_time_ms = (first_token_time - start_time) * 1000 if first_token_time else 0
total_time_ms = (end_time - start_time) * 1000
output_time_ms = total_time_ms - prefill_time_ms
prefill_speed = input_tokens / (prefill_time_ms / 1000) if prefill_time_ms > 0 else 0
output_speed = output_tokens_count / (output_time_ms / 1000) if output_time_ms > 0 else 0
# Parse JSON response
pii_values = []
try:
response_json = json.loads(full_response)
# Extract PII values
pii_values = response_json.get('pii_values', [])
# Strip leading/trailing whitespace and special characters like '-' or empty
cleaned_pii_values = []
for pii in pii_values:
if pii and isinstance(pii, str):
cleaned = pii.strip().strip('-').strip()
if cleaned:
cleaned_pii_values.append(cleaned)
pii_values = cleaned_pii_values
except json.JSONDecodeError as e:
print(f"\n ⚠️ Failed to parse JSON response: {e}")
# Fallback to simple line splitting if JSON parsing fails
pii_values = [line.strip() for line in full_response.split('\n') if line.strip()]
metrics = {
'input_tokens': input_tokens,
'output_tokens': output_tokens_count,
'prefill_time_ms': prefill_time_ms,
'output_time_ms': output_time_ms,
'total_time_ms': total_time_ms,
'prefill_speed_tps': prefill_speed,
'output_speed_tps': output_speed,
'pii_items_found': len(pii_values)
}
return pii_values, metrics
def sanitize_text(self, text: str, pii_values: List[str]) -> Tuple[str, int]:
"""
Replace PII values with [REDACTED] in the text
Returns:
- Sanitized text
- Number of replacements made
"""
sanitized = text
replacements = 0
for pii_value in pii_values:
# Escape special regex characters in PII value
escaped_value = re.escape(pii_value)
# Count occurrences before replacement
occurrences = len(re.findall(escaped_value, sanitized, re.IGNORECASE))
# Replace all occurrences
sanitized = re.sub(escaped_value, '[REDACTED]', sanitized, flags=re.IGNORECASE)
replacements += occurrences
return sanitized, replacements
def sanitize_conversation(
self,
conversation: Dict,
test_id: str = "unknown"
) -> Dict:
"""
Sanitize a single conversation and collect metrics
Returns:
Dictionary with sanitized conversation and metrics
"""
# Format conversation text
conv_text = self.format_conversation(conversation)
conv_id = conversation.get('conversation_id', 'unknown')
print(f"🔍 Processing conversation: {conv_id}")
# Detect PII
pii_values, perf_metrics = self.detect_pii(conv_text)
if pii_values:
print(f" ✅ Found {len(pii_values)} PII items:")
for pii in pii_values:
print(f" - {pii}")
else:
print(" ⚠️ No PII items detected")
# Sanitize the text
sanitized_text, replacements = self.sanitize_text(conv_text, pii_values)
# Create performance metric. detect_pii() returns an empty metrics dict
# when the LLM backend fails (e.g. Ollama not running) — fall back to
# zeros so one failed conversation doesn't crash the whole batch.
metric = PerformanceMetrics(
test_id=test_id,
conversation_id=conv_id,
input_text_length=len(conv_text),
input_tokens=perf_metrics.get('input_tokens', 0),
prefill_time_ms=perf_metrics.get('prefill_time_ms', 0),
output_time_ms=perf_metrics.get('output_time_ms', 0),
total_time_ms=perf_metrics.get('total_time_ms', 0),
output_tokens=perf_metrics.get('output_tokens', 0),
prefill_speed_tps=perf_metrics.get('prefill_speed_tps', 0),
output_speed_tps=perf_metrics.get('output_speed_tps', 0),
pii_items_found=perf_metrics.get('pii_items_found', 0),
replacements_made=replacements,
sanitized_text_length=len(sanitized_text)
)
self.metrics_collector.add_metric(metric)
return {
'conversation_id': conv_id,
'original_length': len(conv_text),
'sanitized_length': len(sanitized_text),
'pii_found': pii_values,
'replacements_made': replacements,
'sanitized_text': sanitized_text,
'metrics': metric.to_dict()
}
def format_conversation(self, conversation: Dict) -> str:
"""Format conversation dictionary into text"""
lines = []
lines.append(f"Conversation ID: {conversation.get('conversation_id', 'unknown')}")
lines.append(f"Timestamp: {conversation.get('timestamp', 'unknown')}")
lines.append("-" * 50)
messages = conversation.get('messages', [])
for msg in messages:
role = msg.get('role', 'unknown').upper()
content = msg.get('content', '')
lines.append(f"{role}: {content}")
lines.append("") # Empty line between messages
return "\n".join(lines)
def save_sanitized_log(self, test_id: str, results: List[Dict]):
"""Save sanitized logs to output directory"""
output_file = OUTPUT_DIR / f"{test_id}_sanitized.txt"
summary_file = OUTPUT_DIR / f"{test_id}_summary.json"
# Save sanitized text
with open(output_file, 'w') as f:
for result in results:
f.write(f"\n{'='*60}\n")
f.write(f"Conversation: {result['conversation_id']}\n")
f.write(f"{'='*60}\n")
f.write(result['sanitized_text'])
f.write("\n")
# Save summary
summary = {
'test_id': test_id,
'total_conversations': len(results),
'total_pii_found': sum(len(r['pii_found']) for r in results),
'total_replacements': sum(r['replacements_made'] for r in results),
'conversations': [
{
'conversation_id': r['conversation_id'],
'pii_count': len(r['pii_found']),
'replacements': r['replacements_made']
}
for r in results
]
}
with open(summary_file, 'w') as f:
json.dump(summary, f, indent=2)
print(f"✅ Sanitized log saved to: {output_file}")
print(f"✅ Summary saved to: {summary_file}")
def process_test_case(self, test_id: str, conversations: List[Dict]) -> List[Dict]:
"""Process all conversations in a test case"""
results = []
print(f"\n{'='*60}")
print(f"Processing Test Case: {test_id}")
print(f"Total Conversations: {len(conversations)}")
print(f"{'='*60}")
for i, conv in enumerate(conversations, 1):
print(f"\n[{i}/{len(conversations)}] ", end="")
result = self.sanitize_conversation(conv, test_id)
results.append(result)
# Save results
self.save_sanitized_log(test_id, results)
# Save metrics
self.metrics_collector.save_metrics()
self.metrics_collector.print_summary()
return results
config.py¶
"""
Configuration for Log Sanitization with Local LLM
"""
import os
from pathlib import Path
# Ollama Configuration
# 默认使用 0.6B 超小模型，呼应本章“小模型也能胜任结构化任务”的论点，
# 且可在 CPU / 消费级设备上运行；可用 --model 覆盖为 qwen3:1.7b、qwen3:4b 等。
OLLAMA_MODEL = "qwen3:0.6b"
OLLAMA_TEMPERATURE = 0.1 # Low temperature for consistent detection
# Paths
PROJECT_ROOT = Path(__file__).parent
OUTPUT_DIR = PROJECT_ROOT / "output"
OUTPUT_DIR.mkdir(exist_ok=True)
# Performance Metrics Configuration
METRICS_FILE = OUTPUT_DIR / "performance_metrics.json"
# Evaluation Framework Path
# The user-memory-evaluation framework lives in chapter3, not chapter2.
# PROJECT_ROOT = chapter2/log-sanitization, so go up two levels to the repo root.
EVAL_FRAMEWORK_PATH = PROJECT_ROOT.parent.parent / "chapter3" / "user-memory-evaluation"
# System Prompt for PII Detection
SYSTEM_PROMPT = """You are a privacy protection agent that detects Level 3 PII.
Level 3 PII includes:
- Social Security Numbers (SSN) - format: XXX-XX-XXXX or XXXXXXXXX
- Credit Card Numbers - format: XXXX XXXX XXXX XXXX or 16 digits
- Credit Card Expiry Date and CVV
- Bank Account Numbers
- Full Residential Addresses
- Medical Record Numbers
- Medical Diagnoses and Treatment Details
- Prescription Information
- Driver's License Numbers
- Passport Numbers
- Financial PINs
- Tax ID Numbers
- Health Insurance IDs
- Biometric Data
- Usernames for Financial Accounts
- Passwords
Analyze the conversation and return a JSON with the exact PII values found. NEVER use placeholders. Simply return the PII values found."""
USER_PROMPT_TEMPLATE = """Analyze the following conversation for Level 3 PII:
{conversation_text}"""
# JSON Schema for structured output
PII_DETECTION_SCHEMA = {
"type": "object",
"properties": {
"pii_values": {
"type": "array",
"items": {
"type": "string"
},
"description": "Array of exact PII values found in the conversation. NEVER use placeholders."
}
},
"required": ["pii_values"]
}
main.py¶
#!/usr/bin/env python3
"""
Main script for Log Sanitization using Local LLM
"""
import argparse
import sys
from collections import Counter
from pathlib import Path
from typing import Optional
from config import OUTPUT_DIR, OLLAMA_MODEL
import regex_sanitizer
from samples import SAMPLES
def main(test_id: Optional[str] = None, limit: Optional[int] = None,
model: str = OLLAMA_MODEL):
"""
Main function to run log sanitization
Args:
test_id: Specific test case ID to process (optional)
limit: Maximum number of test cases to process (optional)
"""
print("🚀 Starting Log Sanitization with Local LLM")
print("=" * 60)
# Initialize components
try:
from agent import LogSanitizationAgent
from test_loader import TestCaseLoader
print("📦 Loading test cases from user-memory-evaluation...")
loader = TestCaseLoader()
print(f"🤖 Initializing Ollama agent (model: {model})...")
agent = LogSanitizationAgent(model=model)
except Exception as e:
print(f"❌ Initialization failed: {e}")
return 1
# Get test cases to process
if test_id:
# Process specific test case
print(f"\n📋 Processing specific test case: {test_id}")
conversations = loader.get_test_case_conversations(test_id)
if not conversations:
print(f"❌ Test case {test_id} not found or has no conversations")
return 1
agent.process_test_case(test_id, conversations)
else:
# Process Layer 3 test cases (most complex, likely to have PII)
print("\n📋 Getting Layer 3 test cases...")
test_cases = loader.get_layer3_test_cases()
if not test_cases:
print("❌ No Layer 3 test cases found")
return 1
print(f"Found {len(test_cases)} Layer 3 test cases")
# Apply limit if specified
if limit:
test_cases = test_cases[:limit]
print(f"Processing first {limit} test cases")
# Process each test case
for i, tc in enumerate(test_cases, 1):
print(f"\n[{i}/{len(test_cases)}] Test Case: {tc['test_id']}")
print(f" Title: {tc['title']}")
print(f" Conversations: {tc['num_conversations']}")
# Get conversation histories
conversations = loader.get_test_case_conversations(tc['test_id'])
if conversations:
agent.process_test_case(tc['test_id'], conversations)
else:
print(f" ⚠️ No conversations found for {tc['test_id']}")
print("\n" + "=" * 60)
print("✅ Log Sanitization Complete!")
print(f"📁 Results saved to: {OUTPUT_DIR}")
return 0
def demo_regex_mode():
"""离线规则脱敏演示：对多个代表性样本展示 before/after 与类别汇总"""
print("🎯 离线规则脱敏演示 (regex 模式，无需 Ollama)")
print("=" * 60)
print(f"共 {len(SAMPLES)} 个代表性样本，覆盖密钥 / 令牌 / 私钥 / PII 等类别\n")
total = Counter()
total_hits = 0
for name, text in SAMPLES:
redacted, findings = regex_sanitizer.sanitize(text)
regex_sanitizer.print_report(name, text, redacted, findings)
total.update(regex_sanitizer.summarize(findings))
total_hits += len(findings)
print(f"\n{'=' * 64}")
print("脱敏类别汇总 (across all samples)")
print("=" * 64)
for category, count in total.most_common():
label = regex_sanitizer.CATEGORY_LABELS.get(category, category)
print(f" {label:<16} {count} 处")
print(f"\n 合计脱敏 {total_hits} 处敏感信息，覆盖 {len(total)} 个类别")
return 0
def sanitize_file(input_path: str, output_path: Optional[str] = None,
mode: str = "regex", model: str = OLLAMA_MODEL):
"""对任意日志文件执行脱敏，结果写入输出文件"""
in_file = Path(input_path)
if not in_file.exists():
print(f"❌ 输入文件不存在: {input_path}")
return 1
text = in_file.read_text(encoding="utf-8", errors="replace")
out_file = Path(output_path) if output_path else in_file.with_suffix(in_file.suffix + ".sanitized")
if mode == "regex":
print(f"🔍 使用离线规则引擎脱敏: {input_path}")
redacted, findings = regex_sanitizer.sanitize(text)
counts = regex_sanitizer.summarize(findings)
else:
print(f"🔍 使用本地 LLM ({model}) 脱敏: {input_path}")
try:
from agent import LogSanitizationAgent
except Exception as e:
print(f"❌ 加载 LLM 引擎失败: {e}")
return 1
agent = LogSanitizationAgent(model=model)
pii_values, _ = agent.detect_pii(text)
redacted, _ = agent.sanitize_text(text, pii_values)
counts = Counter({"pii": len(pii_values)})
findings = pii_values
out_file.write_text(redacted, encoding="utf-8")
print(f"\n✅ 已写入脱敏结果: {out_file}")
print(f" 共脱敏 {sum(counts.values())} 处敏感信息")
for category, count in counts.most_common():
label = regex_sanitizer.CATEGORY_LABELS.get(category, category)
print(f" - {label}: {count} 处")
return 0
def demo_mode(model: str = OLLAMA_MODEL):
"""Run a quick demo with sample PII-containing text (本地 LLM 模式)"""
print("🎯 Running Demo Mode (LLM)")
print("=" * 60)
# Create a sample conversation with Level 3 PII
sample_conversation = {
'conversation_id': 'demo_001',
'timestamp': '2024-01-01 10:00:00',
'messages': [
{
'role': 'user',
'content': 'I need to update my information. My SSN is 123-45-6789.'
},
{
'role': 'assistant',
'content': 'I can help you update your information. Can you confirm your credit card?'
},
{
'role': 'user',
'content': 'Yes, it\'s 4532 1234 5678 9012. Also, my medical record number is MRN-789456.'
},
{
'role': 'assistant',
'content': 'Thank you. I\'ve noted your SSN ending in 6789 and card ending in 9012.'
},
{
'role': 'user',
'content': 'Great. My driver\'s license is DL-123456789 and passport is P987654321.'
}
]
}
try:
from agent import LogSanitizationAgent
agent = LogSanitizationAgent(model=model)
print("\n📝 Sample conversation created with Level 3 PII")
print("🔍 Detecting and sanitizing PII...\n")
result = agent.sanitize_conversation(sample_conversation, 'demo')
print("\n" + "=" * 60)
print("DEMO RESULTS")
print("=" * 60)
print(f"PII Items Found: {len(result['pii_found'])}")
for pii in result['pii_found']:
print(f" - {pii}")
print(f"\nReplacements Made: {result['replacements_made']}")
print("\n--- SANITIZED TEXT ---")
print(result['sanitized_text'])
# Save demo results
agent.save_sanitized_log('demo', [result])
agent.metrics_collector.save_metrics()
agent.metrics_collector.print_summary()
except Exception as e:
print(f"❌ Demo failed: {e}")
return 1
return 0
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="日志脱敏实验：从 Agent 日志 / 工具输出中检测并脱敏敏感信息"
"（API 密钥、令牌、私钥、信用卡、身份证、手机号、邮箱等）。",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""\
两种脱敏引擎：
regex 离线规则引擎（默认），基于正则 + 校验算法，无需 Ollama，结果确定、速度快
llm 本地 LLM 引擎，通过 Ollama 调用小模型（默认 qwen3:0.6b）语义识别 Level 3 PII
常用示例：
python main.py --demo # 离线跑内置样本，展示 before/after 与脱敏汇总
python main.py --demo --mode llm # 用本地 LLM 跑演示样本
python main.py --input app.log # 离线脱敏一个日志文件
python main.py --input app.log -o out.log # 指定输出文件
python main.py --input app.log --mode llm # 用本地 LLM 脱敏文件
python main.py # (LLM) 批量处理 chapter3 评测框架中的 Layer 3 用例
python main.py --test-id layer3_01_travel_coordination
python main.py --limit 3 --model qwen3:1.7b
""",
)
parser.add_argument(
'--mode',
choices=['regex', 'llm'],
default='regex',
help='脱敏引擎：regex=离线规则(默认)，llm=本地 Ollama 模型。'
'（注意：不带 --demo/--input 的批量评测路径始终使用 LLM）'
)
parser.add_argument(
'-i', '--input',
type=str,
metavar='FILE',
help='待脱敏的日志文件路径（配合 --mode 选择引擎）'
)
parser.add_argument(
'-o', '--output',
type=str,
metavar='FILE',
help='脱敏结果输出文件路径（仅 --input 模式生效，默认写到 <输入>.sanitized）'
)
parser.add_argument(
'--model',
type=str,
default=OLLAMA_MODEL,
help=f'Ollama 模型名（默认 {OLLAMA_MODEL}），仅 llm 模式生效'
)
parser.add_argument(
'--test-id',
type=str,
help='仅处理指定 ID 的评测用例（LLM 批量路径）'
)
parser.add_argument(
'--limit',
type=int,
help='最多处理多少个评测用例（LLM 批量路径）'
)
parser.add_argument(
'--demo',
action='store_true',
help='运行演示：默认离线规则引擎跑内置代表性样本；加 --mode llm 则用本地 LLM'
)
args = parser.parse_args()
if args.input:
exit_code = sanitize_file(args.input, args.output, mode=args.mode, model=args.model)
elif args.demo:
if args.mode == 'llm':
exit_code = demo_mode(model=args.model)
else:
exit_code = demo_regex_mode()
else:
exit_code = main(test_id=args.test_id, limit=args.limit, model=args.model)
sys.exit(exit_code)
metrics.py¶
"""
Performance Metrics Module for Log Sanitization
"""
import time
import json
from typing import Dict, List, Optional
from pathlib import Path
from dataclasses import dataclass, asdict
from datetime import datetime
@dataclass
class PerformanceMetrics:
"""Store performance metrics for a single sanitization operation"""
test_id: str
conversation_id: str
input_text_length: int
input_tokens: int
# Timing metrics
prefill_time_ms: float # Time to First Token (TTFT)
output_time_ms: float
total_time_ms: float
# Token metrics
output_tokens: int
prefill_speed_tps: float # tokens per second
output_speed_tps: float
# Sanitization results
pii_items_found: int
replacements_made: int
sanitized_text_length: int
# Timestamps
timestamp: str = ""
def __post_init__(self):
if not self.timestamp:
self.timestamp = datetime.now().isoformat()
def to_dict(self) -> Dict:
"""Convert to dictionary for JSON serialization"""
return asdict(self)
class MetricsCollector:
"""Collect and aggregate performance metrics"""
def __init__(self, output_dir: Path):
self.output_dir = output_dir
self.metrics_file = output_dir / "performance_metrics.json"
self.summary_file = output_dir / "performance_summary.json"
self.metrics: List[PerformanceMetrics] = []
def add_metric(self, metric: PerformanceMetrics):
"""Add a new metric to the collection"""
self.metrics.append(metric)
def calculate_summary(self) -> Dict:
"""Calculate summary statistics across all metrics"""
if not self.metrics:
return {"error": "No metrics collected"}
# Collect all values for each metric
prefill_times = [m.prefill_time_ms for m in self.metrics]
output_times = [m.output_time_ms for m in self.metrics]
total_times = [m.total_time_ms for m in self.metrics]
input_tokens = [m.input_tokens for m in self.metrics]
output_tokens = [m.output_tokens for m in self.metrics]
prefill_speeds = [m.prefill_speed_tps for m in self.metrics]
output_speeds = [m.output_speed_tps for m in self.metrics]
pii_counts = [m.pii_items_found for m in self.metrics]
replacements = [m.replacements_made for m in self.metrics]
def calculate_stats(values: List[float]) -> Dict:
"""Calculate min, max, mean, median for a list of values"""
if not values:
return {"min": 0, "max": 0, "mean": 0, "median": 0}
sorted_values = sorted(values)
n = len(sorted_values)
return {
"min": min(values),
"max": max(values),
"mean": sum(values) / n,
"median": sorted_values[n // 2] if n % 2 == 1 else
(sorted_values[n // 2 - 1] + sorted_values[n // 2]) / 2
}
summary = {
"total_conversations": len(self.metrics),
"timestamp": datetime.now().isoformat(),
"timing_metrics": {
"prefill_time_ms": calculate_stats(prefill_times),
"output_time_ms": calculate_stats(output_times),
"total_time_ms": calculate_stats(total_times)
},
"token_metrics": {
"input_tokens": calculate_stats(input_tokens),
"output_tokens": calculate_stats(output_tokens),
"total_input_tokens": sum(input_tokens),
"total_output_tokens": sum(output_tokens)
},
"speed_metrics": {
"prefill_speed_tps": calculate_stats(prefill_speeds),
"output_speed_tps": calculate_stats(output_speeds)
},
"sanitization_metrics": {
"pii_items_found": calculate_stats(pii_counts),
"replacements_made": calculate_stats(replacements),
"total_pii_found": sum(pii_counts),
"total_replacements": sum(replacements)
}
}
return summary
def save_metrics(self):
"""Save all metrics and summary to files"""
# Save detailed metrics
metrics_data = [m.to_dict() for m in self.metrics]
with open(self.metrics_file, 'w') as f:
json.dump(metrics_data, f, indent=2)
# Save summary
summary = self.calculate_summary()
with open(self.summary_file, 'w') as f:
json.dump(summary, f, indent=2)
print(f"✅ Metrics saved to {self.metrics_file}")
print(f"✅ Summary saved to {self.summary_file}")
def print_summary(self):
"""Print a human-readable summary of metrics"""
summary = self.calculate_summary()
print("\n" + "=" * 60)
print("PERFORMANCE SUMMARY")
print("=" * 60)
print(f"\n📊 Total Conversations Processed: {summary['total_conversations']}")
print("\n⏱️ Timing Metrics (milliseconds):")
timing = summary['timing_metrics']
print(f" Prefill (TTFT): {timing['prefill_time_ms']['mean']:.2f} ms (median: {timing['prefill_time_ms']['median']:.2f})")
print(f" Output Time: {timing['output_time_ms']['mean']:.2f} ms (median: {timing['output_time_ms']['median']:.2f})")
print(f" Total Time: {timing['total_time_ms']['mean']:.2f} ms (median: {timing['total_time_ms']['median']:.2f})")
print("\n📝 Token Metrics:")
tokens = summary['token_metrics']
print(f" Average Input Tokens: {tokens['input_tokens']['mean']:.1f}")
print(f" Average Output Tokens: {tokens['output_tokens']['mean']:.1f}")
print(f" Total Tokens Processed: {tokens['total_input_tokens'] + tokens['total_output_tokens']}")
print("\n⚡ Speed Metrics (tokens/second):")
speed = summary['speed_metrics']
print(f" Prefill Speed: {speed['prefill_speed_tps']['mean']:.1f} tok/s")
print(f" Output Speed: {speed['output_speed_tps']['mean']:.1f} tok/s")
print("\n🔒 Sanitization Results:")
sanitization = summary['sanitization_metrics']
print(f" Total PII Items Found: {sanitization['total_pii_found']}")
print(f" Total Replacements Made: {sanitization['total_replacements']}")
print(f" Average PII per Conversation: {sanitization['pii_items_found']['mean']:.1f}")
print("\n" + "=" * 60)
openrouter_fallback.py¶
"""Universal OpenRouter fallback helper (Chapter 2 experiments).
Goal: every experiment keeps working when the direct provider key is missing
but ``OPENROUTER_API_KEY`` is present. Default behavior is fully preserved:
* If a primary provider key is present -> use the primary provider unchanged.
* Else if ``OPENROUTER_API_KEY`` is present -> route through OpenRouter
(base_url=https://openrouter.ai/api/v1) and translate the model id.
* Else raise a clear error listing every accepted key.
Model translation (only applied when the OpenRouter fallback is active):
* ids already containing "/" are passed through unchanged;
* gpt-* / o1* / o3* / o4* / chatgpt* -> "openai/<id>";
* claude-* -> "anthropic/claude-opus-4.8";
* kimi-* -> "moonshotai/kimi-k2.6";
* anything else -> passed through unchanged.
"""
import os
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
def is_openrouter_key(api_key):
"""OpenRouter keys reliably start with ``sk-or-``."""
return bool(api_key) and api_key.startswith("sk-or-")
def map_model_to_openrouter(model):
"""Translate a bare provider model id into an OpenRouter-qualified id."""
if not model or "/" in model:
return model
low = model.lower()
if low.startswith(("gpt-", "gpt5", "o1", "o3", "o4", "chatgpt")):
return "openai/" + model
if low.startswith("claude"):
return "anthropic/claude-opus-4.8"
if low.startswith("kimi"):
return "moonshotai/kimi-k2.6"
return model
def resolve_llm(model=None, primary_keys=("OPENAI_API_KEY",), primary_base_url=None):
"""Resolve ``(api_key, base_url, model)`` honoring the primary->OpenRouter fallback.
Args:
model: requested model id (may be remapped when the fallback activates).
primary_keys: env var names checked in order for the primary provider.
primary_base_url: base_url used when a primary key is found
(``None`` means the OpenAI SDK default / official endpoint).
"""
or_key = os.getenv("OPENROUTER_API_KEY")
# gpt-5.x (incl. gpt-5.6*) needs OpenAI org-verification on the direct API;
# when an OpenRouter key is present, prefer routing these ids through it.
if or_key and model and model.lower().startswith("gpt-5"):
return or_key, OPENROUTER_BASE_URL, map_model_to_openrouter(model)
for env in primary_keys:
key = os.getenv(env)
if key:
return key, primary_base_url, model
if or_key:
return or_key, OPENROUTER_BASE_URL, map_model_to_openrouter(model)
accepted = ", ".join(list(primary_keys) + ["OPENROUTER_API_KEY"])
raise RuntimeError(
"No LLM API key found. Set one of the primary keys or the universal "
"fallback. Accepted keys: " + accepted + ". See env.example."
)
regex_sanitizer.py¶
"""
基于规则（正则）的离线日志脱敏引擎
与 agent.py 中依赖本地 LLM 的方案互补：本模块不需要任何模型或网络，
纯靠正则表达式 + 校验算法（Luhn、身份证校验码）识别日志 / 工具输出中的
敏感信息，速度快、结果确定，适合作为 Agent 日志落盘前的第一道防线。
覆盖的敏感信息类别（按匹配优先级从高到低）：
- 私钥 / 证书（PEM 块）
- JWT
- 云厂商与第三方密钥（AWS AKIA、GitHub、Slack、Google、OpenAI 风格 sk-）
- HTTP Authorization: Bearer 令牌
- 配置中的口令 / 密钥赋值（password=..., token: ... 等）
- 邮箱地址
- 信用卡号（Luhn 校验）
- IBAN 国际银行账号
- 美国社会安全号（SSN）
- 中国大陆身份证号（校验码验证）
- 中国大陆手机号
- IPv4 地址
每一类都会被替换为带类别标签的占位符（如 [REDACTED_API_KEY]），
既隐去了原值，又保留了“这里原本是什么”的可读性，方便排障。
"""
import re
from collections import Counter
from typing import Dict, List, Tuple
def _luhn_ok(number: str) -> bool:
"""Luhn 校验，用于降低信用卡号的误报率"""
digits = [int(c) for c in number if c.isdigit()]
if not 13 <= len(digits) <= 19:
return False
checksum = 0
parity = len(digits) % 2
for i, d in enumerate(digits):
if i % 2 == parity:
d *= 2
if d > 9:
d -= 9
checksum += d
return checksum % 10 == 0
def _cn_id_ok(value: str) -> bool:
"""中国大陆二代身份证号（18 位）校验码验证"""
s = value.upper()
if len(s) != 18 or not s[:17].isdigit():
return False
weights = [7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
check_codes = "10X98765432"
total = sum(int(s[i]) * weights[i] for i in range(17))
return check_codes[total % 11] == s[17]
# 每条规则：(类别, 占位符, 编译后的正则, 用于取值的分组号, 可选校验函数)
# 分组号为 0 表示整段命中都要脱敏；为 N 表示只脱敏第 N 个捕获组（保留键名等上下文）。
_RULES = [
(
"private_key", "[REDACTED_PRIVATE_KEY]",
re.compile(
r"-----BEGIN (?:RSA |EC |DSA |OPENSSH |PGP )?PRIVATE KEY-----"
r"[\s\S]*?-----END (?:RSA |EC |DSA |OPENSSH |PGP )?PRIVATE KEY-----"
),
0, None,
),
(
"jwt", "[REDACTED_JWT]",
re.compile(r"eyJ[A-Za-z0-9_-]{10,}\.eyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}"),
0, None,
),
(
# 连接串中的口令，如 postgres://user:PASSWORD@host:5432/db
"url_credential", "[REDACTED_URL_CRED]",
re.compile(r"://[^\s:/@]+:([^\s:/@]+)@"),
1, None,
),
(
"aws_access_key", "[REDACTED_AWS_KEY]",
re.compile(r"\bAKIA[0-9A-Z]{16}\b"),
0, None,
),
(
"github_token", "[REDACTED_GITHUB_TOKEN]",
re.compile(r"\bgh[pousr]_[A-Za-z0-9]{36,}\b"),
0, None,
),
(
"slack_token", "[REDACTED_SLACK_TOKEN]",
re.compile(r"\bxox[baprs]-[A-Za-z0-9-]{10,}\b"),
0, None,
),
(
"google_api_key", "[REDACTED_GOOGLE_API_KEY]",
re.compile(r"\bAIza[0-9A-Za-z_-]{35}\b"),
0, None,
),
(
"api_key", "[REDACTED_API_KEY]",
re.compile(r"\bsk-[A-Za-z0-9_-]{20,}\b"),
0, None,
),
(
"bearer_token", "[REDACTED_BEARER_TOKEN]",
re.compile(r"(?i)\bBearer\s+([A-Za-z0-9._~+/=-]{10,})"),
1, None,
),
(
"secret_assignment", "[REDACTED_SECRET]",
re.compile(
r"(?i)(?:password|passwd|pwd|secret|token|api[_-]?key|"
r"access[_-]?key|auth|credential)[\"']?\s*[=:]\s*[\"']?([^\s\"',}]{4,})"
),
1, None,
),
(
"email", "[REDACTED_EMAIL]",
re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b"),
0, None,
),
(
"credit_card", "[REDACTED_CREDIT_CARD]",
re.compile(r"\b(?:\d[ -]?){13,19}\b"),
0, _luhn_ok,
),
(
"iban", "[REDACTED_IBAN]",
re.compile(r"\b[A-Z]{2}\d{2}[A-Z0-9]{11,30}\b"),
0, None,
),
(
"us_ssn", "[REDACTED_SSN]",
re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),
0, None,
),
(
"cn_id_card", "[REDACTED_ID_CARD]",
re.compile(r"\b\d{17}[\dXx]\b"),
0, _cn_id_ok,
),
(
"cn_phone", "[REDACTED_PHONE]",
re.compile(r"(?<!\d)1[3-9]\d{9}(?!\d)"),
0, None,
),
(
"ip_address", "[REDACTED_IP]",
re.compile(r"\b(?:(?:25[0-5]|2[0-4]\d|1?\d?\d)\.){3}(?:25[0-5]|2[0-4]\d|1?\d?\d)\b"),
0, None,
),
]
# 人类可读的类别中文名，用于打印汇总
CATEGORY_LABELS = {
"private_key": "私钥 / 证书",
"jwt": "JWT 令牌",
"url_credential": "连接串凭据",
"aws_access_key": "AWS 访问密钥",
"github_token": "GitHub 令牌",
"slack_token": "Slack 令牌",
"google_api_key": "Google API Key",
"api_key": "API Key (sk-)",
"bearer_token": "Bearer 令牌",
"secret_assignment": "口令 / 密钥赋值",
"email": "邮箱地址",
"credit_card": "信用卡号",
"iban": "IBAN 银行账号",
"us_ssn": "美国社保号(SSN)",
"cn_id_card": "身份证号",
"cn_phone": "手机号",
"ip_address": "IP 地址",
}
def sanitize(text: str) -> Tuple[str, List[Dict]]:
"""
对文本执行离线规则脱敏。
Returns:
- 脱敏后的文本
- 命中列表，每项为 {category, value, placeholder, start, end}
"""
candidates: List[Dict] = []
for priority, (category, placeholder, pattern, group, validator) in enumerate(_RULES):
for m in pattern.finditer(text):
start, end = m.span(group)
if start < 0: # 该捕获组未参与本次匹配
continue
value = text[start:end]
if validator and not validator(value):
continue
candidates.append({
"category": category,
"placeholder": placeholder,
"value": value,
"start": start,
"end": end,
"priority": priority,
})
# 处理重叠：优先级高（数字小）的规则胜出，避免同一段被重复/错误脱敏
candidates.sort(key=lambda c: (c["priority"], c["start"]))
accepted: List[Dict] = []
for c in candidates:
if any(not (c["end"] <= a["start"] or c["start"] >= a["end"]) for a in accepted):
continue
accepted.append(c)
# 按位置顺序重建脱敏文本
accepted.sort(key=lambda c: c["start"])
parts: List[str] = []
last = 0
for c in accepted:
parts.append(text[last:c["start"]])
parts.append(c["placeholder"])
last = c["end"]
parts.append(text[last:])
findings = [
{k: c[k] for k in ("category", "value", "placeholder", "start", "end")}
for c in accepted
]
return "".join(parts), findings
def summarize(findings: List[Dict]) -> Counter:
"""统计各类别命中次数"""
return Counter(f["category"] for f in findings)
def print_report(name: str, original: str, redacted: str, findings: List[Dict]) -> None:
"""打印单条样本的 before/after 与命中明细"""
print(f"\n{'=' * 64}")
print(f"样本: {name} （命中 {len(findings)} 处敏感信息）")
print("=" * 64)
print("--- 脱敏前 (BEFORE) ---")
print(original.rstrip())
print("\n--- 脱敏后 (AFTER) ---")
print(redacted.rstrip())
if findings:
print("\n--- 命中明细 ---")
for f in findings:
label = CATEGORY_LABELS.get(f["category"], f["category"])
print(f" [{label}] {f['value']} -> {f['placeholder']}")
if __name__ == "__main__":
# 直接运行本模块时，对内置样本做一次快速演示
from samples import SAMPLES
total = Counter()
for name, text in SAMPLES:
redacted, findings = sanitize(text)
print_report(name, text, redacted, findings)
total.update(summarize(findings))
print(f"\n{'=' * 64}")
print("脱敏类别汇总")
print("=" * 64)
for category, count in total.most_common():
label = CATEGORY_LABELS.get(category, category)
print(f" {label:<16} {count} 处")
print(f"\n 合计脱敏 {sum(total.values())} 处敏感信息")
samples.py¶
"""
用于日志脱敏演示的代表性样本
这些样本模拟真实 Agent 运行时最容易泄露敏感信息的几种场景：
工具调用的 HTTP 请求/响应、客服对话、数据库连接报错、CI/Git 日志、
以及配置转储。它们混合了密钥类（API Key、令牌、私钥）与 PII 类
（身份证、手机号、信用卡、邮箱）敏感信息，便于展示脱敏的覆盖面。
注意：以下所有密钥、卡号、证件号均为虚构，仅用于演示，不对应任何真实账户。
"""
SAMPLES = [
(
"工具调用日志 (HTTP 请求/响应)",
"""[2024-05-12 09:14:22] TOOL_CALL http_request
url: https://api.example.com/v1/users/8842
headers: {"Authorization": "Bearer sk-proj-ABCD1234efgh5678IJKL9012mnop3456qrst", "X-Api-Key": "AIzaSyD-EXAMPLEfakeKEY1234567890abcdef12"}
response: {"user_id": 8842, "email": "alice.wang@example.com", "phone": "13912345678"}""",
),
(
"客服对话 (PII 泄露)",
"""USER: 你好，我要办理报销，我的身份证号是 11010119900307721X，手机号 13800138000。
ASSISTANT: 好的，请再提供一下银行卡号以便核对。
USER: 卡号是 4111 1111 1111 1111，另外我的美国社保号是 123-45-6789。
ASSISTANT: 收到，我这就为您登记。""",
),
(
"数据库连接报错 (凭据泄露)",
"""[ERROR] db.connect failed after 3 retries
dsn: postgres://admin:S3cr3t_P4ssw0rd@db.internal:5432/prod
fallback_config: {"db_password": "hunter2xyz", "aws_access_key_id": "AKIAIOSFODNN7EXAMPLE"}
host_ip: 192.168.10.24""",
),
(
"CI / Git 日志 (令牌泄露)",
"""Cloning into 'service-repo'...
remote: using deploy token ghp_16C7e42F292c6912E7710c838347Ae178B4a99
Slack notify webhook token: xoxb-PLACEHOLDERfaketoken000000notarealslacktoken
session jwt=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c""",
),
(
"配置转储 (私钥泄露)",
"""[DEBUG] dumping runtime config
service_account_key: |
-----BEGIN RSA PRIVATE KEY-----
MIIEpAIBAAKCAQEA7QwZbq3vX9kLmN0pQrStUvWxYz1234567890abcdefghijkl
mnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0987654321zyxwvutsrqponm
QIDAQAB
-----END RSA PRIVATE KEY-----
admin_contact: ops-team@example.com""",
),
]
test_loader.py¶
"""
Test Case Loader for User Memory Evaluation Framework
"""
import sys
import json
import subprocess
from pathlib import Path
from typing import Dict, List, Optional, Any
from config import EVAL_FRAMEWORK_PATH
class TestCaseLoader:
"""Load test cases from user-memory-evaluation framework"""
def __init__(self):
self.eval_framework_path = EVAL_FRAMEWORK_PATH
if not self.eval_framework_path.exists():
raise ValueError(f"Evaluation framework not found at {self.eval_framework_path}")
def get_all_test_cases(self) -> List[Dict[str, Any]]:
"""Get all available test cases"""
script = """
import sys
import json
from pathlib import Path
import io
# Suppress rich console output
import rich.console
rich.console.Console = lambda *args, **kwargs: type('FakeConsole', (), {
'print': lambda self, *a, **k: None,
'__getattr__': lambda self, name: lambda *a, **k: None
})()
# Redirect output
old_stdout = sys.stdout
sys.stdout = io.StringIO()
try:
from framework import UserMemoryEvaluationFramework
framework = UserMemoryEvaluationFramework()
test_cases = []
for tc in framework.list_test_cases():
test_cases.append({
'test_id': tc.test_id,
'category': tc.category,
'title': tc.title,
'description': tc.description,
'num_conversations': len(tc.conversation_histories),
'user_question': tc.user_question
})
# Restore stdout for JSON output
sys.stdout = old_stdout
print(json.dumps(test_cases))
except Exception as e:
sys.stdout = old_stdout
print(json.dumps([]))
"""
result = subprocess.run(
[sys.executable, "-c", script],
cwd=self.eval_framework_path,
capture_output=True,
text=True
)
if result.returncode != 0:
print(f"Error getting test cases: {result.stderr}")
return []
try:
return json.loads(result.stdout)
except json.JSONDecodeError:
print(f"Error parsing test cases JSON")
return []
def get_layer3_test_cases(self) -> List[Dict[str, Any]]:
"""Get only Layer 3 test cases (most complex)"""
all_cases = self.get_all_test_cases()
return [tc for tc in all_cases if tc['category'] == 'layer3']
def get_test_case_conversations(self, test_id: str) -> List[Dict[str, Any]]:
"""Get detailed conversation histories for a specific test case"""
script = f"""
import sys
import json
from pathlib import Path
import io
# Redirect stdout to suppress any print statements from framework
old_stdout = sys.stdout
sys.stdout = io.StringIO()
try:
from framework import UserMemoryEvaluationFramework
framework = UserMemoryEvaluationFramework()
tc = framework.get_test_case("{test_id}")
# Restore stdout for our JSON output
sys.stdout = old_stdout
if not tc:
print(json.dumps([]))
else:
conversations = []
for conv in tc.conversation_histories:
conv_data = {{
'conversation_id': conv.conversation_id,
'timestamp': conv.timestamp,
'messages': []
}}
for msg in conv.messages:
msg_data = {{
'role': msg.role.value,
'content': msg.content
}}
# Add metadata if it exists
if hasattr(msg, 'metadata'):
msg_data['metadata'] = msg.metadata
conv_data['messages'].append(msg_data)
conversations.append(conv_data)
print(json.dumps(conversations))
except Exception as e:
import traceback
sys.stdout = old_stdout
sys.stderr.write(traceback.format_exc())
print(json.dumps([]))
"""
result = subprocess.run(
[sys.executable, "-c", script],
cwd=self.eval_framework_path,
capture_output=True,
text=True
)
if result.returncode != 0:
print(f"Error getting conversation histories: {result.stderr}")
return []
try:
return json.loads(result.stdout)
except json.JSONDecodeError as e:
print(f"Error parsing conversation histories JSON: {e}")
if result.stdout:
print(f"stdout (first 500 chars): {result.stdout[:500]}")
if result.stderr:
print(f"stderr (first 500 chars): {result.stderr[:500]}")
return []
def format_conversation_text(self, conversation: Dict[str, Any]) -> str:
"""Format a conversation into readable text"""
lines = []
lines.append(f"Conversation ID: {conversation['conversation_id']}")
lines.append(f"Timestamp: {conversation['timestamp']}")
lines.append("-" * 50)
for msg in conversation['messages']:
role = msg['role'].upper()
content = msg['content']
lines.append(f"{role}: {content}")
lines.append("") # Empty line between messages
return "\n".join(lines)
test_loader_debug.py¶
#!/usr/bin/env python3
"""Debug script to test loading conversations"""
from test_loader import TestCaseLoader
def main():
loader = TestCaseLoader()
# Get all test cases
print("Getting all test cases...")
all_cases = loader.get_all_test_cases()
print(f"Found {len(all_cases)} test cases")
# Get Layer 3 test cases
layer3_cases = loader.get_layer3_test_cases()
print(f"Found {len(layer3_cases)} Layer 3 test cases")
if layer3_cases:
# Try to load the first one
first_case = layer3_cases[0]
print(f"\nTrying to load: {first_case['test_id']}")
conversations = loader.get_test_case_conversations(first_case['test_id'])
if conversations:
print(f"Successfully loaded {len(conversations)} conversations")
# Print first conversation snippet
if conversations[0]['messages']:
print(f"First message: {conversations[0]['messages'][0]['content'][:100]}...")
else:
print("Failed to load conversations")
if __name__ == "__main__":
main()
test_sanitize_conversation.py¶
#!/usr/bin/env python3
"""Regression tests for sanitize_conversation() in agent.py.
Bug: detect_pii() catches any backend exception and returns ([], {}), but
sanitize_conversation then subscripted the empty metrics dict
(perf_metrics['input_tokens']) -> KeyError that killed the whole batch.
Fixed with .get(..., 0) defaults so a dead backend degrades gracefully.
"""
import pytest
from agent import LogSanitizationAgent
from metrics import MetricsCollector
def _make_agent(client, tmp_path):
"""Build an agent without __init__ (which requires a live Ollama)."""
ag = LogSanitizationAgent.__new__(LogSanitizationAgent)
ag.model = "qwen3:0.6b"
ag.backend = "ollama"
ag.metrics_collector = MetricsCollector(tmp_path)
ag.client = client
return ag
CONV = {
"conversation_id": "demo_001",
"messages": [{"role": "user", "content": "My SSN is 123-45-6789."}],
}
class _DeadClient:
def chat(self, **kwargs):
raise ConnectionError("[test] Ollama server is not running")
class _FakeClient:
"""Mimics ollama.Client.chat(stream=True) chunk shape."""
def __init__(self, payload: str):
self.payload = payload
def chat(self, **kwargs):
return [{"message": {"content": self.payload}}]
def test_dead_backend_returns_result_with_zero_metrics(tmp_path):
ag = _make_agent(_DeadClient(), tmp_path)
result = ag.sanitize_conversation(CONV, "t1") # must not raise
assert result["pii_found"] == []
assert result["replacements_made"] == 0
assert result["metrics"]["input_tokens"] == 0
assert result["metrics"]["pii_items_found"] == 0
def test_working_backend_still_detects_pii(tmp_path):
ag = _make_agent(_FakeClient('{"pii_values": ["123-45-6789"]}'), tmp_path)
result = ag.sanitize_conversation(CONV, "t1")
assert result["pii_found"] == ["123-45-6789"]
assert result["replacements_made"] == 1
assert "[REDACTED]" in result["sanitized_text"]
assert result["metrics"]["pii_items_found"] == 1