self.persona_history = []
def _initialize_config(self):
"""Private method to load configurations."""
# Load environment variables
HUGGINGFACE_TOKEN = os.getenv("HUGGINGFACE_TOKEN")
if not HUGGINGFACE_TOKEN:
raise ValueError("HUGGINGFACE_TOKEN is not set! Make sure to define it in .env.")
with open("config_article.yml", "r", encoding="utf-8") as file:
config = yaml.safe_load(file)
self.parameters = config["parameters"]
self.persona_steps = config["persona_workflow"]
self.prompts = config["prompts"]
self.rag_queries = config["rag"]
# Initialize Hugging Face LLM
self.llm = HuggingFaceAPI(api_url=config["API"]["model_url"], api_token=HUGGINGFACE_TOKEN)
# Initialize RAG Retriever
embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
self.rag_retriever = RagRetrieveWithMeta(
config['rag']['scientific']['faiss'],
config['rag']['scientific']['embeddings'],
config['rag']['diary']['faiss'],
config['rag']['diary']['embeddings'],
config['rag']['weather']['faiss'],
config['rag']['weather']['embeddings'],
config['rag']['insights']['faiss'],
config['rag']['insights']['embeddings']
# Initialize Memory for Chat History
self.memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
self.persona_history = []
def save_history(self, history):
file_path = f"chat_history_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
# Convert history to structured JSON format
"persona_history": self.persona_history, # 🔹 Now includes both persona and the refinement prompt used
"chat_history": [{"--HUMAN--": msg[0], "--TREE--": msg[1]} for msg in history],
"process_logs": self.process_logs
# Save the JSON data to a file
with open(file_path, "w", encoding="utf-8") as file:
json.dump(history_json, file, indent=4)
return file_path # Returns the file path for download
def get_step_inputs(self, step_inputs):
Fetches and formats the required inputs for a step.
- Uses dictionary-style input definitions from `parameters` in `config.yaml`.
- If an input isn't found, it checks `generated_inputs` for dynamically created values.
# Check in parameters first
for category in self.parameters.values(): # `parameters` now has grouped dictionaries
input_data[item] = category[item]
# If not found, check in dynamically generated inputs
if item in self.generated_inputs:
input_data[item] = self.generated_inputs[item]
# ======== FUNCTION to store and return llm outputs ===============
# =================================================================
def store_result_and_return(self, response, _output=None):
# 🛑 Debug: Check if step_output is set correctly
raise ValueError("⚠️ ERROR: `step_output` is missing in store_result_and_return()!")
# ✅ Store response in generated_inputs dictionary
self.generated_inputs[_output] = response
print(f"\n📝 Stored Result -> `{_output}`:\n{response}\n")
return {_output: response}
# ======== FUNCTION to format llm response ========================
# =================================================================
# def extract_response(self, response: str):
# # Use regex to find everything AFTER '</think>'
# match = re.search(r"</think>\s*(.*)", response, re.DOTALL)
# # If there's a match, return only the actual response (after '</think>')
# return match.group(1).strip() if match else response
def generate_persona_step(self, _step_index):
_inputs = self.persona_steps[_step_index]["inputs"]
_output = self.persona_steps[_step_index]["output"]
_prompt = self.prompts[self.persona_steps[_step_index]["prompt"]]["template"]
required_inputs = self.get_step_inputs(_inputs)
formatted_prompt = _prompt.format(**required_inputs)
response = self.llm.invoke(formatted_prompt)
clean_response = self.extract_response(response)
self.persona_history.append({
return self.store_result_and_return(clean_response, _output=_output)
def get_refinement_prompt(self):
return self.prompts[self.persona_steps[1]["prompt"]]["template"]
def refine_persona(self):
persona_data = self.generate_persona_step(1)
return persona_data["refined_persona"]
# ===== QUESTION CLASSIFICATION ==========================
# ========================================================
# Function to format response - eliminate reasoning
def extract_response(self,response: str):
# Use regex to find everything AFTER '</think>'
match = re.search(r"</think>\s*(.*)", response, re.DOTALL)
# If there's a match, return only the actual response (after '</think>')
return match.group(1).strip() if match else response
# Function to classify question
def classify_question(self,question, classifier_prompt_template):
# classifier_prompt_template = prompts["hop_classification"]["template"]
prompt = classifier_prompt_template.format(question=question)
classification_raw = self.extract_response(self.llm.invoke(prompt).strip().lower())
classification = self.extract_classification(classification_raw)
results = [c.strip() for c in classification.split(",")]
def extract_classification(self, classification_raw):
if "historical" in classification_raw:
classification.append("historical")
if "scientific" in classification_raw:
classification.append("scientific")
if "weather" in classification_raw:
classification.append("weather")
if "insights" in classification_raw:
classification.append("insights")
if "no_rag" in classification_raw:
classification.append("no_rag")
results = ", ".join(classification) #[c.strip() for c in classification.split(",")]
# ===== RAG RETRIEVAL ====================================
# ========================================================
# Function to Handle Multi-Hop RAG Queries
def retrieve_information(self, query, classifications):
if "no_rag" in classifications:
return {},{} # Skip retrieval
retrieved_docs = {} # ✅ Store results by category
for mode in classifications:
print(f"----> current mode: {mode}")
retrieved_texts = self.rag_retriever.rag_wrapper(query, mode=mode)
plain_texts = [doc['content'] for doc in retrieved_texts]
metadata = [doc['metadata'] for doc in retrieved_texts]
query_segment = " ".join(query[:5])
key = query_segment.replace(" ", "_")
retrieved_docs[mode] = plain_texts # ✅ Organized by mode
retrieved_metadata[mode] = metadata
return retrieved_docs, retrieved_metadata
# ===== GENERATE TREE RESPONSE ===========================
# ========================================================
def generate_tree_response(self,instruction, question, history):
# init dictionary to hold full llm process logs
print (instruction + self.prompts['tree_response']['template'])
# Add question to process log
process_log['journalist_question'] = question
# Step 1: Classify Question (Possibly Multi-Hop)
classifications = self.classify_question(question, self.prompts["hop_classification"]["template"])
print(f"1. CLASSIFICATION -- {classifications}")
process_log['classification']=classifications
# Step 2: Retrieve Relevant Knowledge (Categorized)
retrieved_info, retrieved_meta = self.retrieve_information(question, classifications)
print(f"2. RAG RESULTS -- {retrieved_info}")
print(f"2. RAG METADATA -- {retrieved_meta}")
process_log['retrieved_data'] = retrieved_info
process_log['retrieved_metadata'] = retrieved_meta
# Step 3: Get Chat History for Context
chat_history = self.memory.load_memory_variables({})["chat_history"]
# Step 4: Tree Generates Thoughts Before Answering (Always Challenges & Asks a Question)
tree_thoughts = self.generate_tree_thoughts(instruction, question, retrieved_info, chat_history)
print(f"3. INTERNAL THOUGHTS -- {tree_thoughts}")
process_log['tree_thoughts'] = tree_thoughts
tree_response_prompt_template = instruction + self.prompts['tree_response']['template']
# Step 5: Format Final Response Prompt
formatted_prompt = tree_response_prompt_template.format(
chat_history=chat_history,
retrieved_info="\n".join(retrieved_info) if retrieved_info else "No external knowledge needed.",
tree_thoughts=tree_thoughts
# Step 6: Invoke HuggingFaceAPI for Response
raw = self.llm.invoke(formatted_prompt)
print(f"================== RAW DEEPSEEK ANSWER - {raw}")
response = self.extract_response(self.llm.invoke(formatted_prompt).strip().lower())
print(f"4. RAW RESPONSE -- {response}")
process_log['tree_response'] = response
# add process log to history
self.process_logs.append({
f"_process_step_{self.process_step}": process_log
# print(f"TREE: {stylized_response}")
return history + [(question, response)], "" #response
def generate_tree_thoughts(self,instruction,question, retrieved_info, chat_history):
"""Before answering, the tree decides how to challenge and what question to ask."""
#instruction = self.parameters['general_instruction']
thought_prompt_template = instruction + self.prompts['tree_thought']['template']
formatted_thought_prompt = thought_prompt_template.format(
chat_history=chat_history,
retrieved_info=retrieved_info,
print(formatted_thought_prompt)
# ✅ Now we correctly invoke the LLM using the full thought template
tree_thoughts = self.extract_response(
self.llm.invoke(formatted_thought_prompt).strip().lower()) # llm.invoke(thought_prompt)
"""Resets the conversation history in memory."""
self.persona_history = []
return [] # Return empty history to reset UI