Further enhance and add in a more super advanced coding chain-of-thought with few-shots and a more sophisticated AI workers. And also add in an advanced meta-learning w/a more lengthy response output, and also a more advanced thought processing.

```
import g4f
from g4f.client import Client
import logging
from logging.handlers import RotatingFileHandler
import pyperclip # Library to copy text to the clipboard
import os
import json
from typing import List, Dict
import uuid
import faiss
import numpy as np
from sentence_transformers import SentenceTransformer
import asyncio
import gradio as gr

# Set up logging with rotation
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
handler = RotatingFileHandler("assistant.log", maxBytes=5*1024*1024, backupCount=2)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)

# Initialize the GPT client with the desired provider
client = Client(provider=g4f.Provider.Blackbox)

# File paths for conversation history, memory, and FAISS index
CONVERSATION_LOG_FILE = "conversation_log.json"
MEMORY_FILE = "memory.json"
FAISS_INDEX_FILE = "faiss.index"
MEMORY_METADATA_FILE = "memory_metadata.json"

# Initialize conversation history
messages: List[Dict[str, str]] = []

# Initialize memory
memory: Dict[str, List[Dict[str, str]]] = {}

# Initialize reasoning state
reasoning_state: Dict[str, Dict] = {}

# Initialize Sentence Transformer Model for Embeddings
embedding_model = SentenceTransformer('all-MiniLM-L6-v2') # Lightweight and effective

# Initialize FAISS Index
def initialize_faiss():
# Dimension should match the embedding model
dimension = 384 # 'all-MiniLM-L6-v2' has 384-dimensional embeddings
if os.path.exists(FAISS_INDEX_FILE):
try:
index = faiss.read_index(FAISS_INDEX_FILE)
logger.info("FAISS index loaded.")
except Exception as e:
logger.error(f"Failed to load FAISS index: {e}")
index = faiss.IndexFlatL2(dimension)
logger.info("Initialized new FAISS index.")
else:
index = faiss.IndexFlatL2(dimension)
logger.info("FAISS index initialized.")
return index

faiss_index = initialize_faiss()

# Initialize Metadata Storage
def load_memory_metadata():
if os.path.exists(MEMORY_METADATA_FILE):
try:
with open(MEMORY_METADATA_FILE, "r") as meta_file:
metadata = json.load(meta_file)
logger.info("Memory metadata loaded.")
except Exception as e:
logger.error(f"Failed to load memory metadata: {e}")
metadata = []
else:
metadata = []
logger.info("Memory metadata initialized.")
return metadata

def save_memory_metadata():
try:
with open(MEMORY_METADATA_FILE, "w") as meta_file:
json.dump(memory_metadata, meta_file, indent=4)
logger.info("Memory metadata saved.")
except Exception as e:
logger.error(f"Failed to save memory metadata: {e}")

memory_metadata = load_memory_metadata()

# Load conversation history if it exists
def load_conversation_history():
global messages
if os.path.exists(CONVERSATION_LOG_FILE):
try:
with open(CONVERSATION_LOG_FILE, "r") as log_file:
messages = json.load(log_file)
logger.info("Conversation history loaded.")
except Exception as e:
logger.error(f"Failed to load conversation history: {e}")
messages = []
else:
# Add system prompt with chain-of-thought instructions for internal processing
system_prompt = {
"role": "system",
"content": (
"You are an advanced AI assistant with the ability to reason through problems step-by-step using chain-of-thought. "
"Maintain context over extended conversations, recall past interactions, and utilize stored memories to provide coherent and contextually relevant responses."
" Ensure thorough reasoning for complex tasks and provide detailed explanations when required."
),
}
messages.append(system_prompt)
logger.info("System prompt initialized.")

# Save conversation history
def save_conversation_history():
try:
with open(CONVERSATION_LOG_FILE, "w") as log_file:
json.dump(messages, log_file, indent=4)
logger.info("Conversation history saved.")
except Exception as e:
logger.error(f"Failed to save conversation history: {e}")

# Load memory if it exists
def load_memory():
global memory, memory_metadata
if os.path.exists(MEMORY_FILE):
try:
with open(MEMORY_FILE, "r") as memory_file:
memory = json.load(memory_file)
logger.info("Memory loaded.")
except Exception as e:
logger.error(f"Failed to load memory: {e}")
memory = {}
else:
memory = {}
logger.info("Memory initialized.")

# memory_metadata already loaded above

# Save memory
def save_memory():
try:
with open(MEMORY_FILE, "w") as memory_file:
json.dump(memory, memory_file, indent=4)
logger.info("Memory saved.")
save_memory_metadata()
except Exception as e:
logger.error(f"Failed to save memory: {e}")

# Function to add memory to FAISS
def add_memory_to_faiss(text: str, metadata: Dict):
try:
embedding = embedding_model.encode(text)
faiss_index.add(np.array([embedding]).astype('float32'))
faiss.write_index(faiss_index, FAISS_INDEX_FILE)
memory_metadata.append(metadata)
save_memory_metadata()
logger.info("Memory added to FAISS index.")
except Exception as e:
logger.error(f"Failed to add memory to FAISS: {e}")

# Function to retrieve relevant memories based on user input
def retrieve_relevant_memories(user_input: str, top_k: int = 5) -> List[Dict]:
try:
embedding = embedding_model.encode(user_input)
embedding = np.array([embedding]).astype('float32')
distances, indices = faiss_index.search(embedding, top_k)
relevant = []
for idx in indices[0]:
if idx < len(memory_metadata):
relevant.append(memory_metadata[idx])
else:
logger.warning(f"FAISS returned out-of-bounds index: {idx}")
return relevant
except Exception as e:
logger.error(f"Failed to retrieve relevant memories: {e}")
return []

# Function to extract code blocks from text
def extract_code_block(text: str) -> str:
"""
Extracts the code block from a text enclosed in triple backticks.
"""
if "```" in text:
parts = text.split("```")
if len(parts) >= 3:
# Case with language specified, e.g., ```python
return parts[2].strip()
elif len(parts) == 2:
# Case without language specification
return parts[1].strip()
return ""

