跳到正文
DEV Community· Mohammed Rafay·· 2 小時前AI 評分53

如何修復 OpenAI Agents SDK RedisSession 在 Redis Cluster 上的 ClusterPipeline 缺少 raise_first_error 錯誤

Fix "'ClusterPipeline' object has no attribute 'raise_first_error'" with OpenAI Agents SDK RedisSession on Redis Cluster

AI 導讀

OpenAI Agents SDK 0.24.0 的 RedisSession 在 Redis Cluster 寫入仍會失敗，錯誤改為 TypeError。

正文

Disclosure: I'm Mohammed, founder of MemorySync. I drafted this with AI assistance from the public GitHub issue and the SDK source. Code status: on Oct 25, 2026 I installed openai-agents 0.24.0 and redis 8.1.0 (Python 3.13). I started a local 3-node Redis Cluster (Redis 8.0.2), confirmed that ClusterPipeline lacks the hooked attributes, reproduced the new TypeError from add_items and clear_session with a RedisCluster client, and ran the custom session below against that cluster, including two Runner.run turns with a stand-in model (no OpenAI key, so no real LLM call). I checked the hash-slot math with redis-py's key_slot.

The error

With the OpenAI Agents SDK's Redis-backed session and a cluster client (redis.asyncio.cluster.RedisCluster), storing history fails:

File "agents/extensions/memory/redis_session.py", line 483, in _write_items_attempt
    raise_first_error = pipe.raise_first_error
AttributeError: 'ClusterPipeline' object has no attribute 'raise_first_error'

This is openai/openai-agents-python#5322, still open as of Oct 25, 2026. The reporter saw it from 0.20.0 through 0.23.1. Per the issue, add_items and (on latest) clear_session fail, get_items and pop_item work, and 0.14.5 worked for all four.

On 0.24.0 you'll get a clearer error instead. I reproduced this against a local Redis Cluster:

TypeError: RedisSession does not support transactions with RedisCluster clients. Use redis.asyncio.Redis connected to a standalone Redis server, or implement a custom Session backend for Redis Cluster.

In 0.24.0 the check sits in the write path, not in __init__. Constructing the session succeeds, and the first write raises. That comes from PR #5361 ("scope Redis Cluster rejection to transactional writes"), which the issue links and which was merged on Oct 14, 2026. On the cluster, get_items and pop_item still worked on 0.24.0, and add_items and clear_session raised the TypeError.

Why it happens

The issue identifies two causes:

  1. Pipeline internals. The write hooks pipe.raise_first_error and pipe.parse_response, which exist on the standalone redis.asyncio.client.Pipeline but not on ClusterPipeline. I checked: hasattr(ClusterPipeline, "raise_first_error") is False, and it's True for Pipeline.
  2. Cross-slot transaction. The write is a WATCH/MULTI transaction over three keys: {prefix}:{id}, {prefix}:{id}:messages and {prefix}:{id}:counter. Those keys hash to different cluster slots, and Redis Cluster only runs a transaction within one slot. So adding the missing hooks wouldn't be enough. With redis-py's key_slot, repro:s1 and repro:s1:messages landed in slots 7598 and 11344.

Fix options

A. Point RedisSession at a standalone (non-cluster) Redis. This is what the SDK's error message suggests. It's simplest if your provider offers a non-cluster mode.

B. Pin openai-agents==0.14.5. The issue reports that it worked on a cluster, and on my local cluster (with redis 8.1.0) add, get, pop and clear all ran. You'd give up every later fix, though. Use this only as a stopgap.

C. Write a small cluster-safe Session. The SDK's sessions are a four-method interface (get_items, add_items, pop_item, clear_session). Use one list key with a Redis Cluster hash tag ({...}), so everything for a session lives in one slot and needs no transaction:

import json
from agents.memory import SessionABC


class ClusterRedisSession(SessionABC):
    """Session history on Redis Cluster: one list key, hash-tagged by session id."""

    def __init__(self, session_id: str, redis_client, key_prefix: str = "agents"):
        self.session_id = session_id
        self._redis = redis_client
        # {...} is a hash tag: every key for this session maps to one slot.
        self._key = f"{key_prefix}:{{{session_id}}}:messages"

    async def get_items(self, limit: int | None = None):
        start = -limit if limit else 0
        raw = await self._redis.lrange(self._key, start, -1)
        return [json.loads(r) for r in raw]

    async def add_items(self, items):
        if items:
            await self._redis.rpush(self._key, *[json.dumps(i) for i in items])

    async def pop_item(self):
        raw = await self._redis.rpop(self._key)
        return json.loads(raw) if raw else None

    async def clear_session(self):
        await self._redis.delete(self._key)

Use it like any other session:

from redis.asyncio.cluster import RedisCluster
from agents import Agent, Runner

client = RedisCluster(host="your-cluster-host", port=6379)
session = ClusterRedisSession("user-123:chat-1", client, key_prefix="agents")
result = await Runner.run(agent, "Hi again", session=session)

What I verified on the local 3-node cluster: add/get (including limit=1 returning the newest item), pop and clear all behaved as expected. Two Runner.run turns with a stand-in model stored both user and assistant items, and the second turn's model input included the first turn's history. repro:{s1}:messages and repro:{s1}:counter both hash to slot 15224. I didn't make a real LLM call or test with concurrent writers.

Trade-offs: RPUSH is atomic for a single call, but this skips the SDK's WATCH-based protections and its session_settings handling. Concurrent writers to the same session interleave in arrival order. The SDK docstring says third-party libraries should implement the Session protocol rather than subclass SessionABC. Subclassing is fine in your own app code. SessionABC declares session_settings with a default of None, so this class doesn't need to set it.

A note on what a session is (and isn't)

A session replays one conversation's history. It doesn't give a user memory across different sessions. That's a separate layer, whether you use the SDK's other memory features or a hosted store. MemorySync has an OpenAI Agents SDK guide for that cross-session layer. It doesn't replace your Redis session store.

Common questions

Will upgrading fix it? As of 0.24.0, no. You get a clear TypeError instead of the AttributeError. Watch #5322 for real cluster support. The reporter proposed hash-tagged keys plus a single-node WATCH/MULTI.

Why do get_items and pop_item still work? According to the issue, only the transactional write paths use the hooked pipeline.

Can I just hash-tag the prefix I pass to RedisSession? The key layout is {prefix}:{id}.... A tag in the prefix would put all sessions in one slot, which concentrates load on one node. It also wouldn't fix the missing pipeline attributes, because 0.24.0 rejects any RedisCluster client on write.

Managed Redis only offers cluster mode. Is option C safe for production? It's a small, readable store, but test it under your real concurrency on a staging cluster first.

Curious what cross-session memory looks like next to session history? Try MemorySync in the browser. No signup is needed.

Sources

來源：DEV Community · dev.to

#教學/實踐#Agent#部署/工程#OpenAI