Cassandra Agent 知识库

设置

安装 cassandra 包

pip install cassandra-driver

运行 cassandra

docker run -d \
--name cassandra-db\
-p 9042:9042 \
cassandra:latest

示例

agent_with_knowledge.py

from agno.agent import Agent
from agno.knowledge.pdf_url import PDFUrlKnowledgeBase
from agno.vectordb.cassandra import Cassandra

from agno.embedder.mistral import MistralEmbedder
from agno.models.mistral import MistralChat

# (可选) 设置你的 Cassandra 数据库

cluster = Cluster()

session = cluster.connect()
session.execute(
    """
    CREATE KEYSPACE IF NOT EXISTS testkeyspace
    WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 }
    """
)

knowledge_base = PDFUrlKnowledgeBase(
    urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"],
    vector_db=Cassandra(table_name="recipes", keyspace="testkeyspace", session=session, embedder=MistralEmbedder()),
)


# knowledge_base.load(recreate=False)  # 首次运行后注释掉

agent = Agent(
    model=MistralChat(provider="mistral-large-latest", api_key=os.getenv("MISTRAL_API_KEY")),
    knowledge=knowledge_base,
    show_tool_calls=True,
)

agent.print_response(
    "What are the health benefits of Khao Niew Dam Piek Maphrao Awn?", markdown=True, show_full_reasoning=True
)

异步支持 ⚡

Cassandra 也支持异步操作，能够实现并发并提高性能。

async_cassandra.py

import asyncio

from agno.agent import Agent
from agno.embedder.mistral import MistralEmbedder
from agno.knowledge.pdf_url import PDFUrlKnowledgeBase
from agno.models.mistral import MistralChat
from agno.vectordb.cassandra import Cassandra

try:
    from cassandra.cluster import Cluster  # type: ignore
except (ImportError, ModuleNotFoundError):
    raise ImportError(
        "Could not import cassandra-driver python package.Please install it with pip install cassandra-driver."
    )

cluster = Cluster()

session = cluster.connect()
session.execute(
    """
    CREATE KEYSPACE IF NOT EXISTS testkeyspace
    WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 }
    """
)

knowledge_base = PDFUrlKnowledgeBase(
    urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"],
    vector_db=Cassandra(
        table_name="recipes",
        keyspace="testkeyspace",
        session=session,
        embedder=MistralEmbedder(),
    ),
)

agent = Agent(
    model=MistralChat(),
    knowledge=knowledge_base,
    show_tool_calls=True,
)

if __name__ == "__main__":
    # 首次运行后注释掉
    asyncio.run(knowledge_base.aload(recreate=False))

    # 创建并使用 agent
    asyncio.run(
        agent.aprint_response(
            "What are the health benefits of Khao Niew Dam Piek Maphrao Awn?",
            markdown=True,
        )
    )

在asyncio.run()中使用aload()和aprint_response()方法，以便在处理高吞吐量应用程序时进行非阻塞操作。

开发者资源

• 查看 Cookbook (同步)
• 查看 Cookbook (异步)