# Function to retrieve the last memory entry of a specific type
def retrieve_last_memory_of_type(memory_type: str) -> str:
"""
Retrieves the last memory entry of a specific type ('code', 'explanation', etc.).
"""
# Iterate in reverse to find the latest relevant memory
for entry in reversed(memory_metadata):
text_lower = entry["text"].lower()
if memory_type == "code" and "code" in text_lower:
# Extract code block
code_block = extract_code_block(entry["text"])
if code_block:
return code_block
elif memory_type == "explanation" and "explanation" in text_lower:
return entry["text"]
return ""

# Update memory based on the latest assistant response
def update_memory(user_input: str, assistant_response: str):
# Combine user input and assistant response for context
combined_text = f"User: {user_input}\nAssistant: {assistant_response}"
# Add to memory dictionary
timestamp = uuid.uuid4().hex # Unique identifier
if timestamp not in memory:
memory[timestamp] = []
memory[timestamp].append({
"user": user_input,
"assistant": assistant_response
})
# Add to FAISS index
add_memory_to_faiss(combined_text, {
"id": timestamp,
"text": combined_text
})
logger.info("Memory updated with semantic embeddings.")

print("Welcome to the Enhanced Coding Assistant with Advanced Reasoning and Long-Term Memory! Type 'exit' to end the conversation.")

def copy_code_to_clipboard(code: str):
"""Copies the provided code to the clipboard."""
try:
pyperclip.copy(code)
logger.info("Code copied to clipboard.")
return "Code copied to clipboard!"
except Exception as e:
logger.error(f"Error while copying code to clipboard: {e}", exc_info=True)
return "An error occurred while trying to copy the code."

# Define AI Worker classes
class AIWorker:
"""Base class for AI Workers."""
def __init__(self, name: str):
self.name = name

async def process(self, task: Dict, context: Dict) -> Dict:
raise NotImplementedError("Each AIWorker must implement the process method.")

class CodeGeneratorWorker(AIWorker):
"""AI Worker specialized in generating code."""
def __init__(self):
super().__init__("CodeGenerator")

