Redis

Redis 支援不同 SDK 之間的上下文模式。 在 .NET 中,將 Redis 支持的搜尋連接到 RAG 的通用搜尋TextSearchProvider。 Agent Framework Redis 套件提供可搜尋的記憶與對話歷程提供者,供 Python 使用。

Pattern API SDK 行為
RAG TextSearchProvider 使用 Redis 搜尋轉接器 .NET 在呼叫前或透過隨選搜尋工具取得相關的 Redis 內容。
可搜尋記憶體 RedisContextProvider Python 透過全文或混合向量搜尋擷取對話細節並擷取相關上下文。
交談歷程記錄 RedisHistoryProvider Python 會持續存在並重新載入會話中完全相同的訊息逐字稿。

新增含有 TextSearchProvider 的 RAG

使用適用於 .NET 的與提供者無關 TextSearchProvider 模式。 透過應用程式選擇的 Redis 用戶端或向量儲存連接器實作其搜尋轉接器,將 Redis 結果映射至 TextSearchProvider.TextSearchResult,並透過 AIContextProviders連接提供者。

此方法支援 Redis 支援的 RAG,無需 Redis 專屬的代理框架上下文提供者套件。

安裝套件

pip install agent-framework-redis --pre

新增可搜尋記憶體

當客服人員應該回憶選定的相關資訊,而非重複播放所有先前訊息時,請使用此模式。

Prerequisites

  • 一個支援 RediSearch 的 Redis 部署,例如 Redis Stack 或相容的託管服務。
  • Microsoft Foundry 專案及範例代理程式的模型部署。
  • 啟用混合向量搜尋時使用的嵌入提供者

配置可搜尋記憶體

使用 application_idagent_iduser_id 來分割記憶體。 想要混合擷取時,可以加裝 Redis 向量器和向量場設定。

# Create a provider with partition scope and OpenAI embeddings

# Please set OPENAI_API_KEY to use the OpenAI vectorizer.
# For chat responses, also set FOUNDRY_PROJECT_ENDPOINT and FOUNDRY_MODEL.

# We attach an embedding vectorizer so the provider can perform hybrid (text + vector)
# retrieval. If you prefer text-only retrieval, instantiate RedisContextProvider without the
# 'vectorizer' and vector_* parameters.
vectorizer = OpenAITextVectorizer(
    model="text-embedding-ada-002",
    api_config={"api_key": os.getenv("OPENAI_API_KEY")},
    cache=EmbeddingsCache(name="openai_embeddings_cache", redis_url=REDIS_URL),
)
# The provider manages persistence and retrieval. application_id/agent_id/user_id
# scope data for multi-tenant separation; thread_id (set later) narrows to a
# specific conversation.
provider = RedisContextProvider(
    source_id="redis_context",
    redis_url=REDIS_URL,
    index_name="redis_basics",
    application_id="matrix_of_kermits",
    agent_id="agent_kermit",
    user_id="kermit",
    redis_vectorizer=vectorizer,
    vector_field_name="vector",
    vector_algorithm="hnsw",
    vector_distance_metric="cosine",
)

將記憶體附加於代理

將提供者加入context_providers。 服務提供者會在執行後儲存對話細節,並在後續執行前呈現相關上下文。

# Create chat client for the agent
client = create_chat_client()
# Create agent wired to the Redis context provider. The provider automatically
# persists conversational details and surfaces relevant context on each turn.
agent = Agent(
    client=client,
    name="MemoryEnhancedAssistant",
    instructions=(
        "You are a helpful assistant. Personalize replies using provided context. "
        "Before answering, always check for stored context"
    ),
    tools=[],
    context_providers=[provider],
)

# Teach a user preference; the agent writes this to the provider's memory
query = "Remember that I enjoy glugenflorgle"
result = await agent.run(query)
print("User: ", query)
print("Agent: ", result)

# Ask the agent to recall the stored preference; it should retrieve from memory
query = "What do I enjoy?"
result = await agent.run(query)

保留對話記錄

當會話在應用程式重新啟動後或在其他實例上必須恢復完整文字記錄時,請使用此模式。

Prerequisites

  • 可透過 REDIS_URL 存取的 Redis 部署。
  • TLS 和經過認證的 Redis 使用者用於生產部署。

RedisHistoryProvider 連接至 context_providers。 提供者會儲存會話中的訊息,並可限制保留訊息數量。

async def example_manual_memory_store() -> None:
    """Basic example of using Redis history provider."""
    print("=== Basic Redis History Provider Example ===")

    # Create Redis history provider
    redis_provider = RedisHistoryProvider(
        source_id="redis_basic_chat",
        redis_url=REDIS_URL,
    )

    # Create agent with Redis history provider
    agent = Agent(
        client=OpenAIChatClient(),
        name="RedisBot",
        instructions="You are a helpful assistant that remembers our conversation using Redis.",
        context_providers=[redis_provider],
    )

    # Create session
    session = agent.create_session()

    # Have a conversation
    print("\n--- Starting conversation ---")
    query1 = "Hello! My name is Alice and I love pizza."
    print(f"User: {query1}")
    response1 = await agent.run(query1, session=session)
    print(f"Agent: {response1.text}")

    query2 = "What do you remember about me?"
    print(f"User: {query2}")
    response2 = await agent.run(query2, session=session)
    print(f"Agent: {response2.text}")

使用穩定的會話 ID,並在用戶端在程序重啟後必須恢復相同邏輯對話時,將序列化 AgentSession 的 ID 持續保存在受信任的應用程式儲存中。

備註

Redis 與 Agent Framework Go 的內容提供者整合目前尚未有相關文件說明。 最新狀態請參閱 Agent Framework Go 倉庫

生產考量

  • 從已認證的應用程式身份(而非模型輸出)推導租戶、搜尋、記憶體及會話範圍。
  • 使用 TLS、Redis 認證和網路隔離。
  • 在需要進行租戶隔離時,請使用獨立的金鑰前綴或部署。
  • 設定持久化、備份、保留與驅逐,以達到所需的耐用度。
  • 將被檢索的記憶體視為不可信輸入,並減少間接提示注入。
  • 在將訊息持久化儲存或為可搜尋內容建立索引之前,請先編修敏感內容。

下一步

深入探討: