Create pipeline/agi_cognitive_pipeline.py

src/pipeline/agi_cognitive_pipeline.py

@@ -0,0 +1,303 @@
+# © 2025 Elena Marziali — Code released under Apache 2.0 license.
+# See LICENSE in the repository for details.
+# Removal of this copyright is prohibited.
+
+# This cell simulates AGI (Artificial General Intelligence) behavior,
+# with capabilities for planning, reasoning, generation, and self-assessment.
+
+# Interactive loop simulating a complete cognitive cycle
+async def agi_interactive_loop(user_input):
+    context = retrieve_multiturn_context(user_input, top_k=3)
+    planning = decompose_task(user_input)
+    results = []
+
+    for subtask in planning:
+        response = await generate_agi_response(subtask, context)
+        results.append(response)
+        update_memory(subtask, response)
+
+    return synthesize_final(results)
+
+
+cross_encoder = CrossEncoder("cross-encoder/nli-deberta-base")
+
+# Simulated historical archive for the question
+memory_archive = {}
+
+# Evaluate and version the generated response
+def evaluate_and_version_response(question, new_response, level="basic", acceptance_threshold=0.75):
+    """
+    Evaluates a new response using CrossEncoder,
+    compares it with previous versions,
+    and decides whether to keep or discard it.
+
+    Returns a dictionary with:
+    - evaluation outcome
+    - version details (if accepted)
+    - confidence and note (if discarded)
+    """
+
+    question_id = question.strip().lower()
+
+    # Step 1: Semantic evaluation of the new response
+    new_score = float(cross_encoder.predict([(question, new_response)])[0])
+
+    new_version = {
+        "id": str(uuid.uuid4()),
+        "response": new_response,
+        "coherence_score": round(new_score, 3),
+        "level": level,
+        "timestamp": datetime.datetime.utcnow().isoformat(),
+        "model_version": "LLM_v1",
+        "improvable": new_score < acceptance_threshold
+    }
+
+    # Step 2: Retrieve previous versions
+    previous_memory = memory_archive.get(question_id, [])
+
+    # If no previous versions exist, save the first one
+    if not previous_memory:
+        memory_archive[question_id] = [new_version]
+        return {
+            "outcome": "New question saved.",
+            "total_versions": 1,
+            "response_accepted": True,
+            "details": new_version
+        }
+
+    # Step 3: Compare with the best saved version
+    best_version = max(previous_memory, key=lambda v: v["coherence_score"])
+    best_score = best_version["coherence_score"]
+
+    if new_score > best_score:
+        memory_archive[question_id].append(new_version)
+        return {
+            "outcome": "New version saved (more coherent than previous).",
+            "total_versions": len(memory_archive[question_id]),
+            "response_accepted": True,
+            "details": new_version
+        }
+
+    # Version discarded: less coherent
+    return {
+        "outcome": "Version discarded: less coherent than existing ones.",
+        "response_accepted": False,
+        "confidence": round(new_score, 3),
+        "note": "The proposed version is less coherent than the previous one.",
+        "new_score": round(new_score, 3),
+        "best_score": round(best_score, 3)
+    }
+
+
+# === Main function: hypothesis generation and creative analysis ===
+def simulate_scientific_creativity(concept, subject, style="generative", level="advanced", language="it"):
+    prompt = f"""
+You are a cognitive scientific assistant with autonomous creative capabilities.
+
+Subject: {subject}
+Central concept: {concept}
+Requested creative style: {style}
+Level: {level}
+
+Objective: Generate an innovative scientific proposal.
+
+Respond with:
+1. An **original hypothesis** related to "{concept}".
+2. A **conceptual model** that can be visually described.
+3. A proposal for a **novel experiment** to test it.
+4. Possible **interdisciplinary applications**.
+5. A reflection on the degree of verifiability and impact.
+
+Translate everything into language: **{language}**
+"""
+    try:
+        response = llm.invoke(prompt.strip())
+        hypothesis_text = getattr(response, "content", str(response)).strip()
+        return hypothesis_text
+    except Exception as e:
+        logging.error(f"[simulate_creativity] Generation error: {e}")
+        return "Error during creative simulation."
+
+# === Classifications ===
+problem_type = analyze_question(example_problem)
+diagram_type_ml = extract_features(example_problem)
+print(f"Problem type: {problem_type}")
+print(f"Recommended representation: {diagram_type_ml}")
+
+logging.info(f"Identified problem type: {problem_type}")
+logging.info(f"Recommended representation type: {diagram_type_ml}")
+
+# === Assign concept from the 'topic' variable ===
+concept = topic.strip()
+
+# === Retrieve articles from arXiv with error handling ===
+try:
+    arxiv_articles = await search_arxiv_async(concept)
+    logging.info(f"arXiv: {len(arxiv_articles)} articles found.")
+except Exception as e:
+    logging.error(f"Error during arXiv search: {e}")
+    arxiv_articles = []
+
+# === Retrieve from other databases ===
+try:
+    pubmed_results = await search_pubmed_async(concept)
+    openalex_results = await search_openalex_async(concept)
+
+    logging.info("Search completed across all databases.")
+except Exception as e:
+    logging.error(f"Error in multi-database search: {e}")
+    pubmed_results = openalex_results = doaj_results = []
+
+# === Formatting for prompt or report ===
+async def retrieve_and_normalize_articles(concept):
+    """
+    Retrieves articles from multiple scientific sources and normalizes them.
+
+    Sources: arXiv, PubMed, OpenAlex, Zenodo
+
+    Returns:
+    - list of normalized articles
+    """
+    articles = []
+
+    try:
+        arxiv_articles = await search_arxiv_async(concept)
+    except Exception as e:
+        logging.error(f"[arxiv] Error: {e}")
+        arxiv_articles = []
+
+    try:
+        pubmed_articles = await search_pubmed_async(concept)
+    except Exception as e:
+        logging.error(f"[pubmed] Error: {e}")
+        pubmed_articles = []
+
+    try:
+        openalex_articles = await search_openalex_async(concept)
+    except Exception as e:
+        logging.error(f"[openalex] Error: {e}")
+        openalex_articles = []
+
+    try:
+        zenodo_articles = await search_zenodo_async(concept)
+    except Exception as e:
+        logging.error(f"[zenodo] Error: {e}")
+        zenodo_articles = []
+
+    sources = {
+        "arxiv": arxiv_articles,
+        "pubmed": pubmed_articles,
+        "openalex": openalex_articles,
+        "zenodo": zenodo_articles
+    }
+
+    for name, source in sources.items():
+        if isinstance(source, list) and all(isinstance(a, dict) for a in source):
+            articles += normalize_source(raw_articles=source, source_name=name)
+        else:
+            logging.warning(f"[{name}] Invalid data or unrecognized structure.")
+
+    logging.info(f"Total normalized articles: {len(articles)}")
+    return articles
+
+# Check if articles exist and format the text
+example_query = "quantum physics"  # Define the query
+articles = await search_multi_database(example_query)
+zenodo_articles = await search_zenodo_async(example_query)
+
+# === Prompt construction and response ===
+# Perform academic database search
+pubmed_results = await search_pubmed_async(concept)
+openalex_results = await search_openalex_async(concept)
+arxiv_results = await search_arxiv_async(concept)
+zenodo_results = await search_zenodo_async(concept)
+
+chart_choice_text = "Chart included" if chart_choice.lower() in ["yes"] else "Text only"
+
+paper_text = ""  # Or provide a predefined text
+
+# Modify language handling in the prompt to avoid errors
+prompt = prompt_template.format(
+    problem=example_problem,
+    topic=topic,
+    concept=concept,
+    level=level,
+    subject=subject,
+    arxiv_search=arxiv_results,
+    paper_text=paper_text,
+    pubmed_search=pubmed_results,
+    zenodo_search=zenodo_results,
+    openalex_search=openalex_results,
+    chart_choice=chart_choice_text,
+    target_language=target_language
+)
+
+try:
+    # Generate response
+    response = llm.invoke(prompt.strip())
+    response_content = getattr(response, "content", str(response))
+
+    if not response_content or "Error" in response_content:
+        raise ValueError("Invalid AI model response")
+    logging.info("Response successfully generated.")
+
+    # Reasoning explanation (metacognition)
+    reasoning_explanation = explain_reasoning(prompt, response_content)
+    print("Reasoning explanation:\n", getattr(reasoning_explanation, "content", reasoning_explanation))
+
+    # Operational decision (AGI Point 5)
+    objective = generate_objective_from_input(example_problem)
+    decision = llm.invoke(f"Objective: {objective}\nPrompt: {prompt.strip()}")
+    action = getattr(decision, "content", str(decision)).strip()
+    print(f"Agent's autonomous decision: {action}")
+
+except Exception as e:
+    logging.error(f"General error in AGI operational block: {e}")
+
+
+# This cell executes a generation + metacognition cycle
+
+final_response = metacognitive_cycle(example_problem, level)
+
+# Generates and evaluates the response for coherence and potential improvement
+def generate_and_evaluate(generation_prompt, question, level):
+    response = llm.invoke(generation_prompt)
+    evaluation_prompt = f"""
+    You received the following response: "{getattr(response, 'content', response)}".
+    - Is it coherent with the question: "{question}"?
+    - Is the tone appropriate for the '{level}' level?
+    - How would you improve the response?
+    """
+    feedback = llm.invoke(evaluation_prompt)
+    return response, feedback
+
+import time
+
+def execute_with_retry(function, max_attempts=3, base_delay=2):
+    for attempt in range(max_attempts):
+        try:
+            return function()
+        except InternalServerError as e:
+            logging.warning(f"Attempt {attempt+1} failed: {e}")
+            time.sleep(base_delay * (attempt + 1))
+        except Exception as e:
+            logging.error(f"Unhandled error: {e}")
+            break
+    return "Persistent error: unable to complete the operation."
+
+
+# === Visualization (optional) ===
+if chart_requested and diagram_type_ml in ["Chart", "Conceptual diagram", "State diagram"]:
+    logging.info("Generating interactive chart...")
+    try:
+        fig, caption = generate_interactive_chart(example_problem)
+        fig.show()
+        logging.info("Chart successfully generated!")
+    except Exception as e:
+        logging.error(f"Error during chart generation: {e}")
+else:
+    logging.info("Chart not requested or not necessary.")
+
+
+from IPython.display import FileLink
+FileLink(file_name)