async def process(self, task: Dict, context: Dict) -> Dict:
try:
user_input = task.get("input")
prompt = (
"You will be provided with a user request. "
"Please generate the corresponding Python code within triple backticks. "
"Use chain-of-thought internally to reason through the problem step-by-step, "
"but only output the final code without any explanations."
)
full_prompt = f"{prompt}\n\nUser Request: {user_input}"
logger.info(f"{self.name} processing task: {user_input}")
response = client.chat.completions.create(
messages=[
{"role": "system", "content": full_prompt}
],
model="blackbox",
# Adjust parameters to allow longer responses
max_tokens=3500, # Increased token limit
temperature=0.7, # Adjust temperature for creativity
)
if response.choices and len(response.choices) > 0:
code = response.choices[0].message.content
# Ensure only code block is returned
code = extract_code_block(code)
# Handle long code by chunking logically
code_chunks = chunk_text_logically(code, max_length=2000)
return {"code_chunks": code_chunks}
else:
code = "I'm sorry, but I couldn't generate the code at this time."
logger.warning("CodeGeneratorWorker: GPT response choices are empty.")
return {"code_chunks": [code]}
except Exception as e:
logger.error(f"{self.name} encountered an error: {e}", exc_info=True)
return {"code_chunks": ["An error occurred while generating the code."]}

class ExplanationWorker(AIWorker):
"""AI Worker specialized in explaining code or concepts."""
def __init__(self):
super().__init__("Explanation")

async def process(self, task: Dict, context: Dict) -> Dict:
try:
code = task.get("code")
if not code:
return {"explanation_chunks": ["No code provided for explanation."]}
prompt = (
"You are provided with a Python code snippet. "
"Use your chain-of-thought reasoning to understand the code step-by-step, "
"and provide a detailed explanation of what the code does."
" Ensure the explanation is clear and concise without including the internal reasoning process."
)
full_prompt = f"{prompt}\n\nCode:\n```python\n{code}\n```"
logger.info(f"{self.name} processing task.")
response = client.chat.completions.create(
messages=[
{"role": "system", "content": full_prompt}
],
model="gpt-4o",
max_tokens=3000, # Increased token limit
temperature=0.7,
)
if response.choices and len(response.choices) > 0:
explanation = response.choices[0].message.content
# Handle long explanations by chunking logically
explanation_chunks = chunk_text_logically(explanation, max_length=2000)
return {"explanation_chunks": explanation_chunks}
else:
explanation = "I'm sorry, but I couldn't provide an explanation at this time."
logger.warning("ExplanationWorker: GPT response choices are empty.")
return {"explanation_chunks": [explanation]}
except Exception as e:
logger.error(f"{self.name} encountered an error: {e}", exc_info=True)
return {"explanation_chunks": ["An error occurred while explaining the code."]}

class DataRetrievalWorker(AIWorker):
"""AI Worker specialized in retrieving data or information."""
def __init__(self):
super().__init__("DataRetrieval")

async def process(self, task: Dict, context: Dict) -> Dict:
try:
query = task.get("query")
prompt = (
"You are provided with a data retrieval request. "
"Analyze the request using chain-of-thought reasoning to understand the required information. "
"Provide the final detailed information without any internal explanations."
)
full_prompt = f"{prompt}\n\nData Request: {query}"
logger.info(f"{self.name} processing task: {query}")
response = client.chat.completions.create(
messages=[
{"role": "system", "content": full_prompt}
],
model="gpt-4o",
max_tokens=3500, # Increased token limit
temperature=0.7,
)
if response.choices and len(response.choices) > 0:
data = response.choices[0].message.content
# Handle long data by chunking logically
data_chunks = chunk_text_logically(data, max_length=2000)
return {"data_chunks": data_chunks}
else:
data = "I'm sorry, but I couldn't retrieve the data at this time."
logger.warning("DataRetrievalWorker: GPT response choices are empty.")
return {"data_chunks": [data]}
except Exception as e:
logger.error(f"{self.name} encountered an error: {e}", exc_info=True)
return {"data_chunks": ["An error occurred while retrieving the data."]}

class PruningWorker(AIWorker):
"""AI Worker specialized in code optimization."""
def __init__(self):
super().__init__("Pruning")

async def process(self, task: Dict, context: Dict) -> Dict:
try:
code = task.get("code")
if not code:
return {"optimized_code_chunks": ["No code provided for optimization."]}
prompt = (
"You are provided with a Python code snippet that implements the Quicksort algorithm. "
"Use your chain-of-thought reasoning to analyze and optimize the code for better performance and readability. "
"Provide only the optimized code within triple backticks without any additional explanations."
)
full_prompt = f"{prompt}\n\nOriginal Code:\n```python\n{code}\n```"
logger.info(f"{self.name} processing task.")
response = client.chat.completions.create(
messages=[
{"role": "system", "content": full_prompt}
],
model="blackbox",
max_tokens=3500, # Increased token limit
temperature=0.7,
)
if response.choices and len(response.choices) > 0:
optimized_code = response.choices[0].message.content
optimized_code = extract_code_block(optimized_code)
# Handle long optimized code by chunking logically
optimized_code_chunks = chunk_text_logically(optimized_code, max_length=2000)
return {"optimized_code_chunks": optimized_code_chunks}
else:
optimized_code = "I'm sorry, but I couldn't optimize the code at this time."
logger.warning("PruningWorker: GPT response choices are empty.")
return {"optimized_code_chunks": [optimized_code]}
except Exception as e:
logger.error(f"{self.name} encountered an error: {e}", exc_info=True)
return {"optimized_code_chunks": ["An error occurred while optimizing the code."]}

