On this tutorial, we delve into constructing a sophisticated AI agent with agent reminiscence utilizing Cognee and Hugging Face fashions, using totally free, open-source instruments that work seamlessly in Google Colab and different pocket book. We configure Cognee for reminiscence storage and retrieval, combine a light-weight conversational mannequin for producing responses, and convey all of it collectively into an clever agent that learns, causes, and interacts naturally. Whether or not it’s processing paperwork throughout domains or partaking in dialogue with contextual understanding, we stroll by means of every step to create a succesful agent with out counting on paid APIs. Take a look at the Full Codes right here. Be happy to test different AI Agent and Agentic AI Codes and Tutorial for varied functions.
!pip set up cognee transformers torch sentence-transformers speed up
import asyncio
import os
import json
from typing import Checklist, Dict, Any
from datetime import datetime
from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
import torch
import cogneeWe start by putting in all of the important libraries, together with Cognee, Transformers, Torch, and Sentence-Transformers, to energy our AI agent. We then import the required modules to deal with tokenization, mannequin loading, asynchronous duties, and reminiscence integration. This setup ensures we now have all the pieces able to construct, practice, and work together with our clever agent. Take a look at the Full Codes right here. Be happy to test different AI Agent and Agentic AI Codes and Tutorial for varied functions.
async def setup_cognee():
"""Setup Cognee with correct configuration"""
strive:
await cognee.config.set("EMBEDDING_MODEL", "sentence-transformers/all-MiniLM-L6-v2")
await cognee.config.set("EMBEDDING_PROVIDER", "sentence_transformers")
print("✅ Cognee configured efficiently")
return True
besides Exception as e:
print(f"⚠️ Cognee config error: {e}")
strive:
os.environ["EMBEDDING_MODEL"] = "sentence-transformers/all-MiniLM-L6-v2"
os.environ["EMBEDDING_PROVIDER"] = "sentence_transformers"
print("✅ Cognee configured by way of surroundings")
return True
besides Exception as e2:
print(f"⚠️ Various config failed: {e2}")
return FalseWe arrange Cognee by configuring the embedding mannequin and supplier to make use of all-MiniLM-L6-v2, a light-weight and environment friendly sentence-transformer. If the first technique fails, we fall again to manually setting surroundings variables, making certain Cognee is all the time able to course of and retailer embeddings. Take a look at the Full Codes right here. Be happy to test different AI Agent and Agentic AI Codes and Tutorial for varied functions.
class HuggingFaceLLM:
def __init__(self, model_name="microsoft/DialoGPT-medium"):
print(f"🤖 Loading Hugging Face mannequin: {model_name}")
self.system = "cuda" if torch.cuda.is_available() else "cpu"
print(f"📱 Utilizing system: {self.system}")
if "DialoGPT" in model_name:
self.tokenizer = AutoTokenizer.from_pretrained(model_name, padding_side="left")
self.mannequin = AutoModelForCausalLM.from_pretrained(model_name)
if self.tokenizer.pad_token is None:
self.tokenizer.pad_token = self.tokenizer.eos_token
else:
self.generator = pipeline(
"text-generation",
mannequin="distilgpt2",
system=0 if self.system == "cuda" else -1,
max_length=150,
do_sample=True,
temperature=0.7
)
self.tokenizer = None
self.mannequin = None
print("✅ Mannequin loaded efficiently!")
def generate_response(self, immediate: str, max_length: int = 100) -> str:
strive:
if self.mannequin is just not None:
inputs = self.tokenizer.encode(immediate + self.tokenizer.eos_token, return_tensors="pt")
with torch.no_grad():
outputs = self.mannequin.generate(
inputs,
max_length=inputs.form[1] + max_length,
num_return_sequences=1,
temperature=0.7,
do_sample=True,
pad_token_id=self.tokenizer.eos_token_id
)
response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
response = response[len(prompt):].strip()
return response if response else "I perceive."
else:
outcome = self.generator(immediate, max_length=max_length, truncation=True)
return outcome[0]['generated_text'][len(prompt):].strip()
besides Exception as e:
print(f"⚠️ Technology error: {e}")
return "I am processing that data."
hf_llm = NoneWe outline the HuggingFaceLLM class to deal with textual content era utilizing light-weight Hugging Face fashions, corresponding to DialoGPT or DistilGPT2. We detect whether or not a GPU is on the market and cargo the suitable tokenizer and mannequin accordingly. This setup permits our agent to generate clever and context-aware responses to person queries. Take a look at the Full Codes right here. Be happy to test different AI Agent and Agentic AI Codes and Tutorial for varied functions.
class AdvancedAIAgent:
"""
Superior AI Agent with persistent reminiscence, studying capabilities,
and multi-domain data processing utilizing Cognee
"""
def __init__(self, agent_name: str = "CogneeAgent"):
self.identify = agent_name
self.memory_initialized = False
self.knowledge_domains = []
self.conversation_history = []
self.manual_memory = []
async def initialize_memory(self):
"""Initialize the agent's reminiscence system and HF mannequin"""
world hf_llm
if hf_llm is None:
hf_llm = HuggingFaceLLM("microsoft/DialoGPT-medium")
setup_success = await setup_cognee()
strive:
await cognee.prune()
print(f"✅ {self.identify} reminiscence system initialized")
self.memory_initialized = True
besides Exception as e:
print(f"⚠️ Reminiscence initialization warning: {e}")
self.memory_initialized = True
async def learn_from_text(self, textual content: str, area: str = "common"):
"""Add data to the agent's reminiscence with area tagging"""
if not self.memory_initialized:
await self.initialize_memory()
enhanced_text = f"[DOMAIN: {domain}] [TIMESTAMP: {datetime.now().isoformat()}]n{textual content}"
strive:
await cognee.add(enhanced_text)
await cognee.cognify()
if area not in self.knowledge_domains:
self.knowledge_domains.append(area)
print(f"📚 Realized new data in area: {area}")
return True
besides Exception as e:
print(f"❌ Studying error: {e}")
strive:
await cognee.add(textual content)
await cognee.cognify()
if area not in self.knowledge_domains:
self.knowledge_domains.append(area)
print(f"📚 Realized (simplified): {area}")
return True
besides Exception as e2:
print(f"❌ Simplified studying failed: {e2}")
if not hasattr(self, 'manual_memory'):
self.manual_memory = []
self.manual_memory.append({"textual content": textual content, "area": area})
if area not in self.knowledge_domains:
self.knowledge_domains.append(area)
print(f"📚 Saved in guide reminiscence: {area}")
return True
async def learn_from_documents(self, paperwork: Checklist[Dict[str, str]]):
"""Batch studying from a number of paperwork"""
print(f"📖 Processing {len(paperwork)} paperwork...")
for i, doc in enumerate(paperwork):
textual content = doc.get("content material", "")
area = doc.get("area", "common")
title = doc.get("title", f"Document_{i+1}")
enhanced_content = f"Title: {title}n{textual content}"
await self.learn_from_text(enhanced_content, area)
if i % 3 == 0:
print(f" Processed {i+1}/{len(paperwork)} paperwork")
async def query_knowledge(self, query: str, domain_filter: str = None) -> Checklist[str]:
"""Question the agent's data base with non-obligatory area filtering"""
strive:
if domain_filter:
enhanced_query = f"[DOMAIN: {domain_filter}] {query}"
else:
enhanced_query = query
search_results = await cognee.search("SIMILARITY", enhanced_query)
outcomes = []
for lead to search_results:
if hasattr(outcome, 'textual content'):
outcomes.append(outcome.textual content)
elif hasattr(outcome, 'content material'):
outcomes.append(outcome.content material)
elif hasattr(outcome, 'worth'):
outcomes.append(str(outcome.worth))
elif isinstance(outcome, dict):
content material = outcome.get('textual content') or outcome.get('content material') or outcome.get('knowledge') or outcome.get('worth')
if content material:
outcomes.append(str(content material))
else:
outcomes.append(str(outcome))
elif isinstance(outcome, str):
outcomes.append(outcome)
else:
result_str = str(outcome)
if len(result_str) > 10:
outcomes.append(result_str)
if not outcomes and hasattr(self, 'manual_memory'):
for merchandise in self.manual_memory:
if domain_filter and merchandise['domain'] != domain_filter:
proceed
if any(phrase.decrease() in merchandise['text'].decrease() for phrase in query.cut up()):
outcomes.append(merchandise['text'])
return outcomes[:5]
besides Exception as e:
print(f"🔍 Search error: {e}")
outcomes = []
if hasattr(self, 'manual_memory'):
for merchandise in self.manual_memory:
if domain_filter and merchandise['domain'] != domain_filter:
proceed
if any(phrase.decrease() in merchandise['text'].decrease() for phrase in query.cut up()):
outcomes.append(merchandise['text'])
return outcomes[:5]
async def reasoning_chain(self, query: str) -> Dict[str, Any]:
"""Superior reasoning utilizing retrieved data"""
print(f"🤔 Processing query: {query}")
relevant_info = await self.query_knowledge(query)
evaluation = {
"query": query,
"relevant_knowledge": relevant_info,
"domains_searched": self.knowledge_domains,
"confidence": min(len(relevant_info) / 3.0, 1.0),
"timestamp": datetime.now().isoformat()
}
if relevant_info and len(relevant_info) > 0:
reasoning = self._synthesize_answer(query, relevant_info)
evaluation["reasoning"] = reasoning
evaluation["answer"] = self._extract_key_points(relevant_info)
else:
evaluation["reasoning"] = "No related data present in reminiscence"
evaluation["answer"] = "I haven't got details about this matter in my present data base."
return evaluation
def _synthesize_answer(self, query: str, knowledge_pieces: Checklist[str]) -> str:
"""AI-powered reply synthesis utilizing Hugging Face mannequin"""
world hf_llm
if not knowledge_pieces:
return "No related data present in my data base."
context = " ".be a part of(knowledge_pieces[:2])
context = context[:300]
immediate = f"Based mostly on this data: {context}nnQuestion: {query}nAnswer:"
strive:
if hf_llm:
synthesized = hf_llm.generate_response(immediate, max_length=80)
return synthesized if synthesized else f"Based mostly on my data: {context[:100]}..."
else:
return f"From my evaluation: {context[:150]}..."
besides Exception as e:
print(f"⚠️ Synthesis error: {e}")
return f"Based mostly on my data: {context[:100]}..."
def _extract_key_points(self, knowledge_pieces: Checklist[str]) -> Checklist[str]:
"""Extract key factors from retrieved data"""
key_points = []
for piece in knowledge_pieces:
clean_piece = piece.change("[DOMAIN:", "").replace("[TIMESTAMP:", "")
sentences = clean_piece.split('.')
if len(sentences) > 0 and len(sentences[0].strip()) > 10:
key_points.append(sentences[0].strip() + ".")
return key_points[:3]
async def conversational_agent(self, user_input: str) -> str:
"""Most important conversational interface with HF mannequin integration"""
world hf_llm
self.conversation_history.append({"position": "person", "content material": user_input})
if any(phrase in user_input.decrease() for phrase in ["learn", "remember", "add", "teach"]):
content_to_learn = user_input.change("study this:", "").change("bear in mind:", "").strip()
await self.learn_from_text(content_to_learn, "dialog")
response = "I've saved that data in my reminiscence! What else would you want to show me?"
elif user_input.decrease().startswith(("what", "how", "why", "when", "the place", "who", "inform me")):
evaluation = await self.reasoning_chain(user_input)
if evaluation["relevant_knowledge"] and hf_llm:
context = " ".be a part of(evaluation["relevant_knowledge"][:2])[:200]
immediate = f"Query: {user_input}nKnowledge: {context}nFriendly response:"
ai_response = hf_llm.generate_response(immediate, max_length=60)
response = ai_response if ai_response else "Here is what I discovered in my data base."
else:
response = "I haven't got particular details about that matter in my present data base."
else:
relevant_context = await self.query_knowledge(user_input)
if hf_llm:
context_info = ""
if relevant_context:
context_info = f" I do know that: {relevant_context[0][:100]}..."
conversation_prompt = f"Person says: {user_input}{context_info}nI reply:"
response = hf_llm.generate_response(conversation_prompt, max_length=50)
if not response or len(response.strip()) We now outline the core of our system, the AdvancedAIAgent class, which brings collectively Cognee’s reminiscence, domain-aware studying, data retrieval, and Hugging Face-powered reasoning. We empower our agent to study from each textual content and paperwork, retrieve contextually related data, and reply to queries with synthesized, clever solutions. Whether or not it’s remembering info, answering questions, or partaking in dialog, this agent adapts, remembers, and responds with human-like fluency. Take a look at the Full Codes right here. Be happy to test different AI Agent and Agentic AI Codes and Tutorial for varied functions.
async def most important():
print("🚀 Superior AI Agent with Cognee Tutorial")
print("=" * 50)
agent = AdvancedAIAgent("TutorialAgent")
await agent.initialize_memory()
print("n📚 DEMO 1: Multi-domain Studying")
sample_documents = [
{
"title": "Python Basics",
"content": "Python is a high-level programming language known for its simplicity and readability.",
"domain": "programming"
},
{
"title": "Climate Science",
"content": "Climate change",
"domain": "science"
},
{
"title": "AI Ethics",
"content": "AI ethics involves ensuring artificial intelligence systems are developed and deployed responsibly, considering fairness, transparency, accountability, and potential societal impacts.",
"domain": "technology"
},
{
"title": "Sustainable Energy",
"content": "Renewable energy sources are crucial for reducing carbon emissions",
"domain": "environment"
}
]
await agent.learn_from_documents(sample_documents)
print("n🔍 DEMO 2: Data Retrieval & Reasoning")
test_questions = [
"What do you know about Python programming?",
"How does climate change relate to energy?",
"What are the ethical considerations in AI?"
]
for query in test_questions:
print(f"n❓ Query: {query}")
evaluation = await agent.reasoning_chain(query)
print(f"💡 Reply: {evaluation.get('reply', 'No reply generated')}")
print(f"🎯 Confidence: {evaluation.get('confidence', 0):.2f}")
print("n💬 DEMO 3: Conversational Agent")
conversation_inputs = [
"Learn this: Machine learning is a subset of AI",
"What is machine learning?",
"How does it relate to Python?",
"Remember that neural networks are inspired by biological neurons"
]
for user_input in conversation_inputs:
print(f"n🗣️ Person: {user_input}")
response = await agent.conversational_agent(user_input)
print(f"🤖 Agent: {response}")
print(f"n📊 DEMO 4: Agent Data Abstract")
print(f"Data domains: {agent.knowledge_domains}")
print(f"Dialog historical past: {len(agent.conversation_history)} exchanges")
print(f"n🎯 Area-specific search:")
programming_results = await agent.query_knowledge("programming ideas", "programming")
print(f"Programming data: {len(programming_results)} outcomes discovered")
if __name__ == "__main__":
print("Beginning Superior AI Agent Tutorial with Hugging Face Fashions...")
print("🤗 Utilizing free fashions from Hugging Face Hub")
print("📱 GPU acceleration out there!" if torch.cuda.is_available() else "💻 Working on CPU")
strive:
await most important()
besides RuntimeError:
import nest_asyncio
nest_asyncio.apply()
asyncio.run(most important())
print("n✅ Tutorial accomplished! You've got discovered:")
print("• The right way to arrange Cognee with Hugging Face fashions")
print("• AI-powered response era")
print("• Multi-domain data administration")
print("• Superior reasoning and retrieval")
print("• Conversational agent with reminiscence")
print("• Free GPU-accelerated inference")We conclude the tutorial by working a complete demonstration of our AI agent in motion. We first educate it from multi-domain paperwork, then check its means to retrieve data and cause intelligently. Subsequent, we have interaction it in a pure dialog, watching it study and recall data taught by customers. Lastly, we view a abstract of its reminiscence, showcasing the way it organizes and filters data by area, all with real-time inference utilizing free Hugging Face fashions.
In conclusion, we’ve constructed a totally practical AI agent that may study from structured knowledge, recall and cause with saved data, and converse intelligently utilizing Hugging Face fashions. We configure Cognee for persistent reminiscence, display domain-specific queries, and even simulate actual conversations with the agent.
Take a look at the Full Codes right here. Be happy to test different AI Agent and Agentic AI Codes and Tutorial for varied functions. Additionally, be happy to observe us on Twitter and don’t overlook to hitch our 100k+ ML SubReddit and Subscribe to our Publication.
Asif Razzaq is the CEO of Marktechpost Media Inc.. As a visionary entrepreneur and engineer, Asif is dedicated to harnessing the potential of Synthetic Intelligence for social good. His most up-to-date endeavor is the launch of an Synthetic Intelligence Media Platform, Marktechpost, which stands out for its in-depth protection of machine studying and deep studying information that’s each technically sound and simply comprehensible by a large viewers. The platform boasts of over 2 million month-to-month views, illustrating its reputation amongst audiences.
