-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_async_eval.py
More file actions
127 lines (100 loc) · 3.67 KB
/
Copy pathtest_async_eval.py
File metadata and controls
127 lines (100 loc) · 3.67 KB
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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
#!/usr/bin/env python3
"""Quick test script to validate async evaluation functionality.
This script tests that async evaluation works without requiring a full evaluation run.
"""
import asyncio
import sys
from pathlib import Path
# Add project root to path
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root / "src"))
from unittest.mock import MagicMock # noqa: E402
from langchain_core.messages import AIMessage, HumanMessage # noqa: E402
from rag_agent.agent import RAGAgent # noqa: E402
from rag_agent.evaluation.datasets.synthetic_loader import SyntheticDatasetLoader # noqa: E402
from rag_agent.evaluation.pipelines.evaluation_pipeline import EvaluationPipeline # noqa: E402
async def test_async_evaluation():
"""Test that async evaluation pipeline works."""
print("Testing async evaluation functionality...")
# Create a mock agent
mock_agent = MagicMock(spec=RAGAgent)
mock_result = {
"messages": [
HumanMessage(content="Test question"),
AIMessage(content="Test answer"),
],
"__execution_trace__": None,
}
# Mock async invoke
async def async_invoke(*args, **kwargs):
return mock_result
mock_agent.ainvoke = async_invoke
# Create pipeline
pipeline = EvaluationPipeline(agent=mock_agent, enable_ragas=False)
dataset = SyntheticDatasetLoader()
# Test async evaluation
result = await pipeline.async_run_evaluation(
dataset=dataset,
experiment_name="async_test",
max_examples=3,
max_concurrent_queries=2,
)
print("✅ Async evaluation completed successfully")
print(
f" Processed {result.get('num_queries', result.get('config', {}).get('num_examples', 'unknown'))} examples"
)
print(f" Experiment ID: {result.get('experiment_id', 'N/A')}")
return True
def test_parallel_evaluation():
"""Test that parallel evaluation works."""
print("\nTesting parallel evaluation functionality...")
# Create a mock agent
mock_agent = MagicMock(spec=RAGAgent)
mock_result = {
"messages": [
HumanMessage(content="Test question"),
AIMessage(content="Test answer"),
],
"__execution_trace__": None,
}
mock_agent.invoke.return_value = mock_result
# Create pipeline
pipeline = EvaluationPipeline(agent=mock_agent, enable_ragas=False)
dataset = SyntheticDatasetLoader()
# Test parallel evaluation
result = pipeline.run_evaluation(
dataset=dataset,
experiment_name="parallel_test",
max_examples=5,
max_concurrent_queries=2,
)
print("✅ Parallel evaluation completed successfully")
print(
f" Processed {result.get('num_queries', result.get('config', {}).get('num_examples', 'unknown'))} examples"
)
print(f" Agent invoke called {mock_agent.invoke.call_count} times")
return True
if __name__ == "__main__":
print("=" * 70)
print("ASYNC/PARALLEL EVALUATION VALIDATION")
print("=" * 70)
# Test parallel execution
try:
test_parallel_evaluation()
except Exception as e:
print(f"❌ Parallel evaluation test failed: {e}")
sys.exit(1)
# Test async execution
try:
asyncio.run(test_async_evaluation())
except Exception as e:
print(f"❌ Async evaluation test failed: {e}")
sys.exit(1)
print("\n" + "=" * 70)
print("✅ ALL TESTS PASSED")
print("=" * 70)
print("\nTo run a full evaluation with async enabled:")
print(" export EVAL_USE_ASYNC=true")
print(" export EVAL_MAX_WORKERS=4")
print(" export OPENAI_API_KEY=your_key")
print(" python scripts/eval_defaults.py")