Agno记忆系统深度研究：架构、机制与未来展望

#!/usr/bin/env python3
"""
Agno 知识库向量数据库配置修复示例 - 最终版本

使用 DeepSeek 聊天模型 + 本地嵌入模型的组合。
"""

from agno.models.deepseek import DeepSeek
from agno.agent import Agent
from agno.knowledge.knowledge import Knowledge
from agno.db.sqlite import SqliteDb
from agno.vectordb.chroma import ChromaDb
import os

def create_knowledge_base():
    """创建知识库 - 使用本地嵌入模型"""
    print("=== 创建知识库 ===")
    
    # 使用本地嵌入模型（无需 API）
    # 这不需要 API key，完全本地运行
    try:
        from agno.knowledge.embedder.sentence_transformer import SentenceTransformerEmbedder
        
        embedder = SentenceTransformerEmbedder(
            id="sentence-transformers/all-MiniLM-L6-v2",  # 轻量级模型，适合中文
            dimensions=384,  # 该模型的输出维度
        )
        
        print("✅ 本地嵌入模型加载成功")
        
    except ImportError:
        print("❌ 未安装 sentence-transformers 包")
        print("请运行: pip install sentence-transformers")
        return None
    except Exception as e:
        print(f"❌ 加载嵌入模型失败: {e}")
        return None
    
    # 创建向量数据库
    vector_db = ChromaDb(
        collection="knowledge_base",
        path="tmp/chromadb",
        persistent_client=True,  # 启用持久化存储
        embedder=embedder,
    )
    
    # 创建知识库
    knowledge_base = Knowledge(
        vector_db=vector_db,
        contents_db=SqliteDb(db_file="knowledge_contents.db")
    )
    
    return knowledge_base

def create_agent():
    """创建 Agent - 使用 DeepSeek 聊天模型"""
    print("\n=== 创建 Agent ===")
    
    # 获取 DeepSeek API key
    api_key = os.getenv("DEEPSEEK_API_KEY") or os.getenv("OPENAI_API_KEY")
    if not api_key:
        print("⚠️  未找到 DEEPSEEK_API_KEY 或 OPENAI_API_KEY 环境变量")
        print("将使用 DeepSeek 模型，但可能无法访问 API")
    
    # 创建 DeepSeek 聊天模型
    model = DeepSeek(
        id="deepseek-chat",
        api_key=api_key,  # 如果设置了 API key 就使用，否则可能使用免费额度或其他方式
    )
    
    return model

def main():
    """主函数"""
    print("=" * 60)
    print("Agno 知识库向量数据库修复测试 - 最终版本")
    print("=" * 60)
    
    # 创建知识库
    knowledge_base = create_knowledge_base()
    
    if knowledge_base is None:
        print("\n❌ 知识库创建失败")
        return
    
    print("✅ 知识库初始化成功！")
    
    # 创建 DeepSeek 聊天模型
    model = create_agent()
    
    # 创建 Agent
    agent = Agent(
        model=model,
        knowledge=knowledge_base,
        instructions="你是一个知识专家，可以基于提供的知识库回答问题。",
    )
    
    # 添加知识到知识库
    print("\n=== 添加知识到知识库 ===")
    sample_texts = [
        "Python 是一种高级编程语言，由 Guido van Rossum 于 1991 年首次发布。",
        "JavaScript 是一种脚本语言，主要用于网页开发。",
        "机器学习是人工智能的一个分支，让计算机能够从数据中学习。",
    ]
    
    for i, text in enumerate(sample_texts):
        print(f"\n添加知识 {i+1}: {text[:50]}...")
        try:
            knowledge_base.add_content(text_content=text)
            print(f"✅ 知识 {i+1} 添加成功")
        except Exception as e:
            print(f"❌ 知识 {i+1} 添加失败: {e}")
    
    # 使用知识库回答问题
    print("\n=== 使用知识库回答问题 ===")
    try:
        agent.print_response("请介绍一下 Python 的历史。")
    except Exception as e:
        print(f"❌ 回答问题时出错: {e}")
    
    # 搜索知识库
    print("\n=== 搜索知识库 ===")
    try:
        search_results = knowledge_base.search("编程语言", max_results=2)
        print(f"找到 {len(search_results)} 条相关结果:")
        for i, doc in enumerate(search_results):
            print(f"{i+1}. {doc.content[:100]}...")
    except Exception as e:
        print(f"❌ 搜索时出错: {e}")

if __name__ == "__main__":
    main()