# New Worker: SummarizationWorker
class SummarizationWorker(AIWorker):
"""AI Worker specialized in summarizing text."""
def __init__(self):
super().__init__("Summarization")

async def process(self, task: Dict, context: Dict) -> Dict:
try:
text = task.get("text")
if not text:
return {"summary_chunks": ["No text provided for summarization."]}
prompt = (
"Provide a concise summary of the following text. "
"Ensure the summary captures all key points without adding any new information. "
"Use clear and precise language."
)
full_prompt = f"{prompt}\n\nText:\n{text}"
logger.info(f"{self.name} processing task.")
response = client.chat.completions.create(
messages=[
{"role": "system", "content": full_prompt}
],
model="blackbox",
max_tokens=4500,
temperature=0.5,
)
if response.choices and len(response.choices) > 0:
summary = response.choices[0].message.content
# Handle long summaries by chunking logically
summary_chunks = chunk_text_logically(summary, max_length=2000)
return {"summary_chunks": summary_chunks}
else:
summary = "I'm sorry, but I couldn't generate a summary at this time."
logger.warning("SummarizationWorker: GPT response choices are empty.")
return {"summary_chunks": [summary]}
except Exception as e:
logger.error(f"{self.name} encountered an error: {e}", exc_info=True)
return {"summary_chunks": ["An error occurred while summarizing the text."]}

# New Worker: TranslationWorker
class TranslationWorker(AIWorker):
"""AI Worker specialized in translating text."""
def __init__(self):
super().__init__("Translation")

async def process(self, task: Dict, context: Dict) -> Dict:
try:
text = task.get("text")
target_language = task.get("language", "Spanish") # Default to Spanish
if not text:
return {"translation_chunks": ["No text provided for translation."]}
prompt = (
f"Translate the following text into {target_language}. "
"Ensure the translation is accurate and maintains the original context. "
"Use proper grammar and syntax."
)
full_prompt = f"{prompt}\n\nText:\n{text}"
logger.info(f"{self.name} processing task.")
response = client.chat.completions.create(
messages=[
{"role": "system", "content": full_prompt}
],
model="gpt-4o",
max_tokens=3500,
temperature=0.3,
)
if response.choices and len(response.choices) > 0:
translation = response.choices[0].message.content
# Handle long translations by chunking logically
translation_chunks = chunk_text_logically(translation, max_length=2000)
return {"translation_chunks": translation_chunks}
else:
translation = "I'm sorry, but I couldn't perform the translation at this time."
logger.warning("TranslationWorker: GPT response choices are empty.")
return {"translation_chunks": [translation]}
except Exception as e:
logger.error(f"{self.name} encountered an error: {e}", exc_info=True)
return {"translation_chunks": ["An error occurred while translating the text."]}

# Additional Worker: DebuggingWorker
class DebuggingWorker(AIWorker):
"""AI Worker specialized in debugging code."""
def __init__(self):
super().__init__("Debugging")

async def process(self, task: Dict, context: Dict) -> Dict:
try:
code = task.get("code")
error_message = task.get("error")
if not code or not error_message:
return {"debugging_chunks": ["Insufficient information provided for debugging."]}
prompt = (
"You are provided with a Python code snippet and an associated error message. "
"Use your chain-of-thought reasoning to identify the cause of the error and provide a corrected version of the code. "
"Include comments explaining the changes made."
)
full_prompt = f"{prompt}\n\nOriginal Code:\n```python\n{code}\n```\n\nError Message:\n{error_message}"
logger.info(f"{self.name} processing task.")
response = client.chat.completions.create(
messages=[
{"role": "system", "content": full_prompt}
],
model="blackbox",
max_tokens=3500,
temperature=0.7,
)
if response.choices and len(response.choices) > 0:
debugging_output = response.choices[0].message.content
# Handle long debugging outputs by chunking logically
debugging_chunks = chunk_text_logically(debugging_output, max_length=2000)
return {"debugging_chunks": debugging_chunks}
else:
debugging_output = "I'm sorry, but I couldn't debug the code at this time."
logger.warning("DebuggingWorker: GPT response choices are empty.")
return {"debugging_chunks": [debugging_output]}
except Exception as e:
logger.error(f"{self.name} encountered an error: {e}", exc_info=True)
return {"debugging_chunks": ["An error occurred while debugging the code."]}

