Medical Agent with Metta

Overview

This guide demonstrates how to integrate SingularityNET's MeTTa (Meta Type Talk) knowledge graph system with Fetch.ai's uAgents framework to create intelligent, reasoning-capable autonomous agents. This integration enables agents to perform structured knowledge reasoning, pattern matching, and dynamic learning while maintaining compatibility with the ASI:One ecosystem.

What is MeTTa?

MeTTa (Meta Type Talk) is SingularityNET's multi-paradigm language for declarative and functional computations over knowledge (meta)graphs. It provides:

• Structured Knowledge Representation: Organize information in logical, queryable formats
• Symbolic Reasoning: Perform complex logical operations and pattern matching
• Knowledge Graph Operations: Build, query, and manipulate knowledge graphs
• Space-based Architecture: Knowledge stored as atoms in logical spaces

Installation & Setup

Prerequisites

Before you begin, ensure you have:

• Python 3.8+ installed
• pip package manager
• An ASI:One API key (get it from https://asi1.ai/)

---

Installation Options

You have two ways to install the required dependencies:

Option 1: Install All Dependencies at Once (Recommended)

Create a requirements.txt file with all necessary dependencies:

openai>=1.0.0
hyperon>=0.2.6
uagents>=0.22.5
uagents-core>=0.3.5
python-dotenv>=1.0.0

Install all dependencies with one command:

pip install -r requirements.txt

---

Option 2: Install Hyperon Separately

If you only want to install Hyperon (MeTTa) first:

pip install hyperon

Verify Hyperon Installation:

python -c "from hyperon import MeTTa; print('Hyperon installed successfully!')"

---

Windows Installation Guide

If you're using Windows OS and encounter issues installing Hyperon, please refer to this detailed video tutorial:

📺 Hyperon Installation on Windows - Video Guide

Architecture Overview

The integration follows this architectural pattern:

<Image will be here>

Core Integration Concepts

1. MeTTa Knowledge Graph Structure

MeTTa organizes knowledge as atoms in logical spaces:

from hyperon import MeTTa, E, S, ValueAtom

# Initialize MeTTa space
metta = MeTTa()

# Add knowledge atoms
metta.space().add_atom(E(S("symptom"), S("fever"), S("flu")))
metta.space().add_atom(E(S("treatment"), S("flu"), ValueAtom("rest, fluids, antiviral drugs")))

Key MeTTa Elements:

• E (Expression): Creates logical expressions
• S (Symbol): Represents symbolic atoms
• ValueAtom: Stores actual values (strings, numbers, etc.)
• Space: Container where atoms are stored and queried

2. Pattern Matching and Querying

MeTTa uses pattern matching for knowledge retrieval:

# Query syntax: !(match &self (relation subject $variable) $variable)
query_str = '!(match &self (symptom fever $disease) $disease)'
results = metta.run(query_str)

# Results: [['flu']] - diseases related to fever

Query Components:

• &self: References the current space
• $variable: Pattern matching variables that capture results
• !(match ...): Query syntax for pattern matching

3. uAgent Chat Protocol Integration

The agent implements Fetch.ai's Chat Protocol for ASI:One compatibility:

from uagents_core.contrib.protocols.chat import (
    ChatMessage, ChatAcknowledgement, TextContent, chat_protocol_spec
)

@chat_proto.on_message(ChatMessage)
async def handle_message(ctx: Context, sender: str, msg: ChatMessage):
# Process user query through MeTTa knowledge graph
    response = process_query(user_query, rag, llm)
    await ctx.send(sender, create_text_chat(response))

4. Dynamic Knowledge Learning

The system can learn and add new knowledge dynamically:

class MedicalRAG:
    def add_knowledge(self, relation_type, subject, object_value):
        """Add new knowledge to the MeTTa graph."""
        if isinstance(object_value, str):
            object_value = ValueAtom(object_value)
        self.metta.space().add_atom(E(S(relation_type), S(subject), object_value))

Core Components

1. agent.py: Main uAgent with Chat Protocol implementation
2. knowledge.py: MeTTa knowledge graph initialization
3. medicalrag.py: Domain-specific RAG (Retrieval-Augmented Generation) system
4. utils.py: LLM integration and query processing logic

Implementation Guide

Step 1: Define Your Knowledge Domain

Create your knowledge graph structure in knowledge.py:

from hyperon import MeTTa, E, S, ValueAtom

def initialize_knowledge_graph(metta: MeTTa):
    """Initialize the MeTTa knowledge graph with symptom, treatment, side effect, and FAQ data."""
    # Symptoms → Diseases
    metta.space().add_atom(E(S("symptom"), S("fever"), S("flu")))
    metta.space().add_atom(E(S("symptom"), S("cough"), S("flu")))
    metta.space().add_atom(E(S("symptom"), S("headache"), S("migraine")))
    metta.space().add_atom(E(S("symptom"), S("fatigue"), S("depression")))
    metta.space().add_atom(E(S("symptom"), S("nausea"), S("flu")))
    metta.space().add_atom(E(S("symptom"), S("dizziness"), S("migraine")))
    metta.space().add_atom(E(S("symptom"), S("anxiety"), S("depression")))
    metta.space().add_atom(E(S("symptom"), S("stomach upset"), S("flu")))
    metta.space().add_atom(E(S("symptom"), S("insomnia"), S("depression"))) 
    
    # Diseases → Treatments 
    metta.space().add_atom(E(S("treatment"), S("flu"), ValueAtom("rest, fluids, antiviral drugs")))
    metta.space().add_atom(E(S("treatment"), S("migraine"), ValueAtom("pain relievers, hydration, dark room")))
    metta.space().add_atom(E(S("treatment"), S("depression"), ValueAtom("therapy, antidepressants")))
    metta.space().add_atom(E(S("treatment"), S("anxiety"), ValueAtom("therapy, medications")))  
    metta.space().add_atom(E(S("treatment"), S("fatigue"), ValueAtom("rest, hydration, balanced diet")))
    metta.space().add_atom(E(S("treatment"), S("insomnia"), ValueAtom("sleep hygiene, medications")))
    metta.space().add_atom(E(S("treatment"), S("stomach upset"), ValueAtom("dietary changes, medications")))
    metta.space().add_atom(E(S("treatment"), S("dizziness"), ValueAtom("hydration, medications")))
    metta.space().add_atom(E(S("treatment"), S("pain relievers"), ValueAtom("rest, hydration")))
    metta.space().add_atom(E(S("treatment"), S("antiviral drugs"), ValueAtom("rest, hydration")))
    metta.space().add_atom(E(S("treatment"), S("antidepressants"), ValueAtom("therapy, lifestyle changes")))
    metta.space().add_atom(E(S("treatment"), S("antidepressants"), ValueAtom("therapy, medications")))  
    
    # Treatments → Side Effects
    metta.space().add_atom(E(S("side_effect"), S("antiviral drugs"), ValueAtom("nausea, dizziness")))
    metta.space().add_atom(E(S("side_effect"), S("pain relievers"), ValueAtom("stomach upset")))
    metta.space().add_atom(E(S("side_effect"), S("antidepressants"), ValueAtom("weight gain, insomnia")))
    metta.space().add_atom(E(S("side_effect"), S("therapy"), ValueAtom("initial discomfort, emotional release")))
    
    # FAQs
    metta.space().add_atom(E(S("faq"), S("Hi"), ValueAtom("Hello! How can I assist you today?")))
    metta.space().add_atom(E(S("faq"), S("What’s wrong with me?"), ValueAtom("I’m not a doctor, but I can help you explore symptoms. What are you feeling?")))
    metta.space().add_atom(E(S("faq"), S("How do I treat a migraine?"), ValueAtom("For migraines, rest in a dark room and stay hydrated. Pain relievers can help.")))

Knowledge Structure Explanation:

• Symptoms → Diseases: Links symptoms like "fever" to conditions like "flu"
• Diseases → Treatments: Maps diseases to recommended treatment options
• Treatments → Side Effects: Documents potential side effects of medications
• FAQs: Common medical questions with predefined answers

Step 2: Implement Medical RAG System

Create your RAG system in medicalrag.py :

import re
from hyperon import MeTTa, E, S, ValueAtom

class MedicalRAG:
    def __init__(self, metta_instance: MeTTa):
        self.metta = metta_instance

    def query_symptom(self, symptom):
        """Find diseases linked to a symptom."""
        symptom = symptom.strip('"')
        query_str = f'!(match &self (symptom {symptom} $disease) $disease)'
        results = self.metta.run(query_str)
        print(results,query_str)

        unique_diseases = list(set(str(r[0]) for r in results if r and len(r) > 0)) if results else []
        return unique_diseases

    def get_treatment(self, disease):
        """Find treatments for a disease."""
        disease = disease.strip('"')
        query_str = f'!(match &self (treatment {disease} $treatment) $treatment)'
        results = self.metta.run(query_str)
        print(results,query_str)
        return [r[0].get_object().value for r in results if r and len(r) > 0] if results else []

    def get_side_effects(self, treatment):
        """Find side effects of a treatment."""
        treatment = treatment.strip('"')
        query_str = f'!(match &self (side_effect {treatment} $effect) $effect)'
        results = self.metta.run(query_str)
        print(results,query_str)

        return [r[0].get_object().value for r in results if r and len(r) > 0] if results else []

    def query_faq(self, question):
        """Retrieve FAQ answers."""
        query_str = f'!(match &self (faq "{question}" $answer) $answer)'
        results = self.metta.run(query_str)
        print(results,query_str)

        return results[0][0].get_object().value if results and results[0] else None

    def add_knowledge(self, relation_type, subject, object_value):
        """Add new knowledge dynamically."""
        if isinstance(object_value, str):
            object_value = ValueAtom(object_value)
        self.metta.space().add_atom(E(S(relation_type), S(subject), object_value))
        return f"Added {relation_type}: {subject} → {object_value}"

Key Methods:

• query_symptom(): Finds diseases associated with a symptom
• get_treatment(): Retrieves treatments for a disease
• get_side_effects(): Gets side effects of a treatment
• query_faq(): Answers frequently asked questions
• add_knowledge(): Dynamically adds new medical knowledge

Step 3: Customize Medical Query Processing

Modify utils.py for medical intent classification and response generation:

import json
from openai import OpenAI
from .medicalrag import MedicalRAG

class LLM:
    def __init__(self, api_key):
        self.client = OpenAI(
            api_key=api_key,
            base_url="https://api.asi1.ai/v1"
        )

    def create_completion(self, prompt, max_tokens=200):
        completion = self.client.chat.completions.create(
            messages=[{"role": "user", "content": prompt}],
            model="asi1-mini",  # ASI:One model name
            max_tokens=max_tokens
        )
        return completion.choices[0].message.content

def get_intent_and_keyword(query, llm):
    """Use ASI:One API to classify intent and extract a keyword."""
    prompt = (
        f"Given the query: '{query}'\n"
        "Classify the intent as one of: 'symptom', 'treatment', 'side effect', 'faq', or 'unknown'.\n"
        "Extract the most relevant keyword (e.g., a symptom, disease, or treatment) from the query.\n"
        "Return *only* the result in JSON format like this, with no additional text:\n"
        "{\n"
        "  \"intent\": \"<classified_intent>\",\n"
        "  \"keyword\": \"<extracted_keyword>\"\n"
        "}"
    )
    response = llm.create_completion(prompt)
    try:
        result = json.loads(response)
        return result["intent"], result["keyword"]
    except json.JSONDecodeError:
        print(f"Error parsing ASI:One response: {response}")
        return "unknown", None

def generate_knowledge_response(query, intent, keyword, llm):
    """Use ASI:One to generate a response for new knowledge based on intent."""
    if intent == "symptom":
        prompt = (
            f"Query: '{query}'\n"
            "The symptom '{keyword}' is not in my knowledge base. Suggest a plausible disease it might be linked to.\n"
            "Return *only* the disease name, no additional text."
        )
    elif intent == "treatment":
        prompt = (
            f"Query: '{query}'\n"
            "The disease or condition '{keyword}' has no known treatments in my knowledge base. Suggest a plausible treatment.\n"
            "Return *only* the treatment description, no additional text."
        )
    elif intent == "side effect":
        prompt = (
            f"Query: '{query}'\n"
            "The treatment '{keyword}' has no known side effects in my knowledge base. Suggest plausible side effects.\n"
            "Return *only* the side effects description, no additional text."
        )
    elif intent == "faq":
        prompt = (
            f"Query: '{query}'\n"
            "This is a new FAQ not in my knowledge base. Provide a concise, helpful answer.\n"
            "Return *only* the answer, no additional text."
        )
    else:
        return None
    return llm.create_completion(prompt)

def process_query(query, rag: MedicalRAG, llm: LLM):
    intent, keyword = get_intent_and_keyword(query, llm)
    print(f"Intent: {intent}, Keyword: {keyword}")
    prompt = ""

    if intent == "faq":
        faq_answer = rag.query_faq(query)
        if not faq_answer and keyword:
            new_answer = generate_knowledge_response(query, intent, keyword, llm)
            rag.add_knowledge("faq", query, new_answer)
            print(f"Knowledge graph updated - Added FAQ: '{query}' → '{new_answer}'")
            prompt = (
                f"Query: '{query}'\n"
                f"FAQ Answer: '{new_answer}'\n"
                "Humanize this for a medical assistant with a friendly tone."
            )
        elif faq_answer:
            prompt = (
                f"Query: '{query}'\n"
                f"FAQ Answer: '{faq_answer}'\n"
                "Humanize this for a medical assistant with a friendly tone."
            )
    elif intent == "symptom" and keyword:
        diseases = rag.query_symptom(keyword)
        if not diseases:
            disease = generate_knowledge_response(query, intent, keyword, llm)
            rag.add_knowledge("symptom", keyword, disease)
            print(f"Knowledge graph updated - Added symptom: '{keyword}' → '{disease}'")
            treatments = rag.get_treatment(disease) or ["rest, consult a doctor"]
            side_effects = [rag.get_side_effects(t) for t in treatments] if treatments else []
            prompt = (
                f"Query: '{query}'\n"
                f"Symptom: {keyword}\n"
                f"Related Disease: {disease}\n"
                f"Treatments: {', '.join(treatments)}\n"
                f"Side Effects: {', '.join([', '.join(se) for se in side_effects if se])}\n"
                "Generate a concise, empathetic response for a medical assistant."
            )
        else:
            disease = diseases[0]
            treatments = rag.get_treatment(disease)
            side_effects = [rag.get_side_effects(t) for t in treatments] if treatments else []
            prompt = (
                f"Query: '{query}'\n"
                f"Symptom: {keyword}\n"
                f"Related Disease: {disease}\n"
                f"Treatments: {', '.join(treatments)}\n"
                f"Side Effects: {', '.join([', '.join(se) for se in side_effects if se])}\n"
                "Generate a concise, empathetic response for a medical assistant."
            )
    elif intent == "treatment" and keyword:
        treatments = rag.get_treatment(keyword)
        if not treatments:
            treatment = generate_knowledge_response(query, intent, keyword, llm)
            rag.add_knowledge("treatment", keyword, treatment)
            print(f"Knowledge graph updated - Added treatment: '{keyword}' → '{treatment}'")
            prompt = (
                f"Query: '{query}'\n"
                f"Disease: {keyword}\n"
                f"Treatments: {treatment}\n"
                "Provide a helpful treatment suggestion."
            )
        else:
            prompt = (
                f"Query: '{query}'\n"
                f"Disease: {keyword}\n"
                f"Treatments: {', '.join(treatments)}\n"
                "Provide a helpful treatment suggestion."
            )
    elif intent == "side effect" and keyword:
        side_effects = rag.get_side_effects(keyword)
        if not side_effects:
            side_effect = generate_knowledge_response(query, intent, keyword, llm)
            rag.add_knowledge("side_effect", keyword, side_effect)
            print(f"Knowledge graph updated - Added side effect: '{keyword}' → '{side_effect}'")
            prompt = (
                f"Query: '{query}'\n"
                f"Treatment: {keyword}\n"
                f"Side Effects: {side_effect}\n"
                "Provide a concise explanation of side effects."
            )
        else:
            prompt = (
                f"Query: '{query}'\n"
                f"Treatment: {keyword}\n"
                f"Side Effects: {', '.join(side_effects)}\n"
                "Provide a concise explanation of side effects."
            )
    
    if not prompt:
        prompt = f"Query: '{query}'\nNo specific info found. Offer general assistance."

    prompt += "\nFormat response as: 'Selected Question: <question>' on first line, 'Humanized Answer: <response>' on second."
    response = llm.create_completion(prompt)
    try:
        selected_q = response.split('\n')[0].replace("Selected Question: ", "").strip()
        answer = response.split('\n')[1].replace("Humanized Answer: ", "").strip()
        return {"selected_question": selected_q, "humanized_answer": answer}
    except IndexError:
        return {"selected_question": query, "humanized_answer": response}

Intent Classification:

• symptom: User describes symptoms → find related diseases and treatments
• treatment: User asks about treatments → retrieve treatment options
• side effect: User asks about side effects → find medication side effects
• faq: General medical questions → retrieve from FAQ knowledge base

Step 4: Configure Agent

Update agent.py with medical agent configuration:

from datetime import datetime, timezone
import mailbox
from uuid import uuid4
from typing import Any, Dict
import json
import os
from dotenv import load_dotenv
from uagents import Context, Model, Protocol, Agent
from hyperon import MeTTa

from uagents_core.contrib.protocols.chat import (
    ChatAcknowledgement,
    ChatMessage,
    EndSessionContent,
    StartSessionContent,
    TextContent,
    chat_protocol_spec,
)

# Import components from separate files
from metta.medicalrag import MedicalRAG
from metta.knowledge import initialize_knowledge_graph
from metta.utils import LLM, process_query

# Load environment variables
load_dotenv()

# Initialize agent
agent = Agent(name="Medical MeTTa Agent", 
              seed="Medical MeTTa Agent Seed" 
              port=8005, mailbox=True, 
              publish_agent_details=True)

class MedicalQuery(Model):
    query: str
    intent: str
    keyword: str

def create_text_chat(text: str, end_session: bool = False) -> ChatMessage:
    """Create a text chat message."""
    content = [TextContent(type="text", text=text)]
    if end_session:
        content.append(EndSessionContent(type="end-session"))
    return ChatMessage(
        timestamp=datetime.now(timezone.utc),
        msg_id=uuid4(),
        content=content,
    )

# Initialize global components
metta = MeTTa()
initialize_knowledge_graph(metta)
rag = MedicalRAG(metta)
llm = LLM(api_key=os.getenv("ASI_ONE_API_KEY"))

# Protocol setup
chat_proto = Protocol(spec=chat_protocol_spec)

@chat_proto.on_message(ChatMessage)
async def handle_message(ctx: Context, sender: str, msg: ChatMessage):
    """Handle incoming chat messages and process medical queries."""
    ctx.storage.set(str(ctx.session), sender)
    await ctx.send(
        sender,
        ChatAcknowledgement(timestamp=datetime.now(timezone.utc), acknowledged_msg_id=msg.msg_id),
    )

    for item in msg.content:
        if isinstance(item, StartSessionContent):
            ctx.logger.info(f"Got a start session message from {sender}")
            continue
        elif isinstance(item, TextContent):
            user_query = item.text.strip()
            ctx.logger.info(f"Got a medical query from {sender}: {user_query}")
            
            try:
                # Process the query using the medical assistant logic
                response = process_query(user_query, rag, llm)
                
                # Format the response
                if isinstance(response, dict):
                    answer_text = f"**{response.get('selected_question', user_query)}**\n\n{response.get('humanized_answer', 'I apologize, but I could not process your query.')}"
                else:
                    answer_text = str(response)
                
                # Send the response back
                await ctx.send(sender, create_text_chat(answer_text))
                
            except Exception as e:
                ctx.logger.error(f"Error processing medical query: {e}")
                await ctx.send(
                    sender, 
                    create_text_chat("I apologize, but I encountered an error processing your medical query. Please try again.")
                )
        else:
            ctx.logger.info(f"Got unexpected content from {sender}")

@chat_proto.on_message(ChatAcknowledgement)
async def handle_ack(ctx: Context, sender: str, msg: ChatAcknowledgement):
    """Handle chat acknowledgements."""
    ctx.logger.info(f"Got an acknowledgement from {sender} for {msg.acknowledged_msg_id}")

# Register the protocol
agent.include(chat_proto, publish_manifest=True)

if __name__ == "__main__":
    agent.run()

Agent Features:

• Processes medical queries through MeTTa knowledge graph
• Uses ASI:One LLM for intent classification and response generation
• Dynamically learns new medical knowledge
• Compatible with ASI:One ecosystem via Chat Protocol

Testing and Deployment

Local Testing

1. Start the agent:

   python agent.py

• Visit the inspector URL from console output
• Connect via Mailbox
• Test via Chat Interface: Use "Chat with Agent" button

Query your Agent from ASI1 LLM

1. Login to the ASI1 LLM, either using your Google Account or the ASI1 Wallet and Start a New Chat.
2. Toggle the "Agents" switch to enable ASI1 to connect with Agents on Agentverse.