Spaces:
Running
Running
File size: 3,108 Bytes
24f95f0 | 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 | import asyncio
import json
import logging
import os
import sys
# Configure logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s — %(message)s")
logger = logging.getLogger("VERIFY")
# Add backend and app to path
sys.path.append(os.path.join(os.getcwd(), "backend"))
sys.path.append(os.path.join(os.getcwd(), "backend", "app"))
async def verify():
logger.info("--- JANUS E2E VERIFICATION START ---")
try:
from app.graph import get_compiled_graph
from app.config import HUGGINGFACE_API_KEY, HUGGINGFACE_MODEL, PRIMARY_PROVIDER
logger.info(f"Configuration: Primary={PRIMARY_PROVIDER}, Model={HUGGINGFACE_MODEL}")
logger.info(f"HF Token present: {bool(HUGGINGFACE_API_KEY)}")
graph = get_compiled_graph()
# Test Query
user_input = "Provide a deep analysis of the current AI chip market competition, incorporating your formed opinions on energy efficiency."
initial_state = {
"user_input": user_input,
"complexity": "high",
"replan_count": 0,
"context": {
"self_reflection": {
"opinions": [
{"topic": "energy efficiency", "statement": "Energy efficiency is no longer just a perk; it is the primary bottleneck for massive-scale AI inference, favoring custom silicon like TPUs and LPUs over generic GPUs."}
]
}
}
}
logger.info("Executing Graph (this may take a minute surring HF)...")
final_state = await graph.ainvoke(initial_state)
logger.info("Graph Execution COMPLETE.")
# Validation
result = final_state.get("final", {})
response = result.get("response", "")
print("\n" + "="*50)
print("VERIFICATION RESULTS")
print("="*50)
if "<think>" in response:
print("[PASS] Reasoning Transparency (<think> block) detected.")
else:
print("[FAIL] Reasoning Transparency (<think> block) MISSING.")
if "I " in response or "My " in response or "me " in response.lower():
print("[PASS] Unified Persona (First-person pronouns) detected.")
else:
print("[FAIL] Unified Persona (First-person pronouns) MISSING.")
research = final_state.get("research", {})
if research:
print(f"[PASS] Research node contributed findings (Sources: {len(research.get('sources', []))}).")
else:
print("[FAIL] Research node seems empty.")
print("-" * 30)
print("SAMPLE RESPONSE PREVIEW:")
if response:
print(response[:800] + "...")
else:
print("EMPTY RESPONSE!")
print("-" * 30)
except Exception as e:
logger.error(f"Verification FAILED with error: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
asyncio.run(verify())
|