LQAgentPlatform/test/test_redis2.py

# ！/usr/bin/ python
# -*- coding: utf-8 -*-
'''
@Project    : lq-agent-api
@File       :test.py.py
@IDE        :PyCharm
@Author     :
@Date       :2025/7/11 12:23
'''
import os
import sys
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))


import redis
#from langchain.storage import RedisStore
from langchain_community.storage import RedisStore
import asyncio
from redis import Redis as SyncRedisClient

async def main():
    # 创建同步Redis客户端（注意不是aioredis）
    redis_client = redis.Redis.from_url(
        "redis://localhost:6379",
        decode_responses=False  # LangChain需要bytes
    )
    # redis_client = SyncRedisClient.from_url(
    #     "redis://localhost:6379",
    #     decode_responses=False  # LangChain需要bytes
    # )
    
    # 创建存储
    store = RedisStore(client=redis_client)
    # AttributeError: 'RedisStore' object has no attribute 'aset'. Did you mean: 'amset'?
    # 异步存储数据（使用aset而非set）
    await store.aset("test_key", b"test_value")
    
    # 异步获取数据（使用aget而非get）
    value = await store.aget("test_key")
    print(f"获取的值: {value}")  # 输出: b'test_value'
    
    # 异步删除数据（使用adelete而非delete）
    await store.adelete("test_key")

    # 关闭连接（可选）
    redis_client.close()

asyncio.run(main())