import os import sys import json import argparse import pandas as pd from typing import Dict, Any from openai import OpenAI # ============================================================ # 1. Ontology (HARD CONSTRAINTS) # ============================================================ DECOMPOSITION_SCHEMA = { "name": "jtbd_decomposition", "schema": { "type": "object", "additionalProperties": False, "properties": { "lost_capability": { "type": "string", "enum": [ "information_control", "attention", "context_continuity", "progress_visibility", "cognitive_load", ], }, "failure_mode": { "type": "string", "maxLength": 120, }, "pressure_source": { "type": "string", "enum": [ "volume", "interruptions", "complexity", "poor_tooling", "time_pressure", "uncertainty", ], }, }, "required": [ "lost_capability", "failure_mode", "pressure_source", ], }, } ALLOWED_CAPABILITIES = { "information_control", "attention", "context_continuity", "progress_visibility", "cognitive_load", } ALLOWED_PRESSURE_SOURCES = { "volume", "interruptions", "complexity", "poor_tooling", "time_pressure", "uncertainty", } CAPABILITY_TO_CLUSTER = { "information_control": "loss_of_information_control", "attention": "context_fragmentation", "context_continuity": "context_fragmentation", "progress_visibility": "visibility_gaps", "cognitive_load": "cognitive_overload", } # ============================================================ # 2. OpenAI Client # ============================================================ def init_openai_client() -> OpenAI: api_key = os.getenv("OPENAI_API_KEY") if not api_key: raise RuntimeError("OPENAI_API_KEY not found in environment variables") return OpenAI(api_key=api_key) # ============================================================ # 3. Prompt (STRICT) # ============================================================ SYSTEM_PROMPT = """You are a semantic decomposition engine. Your task is NOT to explain problems, suggest solutions, or create insights. Your task is to identify which HUMAN CAPABILITY is being degraded. Rules: - Choose EXACTLY ONE capability - Do NOT invent new values - Do NOT explain reasoning - Output must conform to the provided JSON schema """ USER_PROMPT_TEMPLATE = """JTBD description: SITUATION: {situation} STRUGGLE: {struggle} """ # ============================================================ # 4. Normalization # ============================================================ def normalize_text(text: str) -> str: if not isinstance(text, str): return "" text = text.lower().strip() for filler in ["i want to", "so i can", "in order to"]: text = text.replace(filler, "") return text # ============================================================ # 5. LLM Call # ============================================================ def call_llm_for_decomposition( client: OpenAI, situation: str, struggle: str, model: str = "gpt-4.1-mini", ) -> Dict[str, Any]: user_prompt = USER_PROMPT_TEMPLATE.format( situation=situation, struggle=struggle, ) response = client.chat.completions.create( model=model, temperature=0, response_format={ "type": "json_schema", "json_schema": DECOMPOSITION_SCHEMA, }, messages=[ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_prompt}, ], ) # Unele versiuni OpenAI => JSON este în .message.content (string) raw = response.choices[0].message.content # Curățăm textul raw = raw.strip() # Parsăm JSON-ul (validat deja de schema API) try: return json.loads(raw) except Exception as e: print("JSON parse error:", raw) raise e # ============================================================ # 6. Validation # ============================================================ def validate_output(data: Dict[str, Any]) -> bool: if not isinstance(data, dict): return False if data.get("lost_capability") not in ALLOWED_CAPABILITIES: return False if data.get("pressure_source") not in ALLOWED_PRESSURE_SOURCES: return False if not isinstance(data.get("failure_mode"), str): return False return True # ============================================================ # 7. Deterministic Clustering # ============================================================ def assign_cluster(lost_capability: str) -> str: return CAPABILITY_TO_CLUSTER.get(lost_capability, "unknown") # ============================================================ # 8. Main Processing Pipeline # ============================================================ def process_dataframe(df: pd.DataFrame, client: OpenAI, output_path: str) -> pd.DataFrame: temp_path = output_path.replace(".csv", "_resume.csv") # Dacă există un fișier resume, îl încărcăm și continuăm if os.path.exists(temp_path): print(f"Resuming from: {temp_path}") done_df = pd.read_csv(temp_path) start_index = len(done_df) else: done_df = pd.DataFrame() start_index = 0 rows = [] # Dacă retake, păstrăm deja rezultatele if start_index > 0: rows = done_df.to_dict("records") for i in range(start_index, len(df)): row = df.iloc[i] situation = normalize_text(row.get("situation", "")) struggle = normalize_text(row.get("struggle", "")) try: llm_result = call_llm_for_decomposition( client=client, situation=situation, struggle=struggle, ) except Exception as e: print(f"Error on row {i}, continuing… {e}") llm_result = None if not llm_result or not validate_output(llm_result): result = { "lost_capability": None, "failure_mode": None, "pressure_source": None, "cluster": "invalid", } else: cluster = assign_cluster(llm_result["lost_capability"]) result = { "lost_capability": llm_result["lost_capability"], "failure_mode": llm_result["failure_mode"], "pressure_source": llm_result["pressure_source"], "cluster": cluster, } # salvăm progresul rând cu rând rows.append(result) pd.DataFrame(rows).to_csv(temp_path, index=False) print(f"[{i+1}/{len(df)}] saved") final_df = pd.concat([df.reset_index(drop=True), pd.DataFrame(rows)], axis=1) return final_df # ============================================================ # 9. CLI Entry Point # ============================================================ def main(): parser = argparse.ArgumentParser( description="JTBD Semantic Understanding Pipeline" ) parser.add_argument( "input_file", help="Path to input CSV file containing JTBD data", ) args = parser.parse_args() input_path = args.input_file if not os.path.exists(input_path): raise FileNotFoundError(f"Input file not found: {input_path}") output_path = ( os.path.splitext(input_path)[0] + "_semantic_understanding.csv" ) print(f"Loading input file: {input_path}") df = pd.read_csv(input_path) client = init_openai_client() print("Running semantic decomposition pipeline...") enriched_df = process_dataframe(df, client, output_path) # când se termină, mutăm resume în fișierul final temp_path = output_path.replace(".csv", "_resume.csv") enriched_df.to_csv(output_path, index=False) if os.path.exists(temp_path): os.remove(temp_path) print(f"Saved final output to: {output_path}") if __name__ == "__main__": main()