Spaces:
Sleeping
Sleeping
File size: 3,075 Bytes
a91323c | 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 | from __future__ import annotations
import json
from datetime import datetime, UTC
import os
from pathlib import Path
from dotenv import load_dotenv
ROOT = Path(__file__).resolve().parents[1]
DATA_DIR = ROOT / "data"
OUTPUTS_DIR = ROOT / "outputs"
OUTPUTS_DIR.mkdir(parents=True, exist_ok=True)
def main() -> int:
# Ensure project root is on sys.path for 'src' imports
import sys
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
# Load environment variables from .env if present (explicit path to avoid edge cases)
load_dotenv(ROOT / ".env")
# Maintenance/disable flag
if os.getenv("DISABLE_ALL_SERVICES", "").lower() in {"1", "true", "yes"}:
print("All services are disabled by administrator (DISABLE_ALL_SERVICES).")
return 0
from src.models import load_actions, load_strategies, StrategicObjective
from src.alignment import AlignmentEngine
from src.recommendations import generate_recommendations
from src.rag_engine import RAGEngine
strategies = load_strategies(DATA_DIR / "strategic.json")
actions = load_actions(DATA_DIR / "action.json")
engine = AlignmentEngine()
result = engine.align(strategies=strategies, actions=actions, top_k=5)
# Try RAG if key exists; fallback to rule-based recommendations
rag = RAGEngine()
rag_out_per_strategy = []
for r in result["strategy_results"]:
# We only have title in result; pull description from original object for a better prompt
s_obj = next(
(s for s in strategies if s.id == r["strategy_id"]),
StrategicObjective(
id=r["strategy_id"], title=r["strategy_title"], description="", kpis=[]
),
)
rag_json = rag.generate(
strategy=s_obj,
current_score=float(r.get("avg_top3_similarity", 0.0)),
retrieved_actions=[
{
"title": m.get("title"),
"owner": m.get("owner"),
"similarity": float(m.get("similarity", 0.0)),
}
for m in r.get("top_matches", [])
],
)
rag_out_per_strategy.append(
{
"strategy_id": r["strategy_id"],
"strategy_title": r["strategy_title"],
"alignment_label": r["alignment_label"],
"rag": rag_json,
}
)
# Rule-based for comparison
recs = generate_recommendations(result)
payload = {
"result": result,
"rag_recommendations": rag_out_per_strategy,
"recommendations": recs,
}
timestamp = datetime.now(UTC).strftime("%Y%m%d-%H%M%S")
out_path = OUTPUTS_DIR / f"alignment_cli_{timestamp}.json"
out_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
print(f"Overall Score: {result['overall_score']:.2f}")
print(f"Coverage %: {result['coverage_percent']:.2f}")
print(f"Saved output: {out_path}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
|