# Additional Worker: OptimizationWorker
class OptimizationWorker(AIWorker):
"""AI Worker specialized in optimizing algorithms."""
def __init__(self):
super().__init__("Optimization")

async def process(self, task: Dict, context: Dict) -> Dict:
try:
algorithm_description = task.get("algorithm")
if not algorithm_description:
return {"optimization_chunks": ["No algorithm description provided for optimization."]}
prompt = (
"You are provided with a description of an algorithm. "
"Use your chain-of-thought reasoning to analyze the algorithm and provide an optimized version. "
"Focus on improving time and space complexity without compromising correctness."
)
full_prompt = f"{prompt}\n\nAlgorithm Description:\n{algorithm_description}"
logger.info(f"{self.name} processing task.")
response = client.chat.completions.create(
messages=[
{"role": "system", "content": full_prompt}
],
model="gpt-4o",
max_tokens=3500,
temperature=0.7,
)
if response.choices and len(response.choices) > 0:
optimization_output = response.choices[0].message.content
# Handle long optimization outputs by chunking logically
optimization_chunks = chunk_text_logically(optimization_output, max_length=2000)
return {"optimization_chunks": optimization_chunks}
else:
optimization_output = "I'm sorry, but I couldn't optimize the algorithm at this time."
logger.warning("OptimizationWorker: GPT response choices are empty.")
return {"optimization_chunks": [optimization_output]}
except Exception as e:
logger.error(f"{self.name} encountered an error: {e}", exc_info=True)
return {"optimization_chunks": ["An error occurred while optimizing the algorithm."]}

# New Worker: PlannerWorker
class PlannerWorker(AIWorker):
"""AI Worker specialized in planning and task decomposition."""
def __init__(self):
super().__init__("Planner")

async def process(self, task: Dict, context: Dict) -> Dict:
try:
user_input = task.get("input")
prompt = (
"You are an AI assistant with advanced reasoning capabilities. "
"Analyze the following user request and decompose it into a series of actionable sub-tasks required to fulfill the request. "
"Provide a clear and ordered list of these sub-tasks."
)
full_prompt = f"{prompt}\n\nUser Request: {user_input}"
logger.info(f"{self.name} processing planning for: {user_input}")
response = client.chat.completions.create(
messages=[
{"role": "system", "content": full_prompt}
],
model="gpt-4o",
max_tokens=1500,
temperature=0.5,
)
if response.choices and len(response.choices) > 0:
plan = response.choices[0].message.content
# Assume the plan is a numbered or bulleted list; parse it into tasks
sub_tasks = parse_plan(plan)
logger.info(f"Planner generated {len(sub_tasks)} sub-tasks.")
return {"plan": sub_tasks}
else:
logger.warning("PlannerWorker: GPT response choices are empty.")
return {"plan": []}
except Exception as e:
logger.error(f"{self.name} encountered an error: {e}", exc_info=True)
return {"plan": []}

# Helper function to parse the planner's output into sub-tasks
def parse_plan(plan_text: str) -> List[Dict]:
"""
Parses a numbered or bulleted list from the planner into a list of task dictionaries.
"""
sub_tasks = []
lines = plan_text.strip().split('\n')
for line in lines:
line = line.strip()
if line and (line.startswith('1.') or line.startswith('2.') or line.startswith('-') or (line[0].isdigit() and (line[1] in ['.', ')']))):
# Remove numbering or bullets
if line.startswith(('1.', '2.', '3.', '4.', '5.', '6.', '7.', '8.', '9.')) or (line[0].isdigit() and line[1] in ['.', ')']):
task_description = ''.join([c for c in line if not c.isdigit() and c not in ['.', ')', '-']]).strip()
elif line.startswith('-'):
task_description = line[1:].strip()
else:
task_description = line.strip()
if task_description:
sub_tasks.append({"description": task_description})
return sub_tasks

