File size: 2,924 Bytes
d1f3f31 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | import sys
from pathlib import Path
import time
import json
import os
import asyncio
# Ensure the root directory is in the python path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
# Load local envs manually to ensure API keys are loaded
from src.api.server import load_env_file
load_env_file(ROOT / ".env.local")
load_env_file(ROOT / ".env")
from src.aspect_sentiment import AspectSentimentEngine
from src.api.server import predict_with_trained_model, fallback_extraction
from src.aspect_sentiment.llama_extraction import process_text
def main():
data_dir = ROOT / "data" / "raw"
engine = AspectSentimentEngine()
results = []
total_latency = 0
print(f"Testing 10 conversations using Llama model: {engine.llama_model}\n")
for i in range(1, 11):
filename = f"conv_{i:03d}.txt"
file_path = data_dir / filename
if not file_path.exists():
print(f"Skipping {filename} - not found")
continue
text = file_path.read_text(encoding="utf-8", errors="replace")
start = time.perf_counter()
# Analyze using engine
try:
# We use analyze_text for end-to-end testing
response = asyncio.run(engine.analyze_text(
text=text,
source_name=filename,
source_type="text",
language="en",
transcription_confidence=None,
whisper_model=None,
pipeline=[],
processing_ms=0
))
latency = time.perf_counter() - start
total_latency += latency
summary = response.summary
conversion = response.conversionScore
res = {
"file": filename,
"latency_s": round(latency, 2),
"dominant_sentiment": summary.dominant,
"conversion_prediction": conversion.label if conversion else "N/A",
"conversion_prob": f"{conversion.probability:.2f}" if conversion else "N/A",
"total_features_extracted": summary.totalProducts,
"top_products": [p.name for p in response.products[:3]],
"status": "Success"
}
except Exception as e:
res = {
"file": filename,
"status": f"Failed: {str(e)}"
}
results.append(res)
print(f"Processed {filename} in {res.get('latency_s', 0)}s - Conversion: {res.get('conversion_prediction', 'N/A')}")
print(f"\nAverage Latency: {total_latency / 10:.2f}s")
output_path = ROOT / "test_10_results.json"
output_path.write_text(json.dumps(results, indent=2), encoding="utf-8")
print(f"Full results written to {output_path}")
if __name__ == '__main__':
main()
|