File size: 5,845 Bytes
076b5c2 85e421f 076b5c2 85e421f 076b5c2 85e421f 076b5c2 daad250 076b5c2 acdfa32 076b5c2 146b9b4 076b5c2 acdfa32 076b5c2 | 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 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 | import os
import re
import ast
import sys
import yaml
import json
import logging
import pandas as pd
from typing import Dict
from pathlib import Path
def load_system_prompt(system_prompt_file):
if system_prompt_file and os.path.exists(system_prompt_file):
with open(system_prompt_file, "r", encoding="utf-8") as f:
text = f.read()
match = re.search(r"```\n?(.*?)```", text, re.DOTALL)
if match:
return match.group(1).strip()
return text.strip()
else:
raise FileNotFoundError(f"System prompt file not found: {system_prompt_file}")
def load_config(config_path: str = None) -> dict:
if config_path is None:
config_path = Path(__file__).resolve().parent.parent / "configs/config.yaml"
if not Path(config_path).exists():
raise FileNotFoundError(
f"Config file not found: {config_path}\n"
)
with open(config_path, "r", encoding="utf-8") as f:
cfg = yaml.safe_load(f) or {}
if not isinstance(cfg, dict):
raise ValueError(
f"Config file must contain a YAML mapping (dict), got {type(cfg).__name__}. "
)
# Normalize None-valued sections to empty dicts so chained .get() never
# fails with AttributeError (e.g. `llm:` with no sub-keys → None).
# Recurse into nested dicts so `paths:\n vector_db:` also gets normalized.
def _normalize_nulls(d):
for key in list(d.keys()):
if d[key] is None:
d[key] = {}
elif isinstance(d[key], dict):
_normalize_nulls(d[key])
_normalize_nulls(cfg)
return cfg
def parse_llm_json(raw_text):
if not raw_text or not raw_text.strip():
return None, "Empty response"
text = raw_text.strip()
# Strip markdown fences
fence_match = re.search(r"```(?:json)?\s*\n?(.*?)```", text, re.DOTALL)
if fence_match:
text = fence_match.group(1).strip()
# Attempt 1: Standard JSON parsing (double quotes)
try:
return json.loads(text), None
except json.JSONDecodeError:
pass
# Attempt 2: Extract JSON object from surrounding text
brace_match = re.search(r"\{.*\}", text, re.DOTALL)
if brace_match:
extracted = brace_match.group(0)
try:
return json.loads(extracted), None
except json.JSONDecodeError:
pass
# Attempt 3: Python dict syntax (single quotes) — common with Llama models.
# ast.literal_eval safely parses Python literals including single-quoted dicts.
# This handles: {'commodity': 'Coffee', 'intent': 'Crop Nutrition'}
# and also apostrophes: {'reasoning': "The farmer's crop..."}
try:
result = ast.literal_eval(extracted)
if isinstance(result, dict):
return result, None
except (ValueError, SyntaxError):
pass
return None, f"Could not parse JSON from: {text[:200]}"
def setup_logging(output_dir, log_file):
log_path = os.path.join(output_dir, log_file)
# 1. Create a custom module-level logger
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO) # Capture everything at the logger level
# 2. Create a handler to write logs to a file
file_handler = logging.FileHandler(log_path, mode="a")
file_handler.setLevel(logging.INFO) # File only gets errors and criticals
# 3. Create a handler to output logs to the console
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO) # Console gets Info, Warning, Error, Critical
# 4. Define a reusable string format for your log outputs
formatter = logging.Formatter('%(asctime)s | %(name)s | %(levelname)s | %(message)s')
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)
# 5. Attach the configured handlers to your custom logger
logger.addHandler(file_handler)
logger.addHandler(console_handler)
return logger
def validate_classification(parsed, classification_fields:Dict):
errors = []
if parsed is None:
return False, ["Parsed result is None"]
for field in classification_fields.keys():
if field not in parsed:
errors.append(f"Missing field: {field}")
continue
if parsed.get(field) not in classification_fields[field]:
errors.append(f"Invalid {field}: {parsed.get(field)}")
return len(errors) == 0, errors
def load_checkpoint(cfg, logger):
ckpt_path = os.path.join(cfg.output_dir, cfg.checkpoint_file)
if os.path.exists(ckpt_path):
df = pd.read_parquet(ckpt_path)
logger.info(f"RESUMING from checkpoint: {len(df)} queries already classified")
return df, len(df)
return None, 0
def save_checkpoint(df, cfg, logger):
ckpt_path = os.path.join(cfg.output_dir, cfg.checkpoint_file)
df.to_parquet(ckpt_path, index=False)
logger.info(f"Checkpoint saved: {len(df)} queries")
def load_queries(cfg, logger):
# --- Load input data ---
logger.info(f"Loading data from {cfg.input_path}")
df_input = pd.read_csv(cfg.input_path) if cfg.input_path.split('.')[-1] == 'csv' else pd.read_parquet(cfg.input_path)
size = cfg.test_size if cfg.test_size>=0 else df_input.shape[0]
if cfg.shuffle:
df_input = df_input.sample(n=size, random_state=42).reset_index(drop=True)
queries = df_input[cfg.query_column].tolist()
if cfg.classification_type==2:
assert cfg.response_column is not None, 'Adoption phase analysis requires LLM responses. Specify response column.'
responses = df_input[cfg.response_column].tolist()
queries = ['\n\n'.join([f"## Query\n\n{query}", f"### Response\n\n{response}"]) for query, response in zip(queries, responses)]
return df_input, queries
|