# Helper function to chunk text logically
def chunk_text_logically(text: str, max_length: int = 2000) -> List[str]:
"""
Splits the text into chunks of specified maximum length, ensuring that splits occur at logical boundaries
such as paragraph breaks or code blocks to maintain coherence.
"""
chunks = []
current_chunk = ""
lines = text.split('\n')
for line in lines:
# Check if adding the next line exceeds the max length
if len(current_chunk) + len(line) + 1 > max_length:
if current_chunk:
chunks.append(current_chunk.strip())
current_chunk = ""
current_chunk += line + '\n'
if current_chunk:
chunks.append(current_chunk.strip())
return chunks

# Initialize AI Workers with the new additions
ai_workers = {
"code_generator": CodeGeneratorWorker(),
"explanation": ExplanationWorker(),
"data_retrieval": DataRetrievalWorker(),
"pruning": PruningWorker(),
"summarization": SummarizationWorker(),
"translation": TranslationWorker(),
"debugging": DebuggingWorker(),
"optimization": OptimizationWorker(),
"planner": PlannerWorker(), # New Planner Worker
}

async def dispatch_tasks_async(tasks: List[Dict], context: Dict) -> Dict:
"""
Dispatches tasks to the appropriate AI Workers asynchronously and aggregates the results.
"""
results = {}
tasks_coroutines = []
for task in tasks:
worker_type = task.get("worker")
if worker_type in ai_workers:
worker = ai_workers[worker_type]
tasks_coroutines.append(worker.process(task, context))
else:
logger.warning(f"No AI Worker found for type: {worker_type}")
try:
tasks_results = await asyncio.gather(*tasks_coroutines, return_exceptions=True)
for results_item in tasks_results:
if isinstance(results_item, Exception):
logger.error(f"Error in AI Worker processing: {results_item}", exc_info=True)
elif isinstance(results_item, dict):
results.update(results_item)
except Exception as e:
logger.error(f"Error during task dispatching: {e}", exc_info=True)
return results

# Enhanced Incremental Reasoning Function with Multi-Step Problem Solving and Meta Reasoning
async def incremental_reasoning_async(user_input: str, context: Dict) -> Dict:
"""
Performs incremental reasoning by breaking down the user input into tasks using the PlannerWorker,
dispatching them to AI Workers, and aggregating the results.
"""
reasoning_id = str(uuid.uuid4())
reasoning_state[reasoning_id] = {
"steps": [],
"results": {}
}

# Decompose the user input into sub-tasks using the PlannerWorker
planner_task = {"worker": "planner", "input": user_input}
planner_result = await ai_workers["planner"].process(planner_task, reasoning_state)

sub_tasks = planner_result.get("plan", [])

if not sub_tasks:
logger.info("PlannerWorker did not return any sub-tasks. Proceeding without decomposition.")
# Fallback: categorize based on keywords as before
sub_tasks = categorize_tasks(user_input)

# Convert sub_tasks descriptions to specific worker tasks
actionable_tasks = []
for sub_task in sub_tasks:
description = sub_task.get("description", "").lower()
if "generate code" in description or "create code" in description:
actionable_tasks.append({"worker": "code_generator", "input": user_input})
elif "explain" in description:
last_code = retrieve_last_memory_of_type("code")
if last_code:
actionable_tasks.append({"worker": "explanation", "code": last_code})
else:
logger.info("No recent code found for explanation.")
elif "retrieve data" in description or "get information" in description:
actionable_tasks.append({"worker": "data_retrieval", "query": user_input})
elif "summarize" in description or "summary" in description:
actionable_tasks.append({"worker": "summarization", "text": user_input})
elif "translate" in description:
actionable_tasks.append({"worker": "translation", "text": user_input, "language": "French"}) # Example target language
elif "debug" in description or "fix error" in description:
# Assume the user provides code and an error message separated by a delimiter, e.g., '||'
if "||" in user_input:
code_part, error_part = user_input.split("||", 1)
actionable_tasks.append({"worker": "debugging", "code": code_part.strip(), "error": error_part.strip()})
else:
logger.info("Insufficient information for debugging.")
elif "optimize" in description and "algorithm" in description:
actionable_tasks.append({"worker": "optimization", "algorithm": user_input})
elif "optimize code" in description:
actionable_tasks.append({"worker": "pruning", "code": user_input})
# Add more conditions as needed based on planning output

