Una guía de codificación para construir un agente inteligente de IA conversacional con memoria de agente usando modelos de Coboging y Free Achingging Face

!pip install cognee transformers torch sentence-transformers accelerate


import asyncio
import os
import json
from typing import List, Dict, Any
from datetime import datetime
from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
import torch


import cognee

async def setup_cognee():
   """Setup Cognee with proper configuration"""
   try:
       await cognee.config.set("EMBEDDING_MODEL", "sentence-transformers/all-MiniLM-L6-v2")
       await cognee.config.set("EMBEDDING_PROVIDER", "sentence_transformers")
       print("✅ Cognee configured successfully")
       return True
   except Exception as e:
       print(f"⚠️ Cognee config error: {e}")
       try:
           os.environ["EMBEDDING_MODEL"] = "sentence-transformers/all-MiniLM-L6-v2"
           os.environ["EMBEDDING_PROVIDER"] = "sentence_transformers"
           print("✅ Cognee configured via environment")
           return True
       except Exception as e2:
           print(f"⚠️ Alternative config failed: {e2}")
           return False

class HuggingFaceLLM:
   def __init__(self, model_name="microsoft/DialoGPT-medium"):
       print(f"🤖 Loading Hugging Face model: {model_name}")
       self.device = "cuda" if torch.cuda.is_available() else "cpu"
       print(f"📱 Using device: {self.device}")
      
       if "DialoGPT" in model_name:
           self.tokenizer = AutoTokenizer.from_pretrained(model_name, padding_side="left")
           self.model = 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",
               model="distilgpt2",
               device=0 if self.device == "cuda" else -1,
               max_length=150,
               do_sample=True,
               temperature=0.7
           )
           self.tokenizer = None
           self.model = None
      
       print("✅ Model loaded successfully!")
  
   def generate_response(self, prompt: str, max_length: int = 100) -> str:
       try:
           if self.model is not None:
               inputs = self.tokenizer.encode(prompt + self.tokenizer.eos_token, return_tensors="pt")
              
               with torch.no_grad():
                   outputs = self.model.generate(
                       inputs,
                       max_length=inputs.shape[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 understand."
          
           else:
               result = self.generator(prompt, max_length=max_length, truncation=True)
               return result[0]['generated_text'][len(prompt):].strip()
              
       except Exception as e:
           print(f"⚠️ Generation error: {e}")
           return "I'm processing that information."


hf_llm = None

class AdvancedAIAgent:
   """
   Advanced AI Agent with persistent memory, learning capabilities,
   and multi-domain knowledge processing using Cognee
   """
  
   def __init__(self, agent_name: str = "CogneeAgent"):
       self.name = agent_name
       self.memory_initialized = False
       self.knowledge_domains = []
       self.conversation_history = []
       self.manual_memory = [] 
      
   async def initialize_memory(self):
       """Initialize the agent's memory system and HF model"""
       global hf_llm
       if hf_llm is None:
           hf_llm = HuggingFaceLLM("microsoft/DialoGPT-medium")
      
       setup_success = await setup_cognee()
      
       try:
           await cognee.prune() 
           print(f"✅ {self.name} memory system initialized")
           self.memory_initialized = True
       except Exception as e:
           print(f"⚠️ Memory initialization warning: {e}")
           self.memory_initialized = True
  
   async def learn_from_text(self, text: str, domain: str = "general"):
       """Add knowledge to the agent's memory with domain tagging"""
       if not self.memory_initialized:
           await self.initialize_memory()
      
       enhanced_text = f"[DOMAIN: {domain}] [TIMESTAMP: {datetime.now().isoformat()}]\n{text}"
      
       try:
           await cognee.add(enhanced_text)
           await cognee.cognify() 
           if domain not in self.knowledge_domains:
               self.knowledge_domains.append(domain)
           print(f"📚 Learned new knowledge in domain: {domain}")
           return True
       except Exception as e:
           print(f"❌ Learning error: {e}")
           try:
               await cognee.add(text)
               await cognee.cognify()
               if domain not in self.knowledge_domains:
                   self.knowledge_domains.append(domain)
               print(f"📚 Learned (simplified): {domain}")
               return True
           except Exception as e2:
               print(f"❌ Simplified learning failed: {e2}")
               if not hasattr(self, 'manual_memory'):
                   self.manual_memory = []
               self.manual_memory.append({"text": text, "domain": domain})
               if domain not in self.knowledge_domains:
                   self.knowledge_domains.append(domain)
               print(f"📚 Stored in manual memory: {domain}")
               return True
  
   async def learn_from_documents(self, documents: List[Dict[str, str]]):
       """Batch learning from multiple documents"""
       print(f"📖 Processing {len(documents)} documents...")
      
       for i, doc in enumerate(documents):
           text = doc.get("content", "")
           domain = doc.get("domain", "general")
           title = doc.get("title", f"Document_{i+1}")
          
           enhanced_content = f"Title: {title}\n{text}"
           await self.learn_from_text(enhanced_content, domain)
          
           if i % 3 == 0:
               print(f"  Processed {i+1}/{len(documents)} documents")
  
   async def query_knowledge(self, question: str, domain_filter: str = None) -> List[str]:
       """Query the agent's knowledge base with optional domain filtering"""
       try:
           if domain_filter:
               enhanced_query = f"[DOMAIN: {domain_filter}] {question}"
           else:
               enhanced_query = question
              
           search_results = await cognee.search("SIMILARITY", enhanced_query)
          
           results = []
           for result in search_results:
               if hasattr(result, 'text'):
                   results.append(result.text)
               elif hasattr(result, 'content'):
                   results.append(result.content)
               elif hasattr(result, 'value'):
                   results.append(str(result.value))
               elif isinstance(result, dict):
                   content = result.get('text') or result.get('content') or result.get('data') or result.get('value')
                   if content:
                       results.append(str(content))
                   else:
                       results.append(str(result))
               elif isinstance(result, str):
                   results.append(result)
               else:
                   result_str = str(result)
                   if len(result_str) > 10: 
                       results.append(result_str)
          
           if not results and hasattr(self, 'manual_memory'):
               for item in self.manual_memory:
                   if domain_filter and item['domain'] != domain_filter:
                       continue
                   if any(word.lower() in item['text'].lower() for word in question.split()):
                       results.append(item['text'])
          
           return results[:5] 
          
       except Exception as e:
           print(f"🔍 Search error: {e}")
           results = []
           if hasattr(self, 'manual_memory'):
               for item in self.manual_memory:
                   if domain_filter and item['domain'] != domain_filter:
                       continue
                   if any(word.lower() in item['text'].lower() for word in question.split()):
                       results.append(item['text'])
           return results[:5]
  
   async def reasoning_chain(self, question: str) -> Dict[str, Any]:
       """Advanced reasoning using retrieved knowledge"""
       print(f"🤔 Processing question: {question}")
      
       relevant_info = await self.query_knowledge(question)
      
       analysis = {
           "question": question,
           "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(question, relevant_info)
           analysis["reasoning"] = reasoning
           analysis["answer"] = self._extract_key_points(relevant_info)
       else:
           analysis["reasoning"] = "No relevant knowledge found in memory"
           analysis["answer"] = "I don't have information about this topic in my current knowledge base."
      
       return analysis




   def _synthesize_answer(self, question: str, knowledge_pieces: List[str]) -> str:
       """AI-powered answer synthesis using Hugging Face model"""
       global hf_llm
      
       if not knowledge_pieces:
           return "No relevant information found in my knowledge base."
      
       context = " ".join(knowledge_pieces[:2]) 
       context = context[:300] 
      
       prompt = f"Based on this information: {context}\n\nQuestion: {question}\nAnswer:"
      
       try:
           if hf_llm:
               synthesized = hf_llm.generate_response(prompt, max_length=80)
               return synthesized if synthesized else f"Based on my knowledge: {context[:100]}..."
           else:
               return f"From my analysis: {context[:150]}..."
       except Exception as e:
           print(f"⚠️ Synthesis error: {e}")
           return f"Based on my knowledge: {context[:100]}..."
  
   def _extract_key_points(self, knowledge_pieces: List[str]) -> List[str]:
       """Extract key points from retrieved knowledge"""
       key_points = []
       for piece in knowledge_pieces:
           clean_piece = piece.replace("[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:
       """Main conversational interface with HF model integration"""
       global hf_llm
       self.conversation_history.append({"role": "user", "content": user_input})
      
       if any(word in user_input.lower() for word in ["learn", "remember", "add", "teach"]):
           content_to_learn = user_input.replace("learn this:", "").replace("remember:", "").strip()
           await self.learn_from_text(content_to_learn, "conversation")
           response = "I've stored that information in my memory! What else would you like to teach me?"
          
       elif user_input.lower().startswith(("what", "how", "why", "when", "where", "who", "tell me")):
           analysis = await self.reasoning_chain(user_input)
          
           if analysis["relevant_knowledge"] and hf_llm:
               context = " ".join(analysis["relevant_knowledge"][:2])[:200]
               prompt = f"Question: {user_input}\nKnowledge: {context}\nFriendly response:"
               ai_response = hf_llm.generate_response(prompt, max_length=60)
               response = ai_response if ai_response else "Here's what I found in my knowledge base."
           else:
               response = "I don't have specific information about that topic in my current knowledge base."
              
       else:
           relevant_context = await self.query_knowledge(user_input)
          
           if hf_llm:
               context_info = ""
               if relevant_context:
                   context_info = f" I know that: {relevant_context[0][:100]}..."
              
               conversation_prompt = f"User says: {user_input}{context_info}\nI respond:"
               response = hf_llm.generate_response(conversation_prompt, max_length=50)
              
               if not response or len(response.strip()) < 3:
                   response = "That's interesting! I'm learning from our conversation."
           else:
               response = "I'm listening and learning from our conversation."
      
       self.conversation_history.append({"role": "assistant", "content": response})
       return response

async def main():
   print("🚀 Advanced AI Agent with Cognee Tutorial")
   print("=" * 50)
  
   agent = AdvancedAIAgent("TutorialAgent")
   await agent.initialize_memory()
  
   print("\n📚 DEMO 1: Multi-domain Learning")
   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: Knowledge 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 question in test_questions:
       print(f"\n❓ Question: {question}")
       analysis = await agent.reasoning_chain(question)
       print(f"💡 Answer: {analysis.get('answer', 'No answer generated')}")
       print(f"🎯 Confidence: {analysis.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🗣️ User: {user_input}")
       response = await agent.conversational_agent(user_input)
       print(f"🤖 Agent: {response}")
  
   print(f"\n📊 DEMO 4: Agent Knowledge Summary")
   print(f"Knowledge domains: {agent.knowledge_domains}")
   print(f"Conversation history: {len(agent.conversation_history)} exchanges")
  
   print(f"\n🎯 Domain-specific search:")
   programming_results = await agent.query_knowledge("programming concepts", "programming")
   print(f"Programming knowledge: {len(programming_results)} results found")


if __name__ == "__main__":
   print("Starting Advanced AI Agent Tutorial with Hugging Face Models...")
   print("🤗 Using free models from Hugging Face Hub")
   print("📱 GPU acceleration available!" if torch.cuda.is_available() else "💻 Running on CPU")
  
   try:
       await main()
   except RuntimeError:
       import nest_asyncio
       nest_asyncio.apply()
       asyncio.run(main())
  
   print("\n✅ Tutorial completed! You've learned:")
   print("• How to set up Cognee with Hugging Face models")
   print("• AI-powered response generation")
   print("• Multi-domain knowledge management")
   print("• Advanced reasoning and retrieval")
   print("• Conversational agent with memory")
   print("• Free GPU-accelerated inference")