Financial Advisor 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 an intelligent, reasoning-capable financial advisor agent. This integration enables the agent to process investment queries using structured reasoning, pattern matching, and dynamic graph updates while remaining compatible 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 graphs. It provides:
- Structured Knowledge Representation: Organize financial data in logical, queryable relations
- Symbolic Reasoning: Perform pattern-based matching and retrieval
- Knowledge Graph Operations: Build, query, and evolve graph memory
- Space-based Architecture: Knowledge stored as atoms in logical spaces
Installation & Setup
Prerequisites
Before you begin, ensure you have:
- Python 3.11+ installed
- pip package manager
- an ASI:One API key (from https://asi1.ai/)
Installation Options
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
Architecture Overview
Working flow (matches the code path): Chat Protocol receives the investor message → ASI:One returns intent + keyword → InvestmentRAG runs MeTTa pattern queries (and may add atoms when data is missing) → a final ASI:One completion turns retrieved facts into risk-aware guidance → reply goes back through chat to the user.
Core Integration Concepts
1. MeTTa Knowledge Graph Structure
The financial advisor agent organizes investment knowledge in graph relations:
from hyperon import MeTTa, E, S, ValueAtom
def initialize_investment_knowledge(metta: MeTTa):
"""Initialize the MeTTa knowledge graph with investment, risk, portfolio, and strategy data."""
# Risk Profile → Investment Types
metta.space().add_atom(E(S("risk_profile"), S("conservative"), S("bonds")))
metta.space().add_atom(E(S("risk_profile"), S("conservative"), S("dividend_stocks")))
metta.space().add_atom(E(S("risk_profile"), S("moderate"), S("index_funds")))
metta.space().add_atom(E(S("risk_profile"), S("moderate"), S("etfs")))
metta.space().add_atom(E(S("risk_profile"), S("aggressive"), S("growth_stocks")))
metta.space().add_atom(E(S("risk_profile"), S("aggressive"), S("cryptocurrency")))
metta.space().add_atom(E(S("risk_profile"), S("conservative"), S("savings_accounts")))
metta.space().add_atom(E(S("risk_profile"), S("moderate"), S("real_estate")))
metta.space().add_atom(E(S("risk_profile"), S("aggressive"), S("options")))
# Investment Types → Expected Returns
metta.space().add_atom(E(S("expected_return"), S("bonds"), ValueAtom("3-5% annually")))
metta.space().add_atom(E(S("expected_return"), S("dividend_stocks"), ValueAtom("5-7% annually")))
metta.space().add_atom(E(S("expected_return"), S("index_funds"), ValueAtom("6-10% annually")))
metta.space().add_atom(E(S("expected_return"), S("etfs"), ValueAtom("5-12% annually")))
metta.space().add_atom(E(S("expected_return"), S("growth_stocks"), ValueAtom("8-15% annually")))
metta.space().add_atom(E(S("expected_return"), S("cryptocurrency"), ValueAtom("highly volatile, -50% to +200%")))
metta.space().add_atom(E(S("expected_return"), S("savings_accounts"), ValueAtom("1-2% annually")))
metta.space().add_atom(E(S("expected_return"), S("real_estate"), ValueAtom("4-8% annually")))
metta.space().add_atom(E(S("expected_return"), S("options"), ValueAtom("high risk, unlimited gains/losses")))
# Investment Types → Risk Levels
metta.space().add_atom(E(S("risk_level"), S("bonds"), ValueAtom("low risk, stable income")))
metta.space().add_atom(E(S("risk_level"), S("dividend_stocks"), ValueAtom("low-moderate risk, regular dividends")))
metta.space().add_atom(E(S("risk_level"), S("index_funds"), ValueAtom("moderate risk, diversified")))
metta.space().add_atom(E(S("risk_level"), S("etfs"), ValueAtom("low-moderate risk, liquid")))
metta.space().add_atom(E(S("risk_level"), S("growth_stocks"), ValueAtom("high risk, growth potential")))
metta.space().add_atom(E(S("risk_level"), S("cryptocurrency"), ValueAtom("very high risk, extreme volatility")))
metta.space().add_atom(E(S("risk_level"), S("savings_accounts"), ValueAtom("no risk, FDIC insured")))
metta.space().add_atom(E(S("risk_level"), S("real_estate"), ValueAtom("moderate risk, inflation hedge")))
metta.space().add_atom(E(S("risk_level"), S("options"), ValueAtom("very high risk, leveraged exposure")))
# Age Group → Recommended Asset Allocation
metta.space().add_atom(E(S("age_allocation"), S("20s"), ValueAtom("80% stocks, 20% bonds")))
metta.space().add_atom(E(S("age_allocation"), S("30s"), ValueAtom("70% stocks, 30% bonds")))
metta.space().add_atom(E(S("age_allocation"), S("40s"), ValueAtom("60% stocks, 40% bonds")))
metta.space().add_atom(E(S("age_allocation"), S("50s"), ValueAtom("50% stocks, 50% bonds")))
metta.space().add_atom(E(S("age_allocation"), S("60s"), ValueAtom("40% stocks, 60% bonds")))
# Investment Goals → Strategies
metta.space().add_atom(E(S("goal_strategy"), S("retirement"), ValueAtom("diversified index funds, 401k maxing")))
metta.space().add_atom(E(S("goal_strategy"), S("emergency_fund"), ValueAtom("high-yield savings, money market")))
metta.space().add_atom(E(S("goal_strategy"), S("house_down_payment"), ValueAtom("CDs, short-term bonds")))
metta.space().add_atom(E(S("goal_strategy"), S("wealth_building"), ValueAtom("growth stocks, REITs")))
metta.space().add_atom(E(S("goal_strategy"), S("passive_income"), ValueAtom("dividend stocks, bonds")))
# Market Sectors → Top Performers
metta.space().add_atom(E(S("sector_stocks"), S("technology"), ValueAtom("Apple, Microsoft, Google")))
metta.space().add_atom(E(S("sector_stocks"), S("healthcare"), ValueAtom("Johnson & Johnson, Pfizer")))
metta.space().add_atom(E(S("sector_stocks"), S("finance"), ValueAtom("JPMorgan Chase, Berkshire Hathaway")))
metta.space().add_atom(E(S("sector_stocks"), S("energy"), ValueAtom("ExxonMobil, Chevron")))
# Common Investment Mistakes → Warnings
metta.space().add_atom(E(S("mistake"), S("timing_market"), ValueAtom("avoid trying to time market peaks and valleys")))
metta.space().add_atom(E(S("mistake"), S("lack_diversification"), ValueAtom("don't put all money in one stock or sector")))
metta.space().add_atom(E(S("mistake"), S("emotional_trading"), ValueAtom("avoid panic selling or FOMO buying")))
metta.space().add_atom(E(S("mistake"), S("high_fees"), ValueAtom("watch out for expensive mutual fund fees")))
# Investment FAQs
metta.space().add_atom(E(S("faq"), S("How much should I invest?"), ValueAtom("Invest 10-20% of income after emergency fund")))
metta.space().add_atom(E(S("faq"), S("When should I start investing?"), ValueAtom("Start as early as possible for compound growth")))
metta.space().add_atom(E(S("faq"), S("What is diversification?"), ValueAtom("Spreading investments across different assets to reduce risk")))
metta.space().add_atom(E(S("faq"), S("Should I pay off debt first?"), ValueAtom("Pay off high-interest debt before investing")))
2. Pattern Matching and Querying
The RAG layer queries MeTTa relations through symbolic matches:
import re
from hyperon import MeTTa, E, S, ValueAtom
class InvestmentRAG:
def __init__(self, metta_instance: MeTTa):
self.metta = metta_instance
def query_risk_profile(self, risk_profile):
"""Find investment types suitable for a risk profile."""
risk_profile = risk_profile.strip('"')
query_str = f'!(match &self (risk_profile {risk_profile} $investment) $investment)'
results = self.metta.run(query_str)
print(results, query_str)
unique_investments = list(set(str(r[0]) for r in results if r and len(r) > 0)) if results else []
return unique_investments
def get_expected_return(self, investment):
"""Find expected returns for an investment type."""
investment = investment.strip('"')
query_str = f'!(match &self (expected_return {investment} $return) $return)'
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_risk_level(self, investment):
"""Find risk level of an investment type."""
investment = investment.strip('"')
query_str = f'!(match &self (risk_level {investment} $risk) $risk)'
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_age_allocation(self, age_group):
"""Get recommended asset allocation for age group."""
age_group = age_group.strip('"')
query_str = f'!(match &self (age_allocation {age_group} $allocation) $allocation)'
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_goal_strategy(self, goal):
"""Get investment strategy for a specific goal."""
goal = goal.strip('"')
query_str = f'!(match &self (goal_strategy {goal} $strategy) $strategy)'
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_sector_stocks(self, sector):
"""Get top performing stocks in a sector."""
sector = sector.strip('"')
query_str = f'!(match &self (sector_stocks {sector} $stocks) $stocks)'
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_mistake_warning(self, mistake):
"""Get warning about common investment mistakes."""
mistake = mistake.strip('"')
query_str = f'!(match &self (mistake {mistake} $warning) $warning)'
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 investment 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 investment 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}"
3. uAgent Chat Protocol Integration
The agent uses Fetch.ai Chat Protocol to process user investment requests:
from datetime import datetime, timezone
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,
)
from metta.investment_rag import InvestmentRAG
from metta.knowledge import initialize_investment_knowledge
from metta.utils import LLM, process_query
load_dotenv()
agent = Agent(name="Financial Investment Advisor", seed="financial-investment-advisor-seed-1", port=8008, mailbox=True, publish_agent_details=True)
class InvestmentQuery(Model):
query: str
intent: str
keyword: str
def create_text_chat(text: str, end_session: bool = False) -> ChatMessage:
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,
)
metta = MeTTa()
initialize_investment_knowledge(metta)
rag = InvestmentRAG(metta)
llm = LLM(api_key=os.getenv("ASI_ONE_API_KEY"))
chat_proto = Protocol(spec=chat_protocol_spec)
@chat_proto.on_message(ChatMessage)
async def handle_message(ctx: Context, sender: str, msg: ChatMessage):
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 an investment query from {sender}: {user_query}")
try:
response = process_query(user_query, rag, llm)
if isinstance(response, dict):
answer_text = response.get('humanized_answer', 'I apologize, but I could not process your query.')
else:
answer_text = str(response)
await ctx.send(sender, create_text_chat(answer_text))
except Exception as e:
ctx.logger.error(f"Error processing investment query: {e}")
await ctx.send(
sender,
create_text_chat("I apologize, but I encountered an error processing your investment 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):
ctx.logger.info(f"Got an acknowledgement from {sender} for {msg.acknowledged_msg_id}")
agent.include(chat_proto, publish_manifest=True)
if __name__ == "__main__":
agent.run()
4. ASI:One Intent Classification and Response Layer
The assistant classifies intent and generates humanized financial responses:
import json
from openai import OpenAI
from .investment_rag import InvestmentRAG
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",
max_tokens=max_tokens
)
return completion.choices[0].message.content
def get_intent_and_keyword(query, llm):
"""Use ASI:One API to classify investment intent and extract a keyword."""
prompt = (
f"Given the investment query: '{query}'\n"
"Classify the intent as one of: 'risk_profile', 'investment_advice', 'returns', 'allocation', 'goal', 'sector', 'mistake', 'faq', or 'unknown'.\n"
"Extract the most relevant keyword (e.g., conservative, aggressive, retirement, technology, bonds) 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:
cleaned = response.strip()
if cleaned.startswith("```"):
cleaned = "\n".join(cleaned.split("\n")[1:])
if cleaned.endswith("```"):
cleaned = "\n".join(cleaned.split("\n")[:-1])
result = json.loads(cleaned.strip())
return result["intent"], result["keyword"]
except (json.JSONDecodeError, KeyError):
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 == "risk_profile":
prompt = (
f"Query: '{query}'\n"
f"The risk profile '{keyword}' is not in my knowledge base. Suggest plausible investment types for this risk level.\n"
f"Return *only* the investment types, no additional text."
)
elif intent == "investment_advice":
prompt = (
f"Query: '{query}'\n"
f"The investment type '{keyword}' has no specific advice in my knowledge base. Provide general investment guidance.\n"
f"Return *only* the advice, no additional text."
)
elif intent == "returns":
prompt = (
f"Query: '{query}'\n"
f"The investment '{keyword}' has no expected return data in my knowledge base. Suggest realistic return expectations.\n"
f"Return *only* the return information, no additional text."
)
elif intent == "allocation":
prompt = (
f"Query: '{query}'\n"
f"The age group '{keyword}' has no allocation strategy in my knowledge base. Suggest appropriate asset allocation.\n"
f"Return *only* the allocation recommendation, no additional text."
)
elif intent == "goal":
prompt = (
f"Query: '{query}'\n"
f"The investment goal '{keyword}' has no strategy in my knowledge base. Suggest appropriate investment approaches.\n"
f"Return *only* the strategy, no additional text."
)
elif intent == "sector":
prompt = (
f"Query: '{query}'\n"
f"The sector '{keyword}' has no stock list in my knowledge base. Suggest a few plausible example companies or tickers for this sector.\n"
f"Return *only* a short comma-separated list or one sentence, no additional text."
)
elif intent == "mistake":
prompt = (
f"Query: '{query}'\n"
f"The mistake or behavior '{keyword}' is not in my knowledge base. Give one short, practical warning for investors.\n"
f"Return *only* the warning text, no additional text."
)
elif intent == "faq":
prompt = (
f"Query: '{query}'\n"
"This is a new investment question not in my knowledge base. Provide a helpful, concise answer.\n"
"Return *only* the answer, no additional text."
)
else:
return None
return llm.create_completion(prompt)
def process_query(query, rag: InvestmentRAG, 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"Investment Answer: '{new_answer}'\n"
"Provide this as professional investment guidance with appropriate disclaimers."
)
elif faq_answer:
prompt = (
f"Query: '{query}'\n"
f"Investment Answer: '{faq_answer}'\n"
"Provide this as professional investment guidance with appropriate disclaimers."
)
elif intent == "risk_profile" and keyword:
investments = rag.query_risk_profile(keyword)
if not investments:
investment_types = generate_knowledge_response(query, intent, keyword, llm)
rag.add_knowledge("risk_profile", keyword, investment_types)
print(f"Knowledge graph updated - Added risk profile: '{keyword}' → '{investment_types}'")
prompt = (
f"Query: '{query}'\n"
f"Risk Profile: {keyword}\n"
f"Suitable Investments: {investment_types}\n"
"Provide professional investment recommendations with risk disclaimers."
)
else:
investment_details = []
for investment in investments:
returns = rag.get_expected_return(investment)
risks = rag.get_risk_level(investment)
investment_details.append({
'type': investment,
'returns': returns[0] if returns else 'N/A',
'risks': risks[0] if risks else 'N/A'
})
prompt = (
f"Query: '{query}'\n"
f"Risk Profile: {keyword}\n"
f"Investment Options: {investment_details}\n"
"Provide professional investment recommendations with expected returns and risk analysis."
)
elif intent == "returns" and keyword:
returns = rag.get_expected_return(keyword)
risks = rag.get_risk_level(keyword)
if not returns:
return_info = generate_knowledge_response(query, intent, keyword, llm)
rag.add_knowledge("expected_return", keyword, return_info)
print(f"Knowledge graph updated - Added returns: '{keyword}' → '{return_info}'")
prompt = (
f"Query: '{query}'\n"
f"Investment: {keyword}\n"
f"Expected Returns: {return_info}\n"
"Provide return analysis with appropriate risk warnings."
)
else:
prompt = (
f"Query: '{query}'\n"
f"Investment: {keyword}\n"
f"Expected Returns: {', '.join(returns)}\n"
f"Risk Level: {', '.join(risks) if risks else 'Not specified'}\n"
"Provide comprehensive return and risk analysis."
)
elif intent == "allocation" and keyword:
allocation = rag.get_age_allocation(keyword)
if not allocation:
allocation_advice = generate_knowledge_response(query, intent, keyword, llm)
rag.add_knowledge("age_allocation", keyword, allocation_advice)
print(f"Knowledge graph updated - Added allocation: '{keyword}' → '{allocation_advice}'")
prompt = (
f"Query: '{query}'\n"
f"Age Group: {keyword}\n"
f"Recommended Allocation: {allocation_advice}\n"
"Provide age-appropriate asset allocation guidance."
)
else:
prompt = (
f"Query: '{query}'\n"
f"Age Group: {keyword}\n"
f"Recommended Allocation: {', '.join(allocation)}\n"
"Explain the allocation strategy and rationale."
)
elif intent == "goal" and keyword:
strategies = rag.get_goal_strategy(keyword)
if not strategies:
strategy = generate_knowledge_response(query, intent, keyword, llm)
rag.add_knowledge("goal_strategy", keyword, strategy)
print(f"Knowledge graph updated - Added goal strategy: '{keyword}' → '{strategy}'")
prompt = (
f"Query: '{query}'\n"
f"Investment Goal: {keyword}\n"
f"Recommended Strategy: {strategy}\n"
"Provide goal-oriented investment guidance."
)
else:
prompt = (
f"Query: '{query}'\n"
f"Investment Goal: {keyword}\n"
f"Recommended Strategies: {', '.join(strategies)}\n"
"Provide comprehensive goal-based investment planning."
)
elif intent == "sector" and keyword:
stocks = rag.query_sector_stocks(keyword)
if not stocks:
sector_info = generate_knowledge_response(query, intent, keyword, llm)
if not sector_info:
sector_info = (
"No specific names generated; consider sector index funds or ETFs."
)
else:
rag.add_knowledge("sector_stocks", keyword, sector_info)
print(f"Knowledge graph updated - Added sector: '{keyword}' → '{sector_info}'")
prompt = (
f"Query: '{query}'\n"
f"Sector: {keyword}\n"
f"Investment Options: {sector_info}\n"
"Provide sector-specific investment analysis."
)
else:
prompt = (
f"Query: '{query}'\n"
f"Sector: {keyword}\n"
f"Top Performers: {', '.join(stocks)}\n"
"Provide sector analysis and investment recommendations."
)
elif intent == "investment_advice" and keyword:
exp = rag.get_expected_return(keyword)
risks = rag.get_risk_level(keyword)
if exp or risks:
prompt = (
f"Query: '{query}'\n"
f"Investment type: {keyword}\n"
f"Expected return (knowledge base): {', '.join(exp) if exp else 'N/A'}\n"
f"Risk notes (knowledge base): {', '.join(risks) if risks else 'N/A'}\n"
"Synthesize clear investment guidance with appropriate disclaimers."
)
else:
advice = generate_knowledge_response(query, intent, keyword, llm)
if not advice:
advice = (
"Consider diversified, low-cost options aligned with your horizon; "
"consult a licensed professional for personal advice."
)
prompt = (
f"Query: '{query}'\n"
f"Investment type: {keyword}\n"
f"Guidance: {advice}\n"
"Provide professional investment guidance with disclaimers."
)
elif intent == "mistake" and keyword:
warnings = rag.get_mistake_warning(keyword)
if warnings:
wtext = ", ".join(warnings)
else:
wtext = generate_knowledge_response(query, intent, keyword, llm)
if not wtext:
wtext = (
"Prioritize diversification, discipline, and awareness of fees and emotions."
)
else:
rag.add_knowledge("mistake", keyword, wtext)
print(f"Knowledge graph updated - Added mistake: '{keyword}' → '{wtext}'")
prompt = (
f"Query: '{query}'\n"
f"Topic / behavior: {keyword}\n"
f"Warning(s): {wtext}\n"
"Explain clearly for retail investors with professional disclaimers."
)
if not prompt:
prompt = f"Query: '{query}'\nNo specific investment information found. Provide general investment guidance and suggest consulting a financial advisor."
prompt += "\nProvide a clear, professional investment response with appropriate disclaimers about consulting financial professionals. Do not repeat the query."
response = llm.create_completion(prompt, max_tokens=300)
return {"selected_question": query, "humanized_answer": response.strip()}
Core Components
agent.py: Main uAgent with Chat Protocol implementationknowledge.py: Financial graph initialization (risk, returns, allocation, goals)investment_rag.py: Investment-specific RAG retrieval methodsutils.py: ASI:One intent detection and response generation
Implementation Guide
Step 1: Define Your Financial Knowledge Domain
Use knowledge.py to encode relationships like:
- risk profile -> investment type
- investment type -> expected return
- investment type -> risk level
- age group -> portfolio allocation
- goal -> strategy
Step 2: Implement Financial RAG Retrieval
Use investment_rag.py methods to query risk, returns, allocation, and FAQ data.
Step 3: Customize Investment Query Processing
Use utils.py to classify intents (risk_profile, investment_advice, returns, allocation, goal, sector, mistake, faq) and generate final responses.
Step 4: Configure Agent Runtime
Use agent.py to initialize MeTTa, wire InvestmentRAG, and serve Chat Protocol handlers.
Detailed Working (Step-by-Step)
- User sends an investment query via ASI:One chat.
- ASI:One extracts intent and keyword.
InvestmentRAGexecutes MeTTa pattern matches.- Missing knowledge can be generated and added to graph memory.
- Agent responds with concise, risk-aware investment guidance.
Testing and Deployment
Local Testing
- Start the agent:
python agent.py
- Open inspector URL from console output
- Connect via Mailbox
- Test through chat interface
Sample Financial Queries
- "I am a conservative investor, what should I invest in?"
- "What returns can I expect from index funds?"
- "How should a 30-year-old allocate a portfolio?"
- "What strategy works best for retirement?"
- "What are common investing mistakes to avoid?"
Query your Agent from ASI1 LLM
- Login to ASI1 LLM
- Start a new chat and enable agents
- Query your Financial Advisor agent directly
Notes
- This example is educational and not personalized financial advice.
- For production usage, add compliance rules, disclosure controls, and policy guardrails.