if not actionable_tasks:
# If planner did not identify specific tasks, proceed with keyword-based categorization
actionable_tasks = categorize_tasks(user_input)

# Dispatch tasks to AI Workers
logger.info(f"Dispatching {len(actionable_tasks)} actionable tasks for reasoning ID: {reasoning_id}")
results = await dispatch_tasks_async(actionable_tasks, reasoning_state)

reasoning_state[reasoning_id]["steps"].extend(actionable_tasks)
reasoning_state[reasoning_id]["results"].update(results)

return results

def categorize_tasks(user_input: str) -> List[Dict]:
"""
Categorizes user input into tasks based on keyword analysis.
"""
tasks = []
lower_input = user_input.lower()

# Multi-step optimization for Quicksort
if "optimize" in lower_input and "quicksort" in lower_input:
tasks.append({"worker": "pruning", "code": user_input})

# Code generation
if "code" in lower_input or "generate" in lower_input:
tasks.append({"worker": "code_generator", "input": user_input})

# Code explanation
if "explain" in lower_input:
# Retrieve the last generated code if exists
last_code = retrieve_last_memory_of_type("code")
if last_code:
tasks.append({"worker": "explanation", "code": last_code})
else:
logger.info("No recent code found for explanation.")

# Data retrieval
if "data" in lower_input or "information" in lower_input:
tasks.append({"worker": "data_retrieval", "query": user_input})

# Summarization
if "summarize" in lower_input or "summary" in lower_input:
tasks.append({"worker": "summarization", "text": user_input})

# Translation
if "translate" in lower_input:
tasks.append({"worker": "translation", "text": user_input, "language": "French"}) # Example target language

# Debugging
if "debug" in lower_input or "error" in lower_input:
# Assume the user provides code and an error message separated by a delimiter, e.g., '||'
if "||" in user_input:
code_part, error_part = user_input.split("||", 1)
tasks.append({"worker": "debugging", "code": code_part.strip(), "error": error_part.strip()})
else:
logger.info("Insufficient information for debugging.")

# Optimization
if "optimize algorithm" in lower_input or "optimize" in lower_input:
tasks.append({"worker": "optimization", "algorithm": user_input})

return tasks

# Gradio Interface Function with Advanced Features
async def chat_interface(user_input: str, history: List[Dict[str, str]]):
if user_input.lower() == "exit":
response = "Exiting chat..."
# Append the exit message to history
history.append({"role": "assistant", "content": response})
save_conversation_history() # Save the conversation before exiting
save_memory() # Save memory before exiting
return history, history

# Append user message in 'role'/'content' format
history.append({"role": "user", "content": user_input})
messages.append({"role": "user", "content": user_input}) # Sync with messages list

# Retrieve relevant memories based on user input
relevant_memories = retrieve_relevant_memories(user_input)
if relevant_memories:
memory_prompt = "\n".join([f"{entry['text']}" for entry in relevant_memories])
system_message = {
"role": "system",
"content": f"Relevant past interactions:\n{memory_prompt}"
}
messages.append(system_message)
history.append(system_message)
logger.info(f"Retrieved {len(relevant_memories)} relevant memories.")

# Perform incremental reasoning asynchronously
reasoning_results = await incremental_reasoning_async(user_input, reasoning_state)

# Generate assistant response based on reasoning results
assistant_response_chunks = []
if "code_chunks" in reasoning_results:
code_chunks = reasoning_results['code_chunks']
for chunk in code_chunks:
assistant_response_chunks.append(f"Here is the code you requested:\n```python\n{chunk}\n```")

if "optimized_code_chunks" in reasoning_results:
optimized_code_chunks = reasoning_results['optimized_code_chunks']
for chunk in optimized_code_chunks:
assistant_response_chunks.append(f"Optimized Code:\n```python\n{chunk}\n```")

if "explanation_chunks" in reasoning_results:
explanation_chunks = reasoning_results['explanation_chunks']
for chunk in explanation_chunks:
assistant_response_chunks.append(f"**Explanation:**\n{chunk}\n")

if "data_chunks" in reasoning_results:
data_chunks = reasoning_results['data_chunks']
for chunk in data_chunks:
assistant_response_chunks.append(f"**Data:**\n{chunk}\n")

if "summary_chunks" in reasoning_results:
summary_chunks = reasoning_results['summary_chunks']
for chunk in summary_chunks:
assistant_response_chunks.append(f"**Summary:**\n{chunk}\n")

