src/reasoning/agent_metacognition.py

# © 2025 Elena Marziali — Code released under Apache 2.0 license.
# See LICENSE in the repository for details.
# Removal of this copyright is prohibited.

# === metacognitive_cycle ===
# Executes an iterative cycle of evaluation and improvement of the generated response.
# Combines qualitative feedback and semantic coherence score to decide whether to reformulate.
# Useful for simulating reflective and adaptive behavior.

def generate_objective_from_input(user_input):
    """
    Generates a high-level operational objective based on the user's input.
    Useful for AGI-style planning and decision-making.
    """
    prompt = f"""
    You are an autonomous scientific agent. Based on the following input:
    "{user_input}"

    Define a clear and actionable objective that guides the agent's next steps.
    """
    try:
        response = llm.invoke(prompt.strip())
        return getattr(response, "content", str(response)).strip()
    except Exception as e:
        logging.error(f"Error generating objective: {e}")
        return "Objective generation failed."


def metacognitive_cycle(question, level, max_iter=2):
    response = llm.invoke(question)
    response_text = extract_text_from_ai(response)

    for i in range(max_iter):
        feedback = auto_feedback_response(question, response_text, level)
        score = evaluate_coherence(question, response_text)

        print(f"\nIteration {i+1} – Coherence: {score:.3f}")
        print("Feedback:", extract_text_from_ai(feedback))

        if score < 0.7:
            response_text = extract_text_from_ai(improve_response(question, response_text, level))
        else:
            break

    return response_text

# Evaluate response with self-assessment and interactive improvement
# Evaluates the response and reformulates it if poorly constructed
def evaluate_responses_with_ai(question, generate_response_fn, n_variants=3, reformulation_threshold=0.6):
    temperature_values = [0.7, 0.4, 0.9][:n_variants]
    responses = [generate_response_fn(question, temperature=t) for t in temperature_values]

    scores = [evaluate_coherence(question, r) for r in responses]
    idx = scores.index(max(scores))
    confidence = scores[idx]
    best_response = responses[idx]

    if confidence < reformulation_threshold:
        new_question = reformulate_question(question)
        return evaluate_responses_with_ai(new_question, generate_response_fn)

    return {
        "response": best_response,
        "confidence": round(confidence, 3),
        "note": generate_note(confidence)
    }

def evaluate_responses_with_ai_simple(question, response, level="basic"):
    """
    Evaluates the quality of the generated response relative to the question.
    Returns a dictionary with:
    - semantic coherence score
    - reason for weakness
    - suggested reformulation
    - reflection on reasoning
    - flag for auto-improvement
    """

    evaluation_prompt = f"""
    User question: "{question}"
    Generated response: "{response}"
    Required level: {level}

    Evaluate the response in 5 points:
    1. Semantic coherence (0–1)
    2. Conceptual completeness
    3. Argumentative structure
    4. Adequacy to the required level
    5. Ability to stimulate new questions

    If the response is weak:
    - Explain the reason
    - Suggest a reformulation
    - Reflect on how the system reasoned

    Return everything in structured format.
    """

    try:
        ai_evaluation = llm.invoke(evaluation_prompt)
        raw_output = getattr(ai_evaluation, "content", str(ai_evaluation))
    except Exception as e:
        print("Evaluation error:", e)
        return {
            "semantic_score": 0.0,
            "weakness_reason": "System error",
            "new_formulation": None,
            "self_reflection": None,
            "requires_improvement": True
        }

    # Simplified parsing functions (can be enhanced with regex or LLM)
    def extract_score(text):
        match = re.search(r"Semantic coherence\s*[:\-]?\s*(0\.\d+)", text)
        return float(match.group(1)) if match else 0.0

    def extract_reason(text):
        match = re.search(r"Reason\s*[:\-]?\s*(.+)", text)
        return match.group(1).strip() if match else "Reason not found."

    def extract_reformulation(text):
        match = re.search(r"Reformulation\s*[:\-]?\s*(.+)", text)
        return match.group(1).strip() if match else None

    def extract_reflection(text):
        match = re.search(r"Reflection\s*[:\-]?\s*(.+)", text)
        return match.group(1).strip() if match else None

    # Actual parsing
    score = extract_score(raw_output)
    reason = extract_reason(raw_output)
    reformulation = extract_reformulation(raw_output)
    reflection = extract_reflection(raw_output)

    return {
        "response": response,
        "semantic_score": score,
        "weakness_reason": reason,
        "new_formulation": reformulation,
        "self_reflection": reflection,
        "requires_improvement": score < 0.7
    }

def generate_metacognitive_content(question, response, evaluation):
    return f"""
    [Question] {question}
    [Response] {response}
    [Coherence Score] {evaluation['semantic_score']}
    [Weakness Reason] {evaluation['weakness_reason']}
    [Suggested Reformulation] {evaluation['new_formulation']}
    [Cognitive Reflection] {evaluation['self_reflection']}
    [Needs Improvement] {evaluation['requires_improvement']}
    """.strip()

def add_metacognitive_memory(question, response):
    # Cognitive evaluation of the response
    evaluation = evaluate_responses_with_ai(question, response)

    # Generate textual content with all metacognitive data
    textual_content = generate_metacognitive_content(question, response, evaluation)

    # Generate semantic embedding from the full content
    embedding = embedding_model.encode([textual_content])

    # Add to FAISS index
    index.add(np.array(embedding, dtype=np.float32))

    # Save updated index
    with open(INDEX_FILE, "wb") as f:
        pickle.dump(index, f)

    print("Metacognitive memory updated!")

def search_similar_reasoning(query, top_k=5):
    """
    Searches the FAISS metacognitive memory for reasoning most similar to the input query.
    Returns a list of the most relevant textual contents.
    """
    # Encode the query
    query_vector = embedding_model.encode([query])

    # Search for top-K nearest
    distances, indices = index.search(np.array(query_vector, dtype=np.float32), top_k)

    results = []
    for idx in indices[0]:
        try:
            with open("meta_diary.json", "r", encoding="utf-8") as f:
                archive = json.load(f)
                content = archive.get(str(idx))
                if content:
                    results.append(content)
        except Exception as e:
            print(f"Memory retrieval error: {e}")

    return results

def add_metacognition_to_response(response, evaluation):
    reflection = evaluation.get("self_reflection", "")
    note = evaluation.get("weakness_reason", "")
    return f"{response.strip()}\n\n*Metacognitive note:* {note}\n*Agent's reflection:* {reflection}"

def auto_feedback(question, response, level):
    return f"""Analyze the response in relation to the question: "{question}".
Evaluate the content according to the level '{level}' and suggest improvements.
"""

# === Full flow example ===
async def scientific_creativity_flow(concept, subject, language="en", level="advanced"):
    creative_hypothesis = simulate_scientific_creativity(concept, subject, language=language, level=level)
    articles, _ = await search_multi_database(concept)  # Retrieve existing scientific sources
    novelty_evaluation = evaluate_hypothesis_novelty(creative_hypothesis, articles)

    return {
        "hypothesis": creative_hypothesis,
        "novelty": novelty_evaluation
    }