mirror of
https://github.com/arkorty/B.Tech-Project-III.git
synced 2026-04-19 12:41:48 +00:00
78 lines
3.7 KiB
Python
78 lines
3.7 KiB
Python
"""Test Milestone 5: Pattern detection + Cross-group analysis."""
|
|
import asyncio, os, sys
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
|
|
|
from backend.pipeline import process_message_batch
|
|
|
|
# Dev team messages with PLANTED patterns
|
|
DEV_MSGS = [
|
|
{"sender": "Alex", "text": "Let's go with PostgreSQL.", "timestamp": "2026-03-15T10:00:00Z"},
|
|
{"sender": "Raj", "text": "I'll handle the payment module Stripe integration.", "timestamp": "2026-03-15T11:00:00Z"},
|
|
{"sender": "Raj", "text": "Payment webhook setup is done, only I know how this works right now.", "timestamp": "2026-03-16T10:00:00Z"},
|
|
{"sender": "Sam", "text": "Timeout error on checkout again.", "timestamp": "2026-03-17T09:00:00Z"},
|
|
{"sender": "Sam", "text": "Same timeout error. This is the third time.", "timestamp": "2026-03-18T09:00:00Z"},
|
|
{"sender": "Alex", "text": "I'm hardcoding the config for now, no time to do it properly.", "timestamp": "2026-03-18T14:00:00Z"},
|
|
{"sender": "Sam", "text": "We need the design specs for the dashboard. Still waiting.", "timestamp": "2026-03-19T10:00:00Z"},
|
|
{"sender": "Alex", "text": "Dashboard is completely blocked without those design specs.", "timestamp": "2026-03-20T10:00:00Z"},
|
|
]
|
|
|
|
# Product team messages — NOTE: no mention of design specs being needed
|
|
PRODUCT_MSGS = [
|
|
{"sender": "Lisa", "text": "Dark mode is the most requested feature by far.", "timestamp": "2026-03-16T10:00:00Z"},
|
|
{"sender": "Mike", "text": "We should go mobile-first this sprint.", "timestamp": "2026-03-17T10:00:00Z"},
|
|
{"sender": "Sarah", "text": "API stability is more important than mobile. Enterprise clients are complaining.", "timestamp": "2026-03-17T10:30:00Z"},
|
|
{"sender": "Lisa", "text": "I told the client we'd have the dashboard demo ready by Friday.", "timestamp": "2026-03-18T10:00:00Z"},
|
|
{"sender": "Mike", "text": "Let's push for the API-first approach this quarter.", "timestamp": "2026-03-19T10:00:00Z"},
|
|
]
|
|
|
|
async def main():
|
|
from backend.agents.pattern_detector import detect_patterns
|
|
from backend.agents.cross_group_analyst import analyze_cross_group
|
|
|
|
dev_group = "test_dev_m5"
|
|
product_group = "test_product_m5"
|
|
|
|
# Process both groups
|
|
print("Processing dev team messages...")
|
|
dev_signals = await process_message_batch(dev_group, DEV_MSGS)
|
|
print(f" ✅ Dev team: {len(dev_signals)} signals stored")
|
|
|
|
print("Processing product team messages...")
|
|
prod_signals = await process_message_batch(product_group, PRODUCT_MSGS)
|
|
print(f" ✅ Product team: {len(prod_signals)} signals stored")
|
|
|
|
# Test pattern detection
|
|
print("\nRunning pattern detection on dev team...")
|
|
patterns = await detect_patterns(dev_group)
|
|
print(f" Found {len(patterns)} patterns:")
|
|
for p in patterns:
|
|
print(f" [{p.severity}] {p.type}: {p.description[:80]}")
|
|
print(f" ✅ Pattern detection working")
|
|
|
|
# Test cross-group analysis
|
|
print("\nRunning cross-group analysis...")
|
|
from backend.db.chroma import get_all_signals
|
|
summaries = {
|
|
"Acme Dev Team": get_all_signals(dev_group),
|
|
"Acme Product": get_all_signals(product_group),
|
|
}
|
|
insights = await analyze_cross_group(summaries)
|
|
print(f" Found {len(insights)} cross-group insights:")
|
|
for i in insights:
|
|
print(f" [{i.severity}] {i.type}: {i.description[:100]}")
|
|
print(f" ✅ Cross-group analysis working")
|
|
|
|
# Cleanup
|
|
import chromadb
|
|
from backend.config import CHROMA_DB_PATH
|
|
client = chromadb.PersistentClient(path=CHROMA_DB_PATH)
|
|
for name in [dev_group, product_group]:
|
|
try:
|
|
client.delete_collection(f"ll_{name}")
|
|
except:
|
|
pass
|
|
|
|
print("\n🎉 MILESTONE 5 PASSED — Pattern detection + cross-group analysis working")
|
|
|
|
asyncio.run(main())
|