if "translation_chunks" in reasoning_results:
translation_chunks = reasoning_results['translation_chunks']
for chunk in translation_chunks:
assistant_response_chunks.append(f"**Translation:**\n{chunk}\n")

if "debugging_chunks" in reasoning_results:
debugging_chunks = reasoning_results['debugging_chunks']
for chunk in debugging_chunks:
assistant_response_chunks.append(f"**Debugging Information:**\n{chunk}\n")

if "optimization_chunks" in reasoning_results:
optimization_chunks = reasoning_results['optimization_chunks']
for chunk in optimization_chunks:
assistant_response_chunks.append(f"**Algorithm Optimization:**\n{chunk}\n")

# If specific chunks were generated, append them as separate messages
if assistant_response_chunks:
for chunk in assistant_response_chunks:
history.append({"role": "assistant", "content": chunk})
messages.append({"role": "assistant", "content": chunk}) # Sync with messages list
else:
# If no specific tasks were dispatched, fall back to general response
try:
response = client.chat.completions.create(
messages=messages,
model="gpt-4o",
max_tokens=4000, # Increased token limit for longer responses
temperature=0.7,
)
if response.choices and len(response.choices) > 0:
assistant_response = response.choices[0].message.content
# Handle long responses by chunking logically
response_chunks = chunk_text_logically(assistant_response, max_length=2000)
for chunk in response_chunks:
history.append({"role": "assistant", "content": chunk})
messages.append({"role": "assistant", "content": chunk}) # Sync with messages list
else:
assistant_response = "I'm sorry, but I couldn't generate a response at this time."
logger.warning("Fallback GPT response choices are empty.")
history.append({"role": "assistant", "content": assistant_response})
messages.append({"role": "assistant", "content": assistant_response}) # Sync with messages list
except Exception as e:
assistant_response = "An error occurred while generating a response."
logger.error(f"Fallback GPT error: {e}", exc_info=True)
history.append({"role": "assistant", "content": assistant_response})
messages.append({"role": "assistant", "content": assistant_response}) # Sync with messages list

# Log the assistant's response
logger.info(f"Bot: {' '.join([chunk for chunk in assistant_response_chunks])}")

# Update memory with the latest interaction
# For multiple chunks, join them to form the full response
full_assistant_response = "\n\n".join(assistant_response_chunks)
update_memory(user_input, full_assistant_response)

return history, history

# Launch Gradio Interface with Advanced Features
def launch_gradio():
with gr.Blocks(theme="default") as interface:
gr.Markdown("""
# 🧑‍💻 **Enhanced Coding Assistant**

Interact with an AI assistant that can generate, optimize, explain code, summarize texts, translate languages, debug code, and retrieve information. Supports advanced Markdown formatting for rich responses.
""")
with gr.Tab("Chat"):
with gr.Row():
with gr.Column(scale=8):
user_input = gr.Textbox(label="You:", placeholder="Type your message here...", lines=2)
with gr.Column(scale=2):
send_button = gr.Button("Send", variant="primary")
with gr.Row():
with gr.Column():
chatbot = gr.Chatbot(label="Chat History", type="messages") # Updated type parameter
# Hidden state to maintain conversation
state = gr.State([])

with gr.Tab("Conversation Log"):
gr.Markdown("### 📝 Conversation History")
conversation_display = gr.JSON(label="History", value=messages)

with gr.Tab("Memory"):
gr.Markdown("### 🧠 AI Assistant Memory")
memory_display = gr.JSON(label="Memory", value=memory)

# Define the action when send button is clicked or Enter is pressed
send_button.click(fn=chat_interface, inputs=[user_input, state], outputs=[chatbot, state])
user_input.submit(fn=chat_interface, inputs=[user_input, state], outputs=[chatbot, state])

# Updated queue parameters
interface.queue(max_size=10) # Adjusted from concurrency_count=10 based on latest Gradio

interface.launch()

# Run the Gradio interface when the script is executed
if __name__ == "__main__":
try:
load_conversation_history()
load_memory()
launch_gradio()
except KeyboardInterrupt:
print("\nChatbot terminated by user.")
logger.info("Chatbot terminated by user via KeyboardInterrupt.")
except Exception as e:
print(f"An unexpected error occurred: {e}")
logger.error(f"Unexpected error: {e}", exc_info=True)