Repository navigation
Feat/valkey 4 storage - #5703
Open
MatthiasHowellYopp wants to merge 40 commits into
Open
MatthiasHowellYopp wants to merge 40 commits into
MatthiasHowellYopp wants to merge 40 commits into
Conversation
MatthiasHowellYopp
force-pushed
the
feat/valkey-4-storage
branch
3 times, most recently
from
May 7, 2026 20:01
fef7e5e to
975887f
Compare
MatthiasHowellYopp
force-pushed
the
feat/valkey-4-storage
branch
from
May 11, 2026 19:54
975887f to
986d128
Compare
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds Valkey-GLIDE backends and wiring: shared cache config, ValkeyCache, ValkeyStorage with vector search and sync/async bridging, UploadCache multi-backend abstraction, embedding normalization and async-safety, task/agent cache integration, unified-memory drain changes, comprehensive tests, and dependency metadata. ChangesValkey Backend Integration
Estimated code review effort🎯 4 (Complex) | ⏱️ ~75 minutes Possibly related issues
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Warning Review ran into problems🔥 ProblemsStopped waiting for pipeline failures after 30000ms. One of your pipelines takes longer than our 30000ms fetch window to run, so review may not consider pipeline-failure results for inline comments if any failures occurred after the fetch window. Increase the timeout if you want to wait longer or run a Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (5)
lib/crewai/src/crewai/memory/encoding_flow.py (1)
71-94: 💤 Low valueMutating passed-in
MemoryRecordobjects in validator may cause unexpected side effects.This validator modifies
record.embeddingin place (lines 86, 93) onMemoryRecordobjects that are passed intoItemState. If those records are referenced elsewhere (e.g., in storage layer caches or other data structures), this mutation could cause unexpected behavior.Consider creating a copy of the record with the converted embedding, or relying on
MemoryRecord's ownvalidate_embeddingvalidator to handle the conversion at construction time.♻️ Alternative: create new MemoryRecord instead of mutating
`@field_validator`("similar_records", "result_record", mode="before") `@classmethod` def ensure_embedding_is_list(cls, v: Any) -> Any: """Ensure MemoryRecord embeddings are list[float], not bytes.""" if v is None: return None if isinstance(v, list): # Process list of MemoryRecords - for record in v: - if isinstance(record, MemoryRecord) and isinstance( - record.embedding, bytes - ): - import numpy as np - - arr = np.frombuffer(record.embedding, dtype=np.float32) - record.embedding = [float(x) for x in arr] - return v + result = [] + for record in v: + if isinstance(record, MemoryRecord) and isinstance(record.embedding, bytes): + import numpy as np + arr = np.frombuffer(record.embedding, dtype=np.float32) + # Use model_copy to create new record with converted embedding + result.append(record.model_copy(update={"embedding": [float(x) for x in arr]})) + else: + result.append(record) + return result🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@lib/crewai/src/crewai/memory/encoding_flow.py` around lines 71 - 94, The validator ensure_embedding_is_list mutates MemoryRecord.embedding in-place (for fields similar_records and result_record), which can cause side effects; update it to avoid mutating inputs by creating and returning new MemoryRecord instances with converted embeddings (or convert entire list to new records) instead of assigning to record.embedding, or delegate conversion to MemoryRecord's own validator/constructor so ensure_embedding_is_list only returns new objects or primitives and leaves original MemoryRecord instances untouched; adjust handling for both list and single record paths to return copies with embeddings converted from bytes -> list[float].lib/crewai/tests/memory/storage/test_valkey_storage_scope.py (1)
7-7: 💤 Low valueUnused import:
uuid4The
uuid4import is not used anywhere in this test file.-from uuid import uuid4🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@lib/crewai/tests/memory/storage/test_valkey_storage_scope.py` at line 7, Remove the unused import uuid4 from the top of the test file; locate the import statement "from uuid import uuid4" and delete it (or if future use is intended, use it where needed in tests like in test functions or fixtures), ensuring no other code references uuid4.lib/crewai/src/crewai/memory/storage/valkey_cache.py (1)
132-160: 💤 Low value
json.dumpsmay raiseTypeErrorfor non-JSON-serializable values.If a caller passes a non-serializable value (e.g., a custom object, datetime, or bytes),
json.dumps(value)on line 148 will raise aTypeError. Consider either documenting this limitation clearly or adding error handling.🛡️ Suggested defensive handling
async def set( self, key: str, value: Any, ttl: int | None = None, ) -> None: """Set value in cache. Args: key: Cache key. - value: Value to cache (will be serialized to JSON). + value: Value to cache (must be JSON-serializable). ttl: TTL in seconds (None uses default_ttl, 0 = no expiration). + + Raises: + TypeError: If value is not JSON-serializable. """ from glide import ExpirySet, ExpiryType client = await self._get_client() - serialized = json.dumps(value) + try: + serialized = json.dumps(value) + except TypeError as e: + _logger.error(f"Failed to serialize value for key {key}: {e}") + raise🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@lib/crewai/src/crewai/memory/storage/valkey_cache.py` around lines 132 - 160, The set method in valkey_cache.py currently calls json.dumps(value) which can raise TypeError for non-JSON-serializable objects; wrap the serialization in a try/except around json.dumps in the set method (ValKeyCache.set) and handle TypeError by either (a) re-raising a clear error (e.g., ValueError/TypeError) that includes the cache key and the value's type, or (b) falling back to a safe serializer (e.g., json.dumps(value, default=str) or using pickle) based on project conventions; ensure the chosen behavior is consistent and documented in the method docstring.lib/crewai/src/crewai/a2a/utils/agent_card.py (1)
48-58: 💤 Low valueConsider thread-safety for
_cache_configuredflag.The
_cache_configuredflag is not protected by a lock, creating a potential race condition if_ensure_cache_configured()is called concurrently from multiple threads. Whilecaches.set_config()is likely idempotent (making this benign), the pattern intask.py(lines 76-78) uses a threading lock for similar initialization.For consistency and to prevent duplicate initialization logs or wasted work, consider adding a lock:
♻️ Suggested improvement
+import threading + _cache_configured = False +_cache_config_lock = threading.Lock() def _ensure_cache_configured() -> None: """Configure aiocache on first use (lazy initialization).""" global _cache_configured if _cache_configured: return - caches.set_config(get_aiocache_config()) - _cache_configured = True + with _cache_config_lock: + if _cache_configured: + return + caches.set_config(get_aiocache_config()) + _cache_configured = True🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@lib/crewai/src/crewai/a2a/utils/agent_card.py` around lines 48 - 58, The _cache_configured flag in _ensure_cache_configured() can race when called concurrently; protect it with a threading.Lock to ensure only one thread runs caches.set_config(get_aiocache_config()) and flips the flag. Add a module-level lock (e.g., _cache_configured_lock = threading.Lock()), acquire it at the start of _ensure_cache_configured(), re-check _cache_configured inside the lock (double-checked locking), call caches.set_config(...) and set _cache_configured = True while holding the lock to avoid duplicate initialization work or logs.lib/crewai/tests/utilities/test_cache_config.py (1)
89-98: ⚡ Quick winAdd explicit
REDIS_URLcoverage forget_aiocache_config().This class currently validates the no-URL and
VALKEY_URLpaths, but not theREDIS_URL-only path. Adding it would lock in backward-compatible behavior and prevent regressions.Proposed test addition
class TestGetAiocacheConfig: @@ def test_returns_redis_cache_when_url_set(self) -> None: with patch.dict( os.environ, {"VALKEY_URL": "redis://myhost:6380/2"}, clear=True ): config = get_aiocache_config() assert config["default"]["cache"] == "aiocache.RedisCache" assert config["default"]["endpoint"] == "myhost" assert config["default"]["port"] == 6380 assert config["default"]["db"] == 2 + + def test_returns_redis_cache_when_only_redis_url_set(self) -> None: + with patch.dict( + os.environ, {"REDIS_URL": "redis://redis-host:6379/0"}, clear=True + ): + config = get_aiocache_config() + assert config["default"]["cache"] == "aiocache.RedisCache" + assert config["default"]["endpoint"] == "redis-host" + assert config["default"]["port"] == 6379 + assert config["default"]["db"] == 0🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@lib/crewai/tests/utilities/test_cache_config.py` around lines 89 - 98, Add a new unit test for get_aiocache_config that covers the REDIS_URL-only path: create a test (e.g., test_returns_redis_cache_when_redis_url_set) that uses patch.dict(os.environ, {"REDIS_URL": "redis://myhost:6380/2"}, clear=True) to call get_aiocache_config(), then assert config["default"]["cache"] == "aiocache.RedisCache" and that endpoint == "myhost", port == 6380, and db == 2; place this alongside the existing test_returns_redis_cache_when_url_set in test_cache_config.py to ensure backward-compatible behavior when REDIS_URL is provided.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@lib/crewai/src/crewai/memory/storage/valkey_storage.py`:
- Around line 778-779: The update to record.last_accessed uses datetime.now(),
causing a timezone mismatch with MemoryRecord's defaults which use
datetime.utcnow(); replace datetime.now() with datetime.utcnow() in the code
that updates MemoryRecord.last_accessed (the assignment in valkey_storage.py
where record.last_accessed is set) and ensure datetime.utcnow is imported/used
consistently throughout any related methods manipulating
created_at/last_accessed to maintain UTC timestamps.
In `@lib/crewai/src/crewai/memory/unified_memory.py`:
- Around line 363-372: Update the timeout exception handling in the drain/writes
loop so it catches both the builtin TimeoutError and
concurrent.futures.TimeoutError: modify the except block in unified_memory.py
(the block that currently reads "except TimeoutError:" within the save/drain
logic) to catch a tuple including the concurrent.futures timeout class, and
ensure concurrent.futures.TimeoutError is imported or referenced (e.g., import
concurrent.futures or alias the futures TimeoutError) so Python 3.10 futures
timeouts trigger the same warning path (the _logger.warning call and
failed_saves increment).
In `@lib/crewai/src/crewai/utilities/cache_config.py`:
- Around line 14-33: The parse_cache_url function can raise ValueError when
converting parsed.path to int for non-numeric DB names; update parse_cache_url
to defensively handle parsed.path by validating or wrapping
int(parsed.path.lstrip("/")) in a try/except and fall back to 0 (or None) when
conversion fails, ensuring the returned dict (host, port, db, password) never
raises on non-numeric paths; reference parsed.path and the db key in the return
value when implementing this change.
---
Nitpick comments:
In `@lib/crewai/src/crewai/a2a/utils/agent_card.py`:
- Around line 48-58: The _cache_configured flag in _ensure_cache_configured()
can race when called concurrently; protect it with a threading.Lock to ensure
only one thread runs caches.set_config(get_aiocache_config()) and flips the
flag. Add a module-level lock (e.g., _cache_configured_lock = threading.Lock()),
acquire it at the start of _ensure_cache_configured(), re-check
_cache_configured inside the lock (double-checked locking), call
caches.set_config(...) and set _cache_configured = True while holding the lock
to avoid duplicate initialization work or logs.
In `@lib/crewai/src/crewai/memory/encoding_flow.py`:
- Around line 71-94: The validator ensure_embedding_is_list mutates
MemoryRecord.embedding in-place (for fields similar_records and result_record),
which can cause side effects; update it to avoid mutating inputs by creating and
returning new MemoryRecord instances with converted embeddings (or convert
entire list to new records) instead of assigning to record.embedding, or
delegate conversion to MemoryRecord's own validator/constructor so
ensure_embedding_is_list only returns new objects or primitives and leaves
original MemoryRecord instances untouched; adjust handling for both list and
single record paths to return copies with embeddings converted from bytes ->
list[float].
In `@lib/crewai/src/crewai/memory/storage/valkey_cache.py`:
- Around line 132-160: The set method in valkey_cache.py currently calls
json.dumps(value) which can raise TypeError for non-JSON-serializable objects;
wrap the serialization in a try/except around json.dumps in the set method
(ValKeyCache.set) and handle TypeError by either (a) re-raising a clear error
(e.g., ValueError/TypeError) that includes the cache key and the value's type,
or (b) falling back to a safe serializer (e.g., json.dumps(value, default=str)
or using pickle) based on project conventions; ensure the chosen behavior is
consistent and documented in the method docstring.
In `@lib/crewai/tests/memory/storage/test_valkey_storage_scope.py`:
- Line 7: Remove the unused import uuid4 from the top of the test file; locate
the import statement "from uuid import uuid4" and delete it (or if future use is
intended, use it where needed in tests like in test functions or fixtures),
ensuring no other code references uuid4.
In `@lib/crewai/tests/utilities/test_cache_config.py`:
- Around line 89-98: Add a new unit test for get_aiocache_config that covers the
REDIS_URL-only path: create a test (e.g.,
test_returns_redis_cache_when_redis_url_set) that uses patch.dict(os.environ,
{"REDIS_URL": "redis://myhost:6380/2"}, clear=True) to call
get_aiocache_config(), then assert config["default"]["cache"] ==
"aiocache.RedisCache" and that endpoint == "myhost", port == 6380, and db == 2;
place this alongside the existing test_returns_redis_cache_when_url_set in
test_cache_config.py to ensure backward-compatible behavior when REDIS_URL is
provided.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 79394168-72b2-49c2-ad41-99a802d771dd
⛔ Files ignored due to path filters (1)
uv.lockis excluded by!**/*.lock
📒 Files selected for processing (18)
lib/crewai-files/src/crewai_files/cache/upload_cache.pylib/crewai/pyproject.tomllib/crewai/src/crewai/a2a/utils/agent_card.pylib/crewai/src/crewai/a2a/utils/task.pylib/crewai/src/crewai/memory/encoding_flow.pylib/crewai/src/crewai/memory/storage/valkey_cache.pylib/crewai/src/crewai/memory/storage/valkey_storage.pylib/crewai/src/crewai/memory/types.pylib/crewai/src/crewai/memory/unified_memory.pylib/crewai/src/crewai/utilities/cache_config.pylib/crewai/tests/memory/storage/test_valkey_cache.pylib/crewai/tests/memory/storage/test_valkey_storage.pylib/crewai/tests/memory/storage/test_valkey_storage_errors.pylib/crewai/tests/memory/storage/test_valkey_storage_scope.pylib/crewai/tests/memory/storage/test_valkey_storage_search.pylib/crewai/tests/memory/test_embedding_safety.pylib/crewai/tests/utilities/test_cache_config.pypyproject.toml
MatthiasHowellYopp
force-pushed
the
feat/valkey-4-storage
branch
from
May 11, 2026 20:49
986d128 to
ba9bee1
Compare
There was a problem hiding this comment.
🧹 Nitpick comments (4)
lib/crewai/tests/memory/storage/test_valkey_storage_search.py (2)
916-931: 💤 Low valueTest doesn't verify that limit=0 is passed to FT.SEARCH.
The test asserts empty results but doesn't check that the KNN clause contains
[KNN 0 ...]. If the implementation sent a higher limit and filtered client-side, this test would still pass.📝 Verify KNN limit in query
query_embedding = [0.1, 0.2, 0.3, 0.4] results = await valkey_storage.asearch(query_embedding, limit=0) + # Verify KNN limit in query + call_args = mock_ft_search.call_args + query = call_args[0][2] + assert "=>[KNN 0 `@embedding` $BLOB AS score]" in query + # Verify empty results assert len(results) == 0🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@lib/crewai/tests/memory/storage/test_valkey_storage_search.py` around lines 916 - 931, Update the test_search_with_zero_limit_returns_empty test to also verify the KNN limit is passed to the Redis FT.SEARCH call: after calling valkey_storage.asearch(..., limit=0) assert that the patched crewai.memory.storage.valkey_storage.ft.search mock was called and inspect its query argument (from mock_ft_search.call_args or call_args_list) to confirm the KNN clause contains "[KNN 0" (or otherwise includes "KNN 0" with the expected embedding placeholder), ensuring ValkeyStorage.asearch sends limit=0 to FT.SEARCH rather than filtering client-side.
254-254: ⚡ Quick winEscaping assertions are too permissive.
Tests check for escaped OR unescaped versions (e.g.,
"agent\\-1" in query or "agent-1" in query), which means they pass regardless of whether escaping is actually applied. This reduces test effectiveness in catching escaping regressions.🔒 Strengthen escaping verification
- assert "@agent_id:{agent\\-1}" in query or "@agent_id:{agent-1}" in query + assert "@agent_id:{agent\\-1}" in query, "Special characters should be escaped"Apply similar changes to lines 298, 483, and 514.
Also applies to: 298-298, 483-483, 514-514
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@lib/crewai/tests/memory/storage/test_valkey_storage_search.py` at line 254, Replace the permissive OR assertions that accept both escaped and unescaped forms with strict checks that require the escaped form; specifically change occurrences like assert "@agent_id:{agent\\-1}" in query or "@agent_id:{agent-1}" in query to assert "@agent_id:{agent\\-1}" in query (i.e., require the backslash-escaped hyphen) and do the same for the other similar assertions referenced (the ones currently allowing either "@agent_id:{...\\-...}" or unescaped "@agent_id:{...-...}" at the other locations), ensuring the tests fail if escaping is not applied.lib/crewai/src/crewai/memory/storage/valkey_storage.py (2)
252-274: 💤 Low valueBackground loop lifecycle: consider adding explicit shutdown method.
The class-level background event loop runs indefinitely as a daemon thread. While this works for most cases, providing an explicit shutdown mechanism (e.g.,
ValkeyStorage.shutdown_background_loop()) would allow clean resource cleanup in long-running processes that dynamically create/destroy storage instances.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@lib/crewai/src/crewai/memory/storage/valkey_storage.py` around lines 252 - 274, Add an explicit classmethod (e.g., ValkeyStorage.shutdown_background_loop) to cleanly stop the background loop created by _get_or_create_loop: if _bg_loop exists and is running, use loop.call_soon_threadsafe(loop.stop) (and run_coroutine_threadsafe(loop.shutdown_asyncgens(), loop) if available) then join _bg_thread with a timeout, close the loop (or schedule loop.close() on the loop thread), and finally set _bg_loop and _bg_thread to None; protect the shutdown with the existing _bg_lock to avoid races.
481-490: ⚖️ Poor tradeoffConsider validating index schema when reusing existing index.
The code checks if the index exists but doesn't verify that its schema (vector dimensions, algorithm) matches the current configuration. A schema mismatch could lead to confusing runtime errors during search operations.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@lib/crewai/src/crewai/memory/storage/valkey_storage.py` around lines 481 - 490, The check that only tests for the presence of "memory_index" should be extended to validate its schema: after detecting the index via ft.list(client) and the names set, fetch the index metadata (e.g., via the index info/describe call provided by your vector store client) for "memory_index" and compare its vector dimension, metric/algorithm and any relevant field names against the current storage configuration (use the class attributes like self._vector_dim / self._metric or whatever config variables exist in ValKeyStorage). If the schema matches, keep setting self._index_created = True; if it does not match, log a clear error and either recreate the index with the correct schema or raise an exception so callers know to resolve the mismatch. Ensure you use the same client object and the existing symbols (ft.list(client), "memory_index", existing, names, self._index_created) when locating and validating the index.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@lib/crewai/src/crewai/memory/storage/valkey_storage.py`:
- Around line 252-274: Add an explicit classmethod (e.g.,
ValkeyStorage.shutdown_background_loop) to cleanly stop the background loop
created by _get_or_create_loop: if _bg_loop exists and is running, use
loop.call_soon_threadsafe(loop.stop) (and
run_coroutine_threadsafe(loop.shutdown_asyncgens(), loop) if available) then
join _bg_thread with a timeout, close the loop (or schedule loop.close() on the
loop thread), and finally set _bg_loop and _bg_thread to None; protect the
shutdown with the existing _bg_lock to avoid races.
- Around line 481-490: The check that only tests for the presence of
"memory_index" should be extended to validate its schema: after detecting the
index via ft.list(client) and the names set, fetch the index metadata (e.g., via
the index info/describe call provided by your vector store client) for
"memory_index" and compare its vector dimension, metric/algorithm and any
relevant field names against the current storage configuration (use the class
attributes like self._vector_dim / self._metric or whatever config variables
exist in ValKeyStorage). If the schema matches, keep setting self._index_created
= True; if it does not match, log a clear error and either recreate the index
with the correct schema or raise an exception so callers know to resolve the
mismatch. Ensure you use the same client object and the existing symbols
(ft.list(client), "memory_index", existing, names, self._index_created) when
locating and validating the index.
In `@lib/crewai/tests/memory/storage/test_valkey_storage_search.py`:
- Around line 916-931: Update the test_search_with_zero_limit_returns_empty test
to also verify the KNN limit is passed to the Redis FT.SEARCH call: after
calling valkey_storage.asearch(..., limit=0) assert that the patched
crewai.memory.storage.valkey_storage.ft.search mock was called and inspect its
query argument (from mock_ft_search.call_args or call_args_list) to confirm the
KNN clause contains "[KNN 0" (or otherwise includes "KNN 0" with the expected
embedding placeholder), ensuring ValkeyStorage.asearch sends limit=0 to
FT.SEARCH rather than filtering client-side.
- Line 254: Replace the permissive OR assertions that accept both escaped and
unescaped forms with strict checks that require the escaped form; specifically
change occurrences like assert "@agent_id:{agent\\-1}" in query or
"@agent_id:{agent-1}" in query to assert "@agent_id:{agent\\-1}" in query (i.e.,
require the backslash-escaped hyphen) and do the same for the other similar
assertions referenced (the ones currently allowing either
"@agent_id:{...\\-...}" or unescaped "@agent_id:{...-...}" at the other
locations), ensuring the tests fail if escaping is not applied.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 740ada73-a4ea-43e2-b0fb-2031ed8ed6c3
⛔ Files ignored due to path filters (1)
uv.lockis excluded by!**/*.lock
📒 Files selected for processing (18)
lib/crewai-files/src/crewai_files/cache/upload_cache.pylib/crewai/pyproject.tomllib/crewai/src/crewai/a2a/utils/agent_card.pylib/crewai/src/crewai/a2a/utils/task.pylib/crewai/src/crewai/memory/encoding_flow.pylib/crewai/src/crewai/memory/storage/valkey_cache.pylib/crewai/src/crewai/memory/storage/valkey_storage.pylib/crewai/src/crewai/memory/types.pylib/crewai/src/crewai/memory/unified_memory.pylib/crewai/src/crewai/utilities/cache_config.pylib/crewai/tests/memory/storage/test_valkey_cache.pylib/crewai/tests/memory/storage/test_valkey_storage.pylib/crewai/tests/memory/storage/test_valkey_storage_errors.pylib/crewai/tests/memory/storage/test_valkey_storage_scope.pylib/crewai/tests/memory/storage/test_valkey_storage_search.pylib/crewai/tests/memory/test_embedding_safety.pylib/crewai/tests/utilities/test_cache_config.pypyproject.toml
✅ Files skipped from review due to trivial changes (3)
- lib/crewai/pyproject.toml
- pyproject.toml
- lib/crewai/tests/utilities/test_cache_config.py
🚧 Files skipped from review as they are similar to previous changes (11)
- lib/crewai/tests/memory/storage/test_valkey_storage_errors.py
- lib/crewai/src/crewai/utilities/cache_config.py
- lib/crewai/tests/memory/storage/test_valkey_cache.py
- lib/crewai/src/crewai/memory/unified_memory.py
- lib/crewai/src/crewai/memory/types.py
- lib/crewai/tests/memory/test_embedding_safety.py
- lib/crewai/src/crewai/a2a/utils/agent_card.py
- lib/crewai/src/crewai/a2a/utils/task.py
- lib/crewai/src/crewai/memory/storage/valkey_cache.py
- lib/crewai/tests/memory/storage/test_valkey_storage_scope.py
- lib/crewai-files/src/crewai_files/cache/upload_cache.py
MatthiasHowellYopp
force-pushed
the
feat/valkey-4-storage
branch
from
May 13, 2026 14:25
ba9bee1 to
9430075
Compare
There was a problem hiding this comment.
Actionable comments posted: 7
🧹 Nitpick comments (2)
lib/crewai/src/crewai/a2a/utils/task.py (2)
93-99: ⚡ Quick winConsider clarifying
_cache_initializedsemantics.The
_cache_initializedflag is set toTrue(line 113) for both Valkey and aiocache paths, but for the aiocache path, the function does nothing (returns early at line 95) since aiocache is already configured at module level (line 84). This makes the flag's meaning ambiguous: does it mean "cache backend is ready" or "this function has been called"?♻️ Suggested refactor to clarify intent
Option 1: Rename the flag to reflect its actual purpose:
-_cache_initialized = False +_lazy_init_complete = FalseOption 2: Skip setting the flag for the aiocache path and add a comment:
with _cache_init_lock: if _cache_initialized: return if use_valkey_cache(): from crewai.memory.storage.valkey_cache import ValkeyCache conn = parse_cache_url() or {} _task_cache = ValkeyCache( host=conn.get("host", "localhost"), port=conn.get("port", 6379), db=conn.get("db", 0), password=conn.get("password"), default_ttl=3600, ) - - _cache_initialized = True + _cache_initialized = True + # For aiocache, nothing to do here (already configured at module level)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@lib/crewai/src/crewai/a2a/utils/task.py` around lines 93 - 99, The _cache_initialized flag is ambiguous because it's set for both the Valkey initialization and the aiocache path even though the aiocache backend is configured at module import, so change the semantics: either rename _cache_initialized to _init_called (and update all checks that use _cache_initialized and _cache_init_lock in init_task_cache/_init_task_cache to reflect that it only means "this init function executed") OR only set _cache_initialized when this function actually performs initialization for Valkey (leave it false for the aiocache branch) and add a clarifying comment on the aiocache branch stating aiocache is configured at module level; update the guard uses of _cache_initialized and _cache_init_lock accordingly so other code (e.g., callers of init_task_cache, the Valkey init path) get the correct meaning.
156-156: 💤 Low valueConsider making polling interval configurable.
The cancellation polling interval is hardcoded to 0.1 seconds in both Valkey and aiocache paths. For high-frequency task environments or slower networks, a configurable interval (via environment variable or module constant) might be beneficial.
♻️ Example configurable interval
+# Cancellation polling interval in seconds +_CANCEL_POLL_INTERVAL = float(os.getenv("CREWAI_CANCEL_POLL_INTERVAL", "0.1")) + async def poll_for_cancel_valkey() -> bool: """Poll ValkeyCache for cancellation flag.""" while True: if _task_cache is not None and await _task_cache.get( f"cancel:{task_id}" ): return True - await asyncio.sleep(0.1) + await asyncio.sleep(_CANCEL_POLL_INTERVAL) async def poll_for_cancel_aiocache() -> bool: """Poll aiocache for cancellation flag.""" cache = caches.get("default") while True: if await cache.get(f"cancel:{task_id}"): return True - await asyncio.sleep(0.1) + await asyncio.sleep(_CANCEL_POLL_INTERVAL)Also applies to: 164-164
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@lib/crewai/src/crewai/a2a/utils/task.py` at line 156, Replace the hardcoded asyncio.sleep(0.1) used in both the Valkey and aiocache cancellation loops with a configurable poll interval: add a module-level constant like CANCELLATION_POLL_INTERVAL (default 0.1) or read from an env var (e.g., os.getenv("CANCELLATION_POLL_INTERVAL")) and use that value (or accept it as an optional parameter) in the functions handling the Valkey and aiocache polling paths (referencing the Valkey cancellation loop and the aiocache cancellation loop in task.py); update both occurrences (the sleep at line with Valkey and the sleep at the aiocache path) to use the new constant/parameter and validate/convert the env value to float.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@lib/crewai/src/crewai/a2a/utils/task.py`:
- Around line 87-113: _wrap the ValkeyCache creation inside _ensure_task_cache
in a try/except to handle connection errors: when use_valkey_cache() is true and
before assigning to _task_cache, catch exceptions from ValkeyCache(...) (and
from parse_cache_url()) and log the error with context, then set _task_cache to
None (or a safe in-memory fallback) and still set _cache_initialized = True so
subsequent calls don’t crash; reference the symbols _ensure_task_cache,
ValkeyCache, parse_cache_url, use_valkey_cache, _task_cache, _cache_initialized,
and _cache_init_lock when making the change so the init is thread-safe and fails
gracefully.
In `@lib/crewai/src/crewai/memory/storage/valkey_storage.py`:
- Around line 513-519: The metadata_filter clauses built in _vector_search() are
not covered by the FT index because memory_index's schema only defines
VectorField("embedding"), TagField("scope"), TagField("categories"),
NumericField("created_at"), and NumericField("importance"), so metadata_filter
cannot actually restrict FT.SEARCH results; fix by either (A) adding the
necessary metadata fields to the schema (materialize each metadata key used in
metadata_filter as a TagField or NumericField depending on type) and updating
memory_index/schema construction where schema is defined, or (B) if you cannot
change the FT index, remove metadata clauses from the FT query in
_vector_search() and instead apply metadata_filter as a post-filter over the
FT.SEARCH results (filtering returned documents by metadata values before
returning final hits). Ensure the change addresses both the _vector_search()
flow and the other occurrences noted (around lines referenced ~1237-1245) so
metadata filters are actually applied.
- Around line 276-307: Add explicit synchronous and asynchronous close methods
on ValkeyStorage (e.g., close() and aclose()) that encapsulate the existing
cleanup logic currently in __aexit__ and __del__: aclose() should await
self._client.close() and set self._client = None; close() should synchronously
close the client (handling coroutine returns with asyncio.run when needed) and
set self._client = None. Update __aexit__ to call await self.aclose() and update
__del__ to call self.close() (or schedule it on _bg_loop as currently done) so
Memory.close() can deterministically release the Glide/Valkey client by calling
the new close()/aclose() API on ValkeyStorage.
- Around line 1220-1228: The current scope filter builds a Redis tag query like
f"@scope:{{{escaped_scope}*}}" which matches siblings (e.g., "/crew/a" matches
"/crew/ab"); update the logic that uses scope_prefix/escaped_scope/query_parts
so results are boundary-aware by either (A) switching to an
ancestor/explicit-tag representation for scopes, or (B) implementing overfetch +
post-filter: keep the existing relaxed query to fetch candidates, then in the
same method filter the returned items by verifying the stored scope string
strictly matches the requested subtree (e.g., for non-root ensure stored_scope
== scope_prefix or stored_scope startswith scope_prefix + "/" so "/crew/a" does
not match "/crew/ab"); modify the code paths that consume escaped_scope and
query_parts and ensure the final returned result list is the post-filtered
subset.
- Around line 975-983: Replace the raw startswith check with a boundary-safe
comparison: instead of if scope_path.startswith(scope_prefix), match either an
exact scope (scope_path == scope_prefix) or a child scope
(scope_path.startswith(scope_prefix + "/")), treating the root scope specially
(allow prefix "/" to match all). Update the check around scope_path/scope_prefix
before building scope_key and calling client.zrange/members_result so sibling
scopes (e.g., "/crew/a" vs "/crew/ab") no longer bleed into
delete/list_records/count/reset/get_scope_info.
In `@lib/crewai/src/crewai/memory/unified_memory.py`:
- Around line 219-229: parse_cache_url currently returns host/port/db/password
but drops the URL scheme so unified_memory.py always instantiates ValkeyStorage
without TLS; update parse_cache_url to detect the scheme (e.g., set use_tls =
scheme.startswith("rediss")) and return that flag, then modify the ValkeyStorage
construction in UnifiedMemory (the block that calls parse_cache_url and
instantiates ValkeyStorage) to pass use_tls=conn.get("use_tls", False) (or the
returned flag) so ValkeyStorage receives the correct TLS setting.
---
Nitpick comments:
In `@lib/crewai/src/crewai/a2a/utils/task.py`:
- Around line 93-99: The _cache_initialized flag is ambiguous because it's set
for both the Valkey initialization and the aiocache path even though the
aiocache backend is configured at module import, so change the semantics: either
rename _cache_initialized to _init_called (and update all checks that use
_cache_initialized and _cache_init_lock in init_task_cache/_init_task_cache to
reflect that it only means "this init function executed") OR only set
_cache_initialized when this function actually performs initialization for
Valkey (leave it false for the aiocache branch) and add a clarifying comment on
the aiocache branch stating aiocache is configured at module level; update the
guard uses of _cache_initialized and _cache_init_lock accordingly so other code
(e.g., callers of init_task_cache, the Valkey init path) get the correct
meaning.
- Line 156: Replace the hardcoded asyncio.sleep(0.1) used in both the Valkey and
aiocache cancellation loops with a configurable poll interval: add a
module-level constant like CANCELLATION_POLL_INTERVAL (default 0.1) or read from
an env var (e.g., os.getenv("CANCELLATION_POLL_INTERVAL")) and use that value
(or accept it as an optional parameter) in the functions handling the Valkey and
aiocache polling paths (referencing the Valkey cancellation loop and the
aiocache cancellation loop in task.py); update both occurrences (the sleep at
line with Valkey and the sleep at the aiocache path) to use the new
constant/parameter and validate/convert the env value to float.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 30afcf2c-fac3-468d-9c15-93b2802269e5
⛔ Files ignored due to path filters (1)
uv.lockis excluded by!**/*.lock
📒 Files selected for processing (18)
lib/crewai-files/src/crewai_files/cache/upload_cache.pylib/crewai/pyproject.tomllib/crewai/src/crewai/a2a/utils/agent_card.pylib/crewai/src/crewai/a2a/utils/task.pylib/crewai/src/crewai/memory/encoding_flow.pylib/crewai/src/crewai/memory/storage/valkey_cache.pylib/crewai/src/crewai/memory/storage/valkey_storage.pylib/crewai/src/crewai/memory/types.pylib/crewai/src/crewai/memory/unified_memory.pylib/crewai/src/crewai/utilities/cache_config.pylib/crewai/tests/memory/storage/test_valkey_cache.pylib/crewai/tests/memory/storage/test_valkey_storage.pylib/crewai/tests/memory/storage/test_valkey_storage_errors.pylib/crewai/tests/memory/storage/test_valkey_storage_scope.pylib/crewai/tests/memory/storage/test_valkey_storage_search.pylib/crewai/tests/memory/test_embedding_safety.pylib/crewai/tests/utilities/test_cache_config.pypyproject.toml
✅ Files skipped from review due to trivial changes (1)
- lib/crewai/pyproject.toml
🚧 Files skipped from review as they are similar to previous changes (12)
- lib/crewai/src/crewai/a2a/utils/agent_card.py
- pyproject.toml
- lib/crewai/tests/memory/storage/test_valkey_cache.py
- lib/crewai/tests/memory/test_embedding_safety.py
- lib/crewai/src/crewai/utilities/cache_config.py
- lib/crewai/src/crewai/memory/types.py
- lib/crewai/src/crewai/memory/storage/valkey_cache.py
- lib/crewai-files/src/crewai_files/cache/upload_cache.py
- lib/crewai/tests/memory/storage/test_valkey_storage_errors.py
- lib/crewai/tests/memory/storage/test_valkey_storage_scope.py
- lib/crewai/tests/utilities/test_cache_config.py
- lib/crewai/tests/memory/storage/test_valkey_storage_search.py
MatthiasHowellYopp
force-pushed
the
feat/valkey-4-storage
branch
2 times, most recently
from
May 13, 2026 15:00
2e5f9bb to
c5a9a8d
Compare
This was referenced May 13, 2026
This was referenced May 13, 2026
MatthiasHowellYopp
force-pushed
the
feat/valkey-4-storage
branch
2 times, most recently
from
May 19, 2026 18:35
b83ca28 to
aa9ff4e
Compare
Contributor
Author
|
@greysonlalonde hoping I can get a review on this. |
MatthiasHowellYopp
force-pushed
the
feat/valkey-4-storage
branch
7 times, most recently
from
May 26, 2026 14:01
8dfe861 to
62bebab
Compare
MatthiasHowellYopp
force-pushed
the
feat/valkey-4-storage
branch
from
May 27, 2026 14:07
62bebab to
a53fc6b
Compare
Add None/empty input handling to RecallMemoryTool and RememberTool. Filter empty strings, convert string inputs to lists, and return descriptive error messages. Update tool descriptions in en.json to include explicit parameter examples. Part 2/4 of Valkey storage implementation.
Drop min_length=1 and widen queries/contents to list[str] | str | None on RecallMemorySchema/RememberSchema. min_length=1 rejected empty lists at schema validation, raising a Pydantic error before _run could return its guidance string — so ToolUsage retried the same bad payload. The schema now mirrors what _run accepts (empty lists, bare strings), letting _run own the validation and return a useful message. Adds schema-level tests. Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
Add bytes→float validators on MemoryRecord and ItemState to handle Valkey returning embeddings as raw bytes. Make embed_texts() safe when called from an async context by using a thread pool. Improve drain_writes() with per-save timeouts and error logging instead of raising on failure. Part 3/4 of Valkey storage implementation.
…embed - unified_memory: _background_encode_batch caught every RuntimeError and returned [], silently dropping embedder/LLM init and other encode failures. Now only the executor-shutdown RuntimeError is swallowed; any other emits MemorySaveFailedEvent and propagates so memories don't disappear silently. - unified_memory: restore Memory default llm to gpt-5.4-mini (an unintended revert to gpt-4o-mini diverged from the rest of CrewAI); fix the matching model name in the init-failure error message. - types.embed_texts: detect the running loop before invoking the embedder so an embedder RuntimeError can no longer be misread as "no running loop" and retried on the event-loop thread (double-invocation). - Add regression test for generic background RuntimeError. Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
…h remote on clear - _background_encode_batch no longer emits MemorySaveFailedEvent itself; it re-raises so the future carries the error and _on_save_done emits the event exactly once. Previously both paths fired, double-reporting each failure. - drain_writes now cancels and untracks a save that exceeds timeout_per_save. It was left in _pending_saves, so every later recall() waited another full timeout and close()'s shutdown(wait=True) could block indefinitely. - UploadCache.aclear now flushes the whole backend namespace via a new backend clear() (ValkeyCache.clear scans+deletes by prefix; AiocacheBackend delegates to cache.clear). It previously deleted only locally-tracked keys, leaving remote entries from other processes/runs behind. Tests updated for single-emit; added drain-timeout untrack test. Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
embed_texts returned empty vectors on its 30s timeout, so EncodingFlow treated timed-out items as new and persisted them with a missing/zero embedding - unsearchable memories that pollute the index with no signal that encoding failed. Add raise_on_timeout (default False to preserve recall's tolerate-a-miss behavior); the save-path callers in EncodingFlow (batch_embed and the update path) pass raise_on_timeout=True so a timed-out batch raises TimeoutError and fails the save instead. Adds regression tests for both timeout modes. Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
raise_on_timeout only fired in the async-context branch (thread pool + .result(timeout=30)). EncodingFlow's batch_embed/execute_plans run via asyncio.to_thread, so embed_texts is invoked with no running loop and took the direct embedder(...) branch with no timeout — the fail-loud save protection never triggered on exactly the paths that pass raise_on_timeout=True. Route the no-loop branch through the pool with the 30s timeout when raise_on_timeout is set. Adds a no-event-loop timeout regression test. Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
Remove an unused '# noqa' directive flagged by RUF100. No behavior change. Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
close() drained writes, closed storage, then shut the save pool down. But drain_writes() abandons a still-running save on timeout (cancel() is a no-op once the future is executing), so that save could keep issuing HSET/ZADD after _storage.close() closed the Valkey client, leaving a half-written batch. Shut the pool down (wait=True) before closing storage so no save is in flight when the client closes. Adds an ordering regression test. Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
CI runs 'ruff format --check'; collapse the wrapped lines ruff prefers. No behavior change. Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
AiocacheBackend.clear() called aiocache clear() with no namespace; the Redis backend treats that as FLUSHDB and wipes the whole database, so UploadCache reset could delete unrelated keys (A2A cancel flags, other app data) in a shared instance. Pass the cache namespace so only this cache's keys are removed, falling back to the default clear when no namespace is set. Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
Extract duplicated Redis URL parsing into a shared cache_config utility. Introduce ValkeyCache as a lightweight async key/value cache using valkey-glide. Wire it into A2A task handling, agent card caching, and file upload caching. Part 1/4 of Valkey storage implementation. fix: async-safe embeddings and resilient drain_writes Add bytes→float validators on MemoryRecord and ItemState to handle Valkey returning embeddings as raw bytes. Make embed_texts() safe when called from an async context by using a thread pool. Improve drain_writes() with per-save timeouts and error logging instead of raising on failure. Part 3/4 of Valkey storage implementation. feat(valkey): ValkeyStorage vector memory backend Add ValkeyStorage, a distributed StorageBackend implementation using Valkey-GLIDE with Valkey Search for vector similarity. Wire it into Memory as the 'valkey' storage option. Pin scrapegraph-py<2 to fix unrelated upstream breakage. Part 4/4 of Valkey storage implementation. fix: use datetime.utcnow() for last_accessed consistency MemoryRecord defaults use utcnow() for created_at and last_accessed. Match that in ValkeyStorage.update_record() to avoid timezone inconsistency in recency scoring. feat(valkey): shared cache config + ValkeyCache for A2A and file uploads Extract duplicated Redis URL parsing into a shared cache_config utility. Introduce ValkeyCache as a lightweight async key/value cache using valkey-glide. Wire it into A2A task handling, agent card caching, and file upload caching. Part 1/4 of Valkey storage implementation. fix: handle non-numeric database path in cache URL parsing Extract _parse_db_from_path() helper that catches ValueError for paths like /mydb and defaults to 0 with a warning, instead of crashing. fix: async-safe embeddings and resilient drain_writes Add bytes→float validators on MemoryRecord and ItemState to handle Valkey returning embeddings as raw bytes. Make embed_texts() safe when called from an async context by using a thread pool. Improve drain_writes() with per-save timeouts and error logging instead of raising on failure. Part 3/4 of Valkey storage implementation. fix: catch concurrent.futures.TimeoutError for Python 3.10 compat In Python <3.11, concurrent.futures.TimeoutError is distinct from the builtin TimeoutError. Catch both so the timeout warning path works on all supported Python versions.
- Remove explicit variable annotations using unfollowed glide types (VectorFieldAttributesHnsw | VectorFieldAttributesFlat, list[Field]) which triggered [no-any-unimported] errors on lines 533 and 550 - Remove unused Field import - Remove two unused type: ignore[arg-type] comments on hset() calls that mypy 1.19.1 no longer requires (glide stub now accepts dict[str, str | bytes] directly)
… query
metadata_filter keys are not part of the FT index schema (only scope,
categories, created_at, importance are indexed). Injecting arbitrary
metadata keys as @{key}:{value} clauses in the FT.SEARCH query would
error or silently return wrong results.
Fix: remove metadata_filter from the FT query, over-fetch by 3x when
metadata_filter is present, and apply AND-logic post-filtering on the
deserialized record.metadata dict after retrieval.
Fixes crewAIInc#5795
Signed-off-by: Matthias Howell <matthias.howell@improving.com>
… client Address review feedback on the Valkey storage backend: - _areset(scope_prefix=None) now enumerates all record:* keys and deletes them by id. It previously delegated to adelete() with no filter, which is a no-op by design, so full reset (Memory.reset_all / unscoped reset) deleted nothing. - vector_dim defaults to None and is auto-detected from the first embedding (save) or the query vector (search), mirroring LanceDBStorage. Hardcoding 1536 mismatched the default embedder and built the index at the wrong size. - Scope searches over-fetch (limit*3) like metadata_filter does, so the scope-boundary post-filter that drops sibling matches can't shrink results below limit. - ValkeyCache/ValkeyCacheBackend now accept use_tls and pass it to Glide, so A2A cancellation and the upload cache honor rediss/valkeys TLS endpoints (ValkeyStorage already did). - Add client_info_tag="crewai" to the storage Glide client (LIB-NAME GlidePy(crewai)) for CLIENT INFO attribution; metadata only. Tests updated for the over-fetch and full-reset behavior. Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
…cancel watcher - created_at is stored as a numeric epoch instead of an ISO string so the FT created_at NumericField can actually index it (Valkey Search cannot parse an ISO string as a number, which could skip/ fail indexing). Deserialization parses the epoch and still falls back to ISO for any legacy values. - _ensure_vector_index treats an 'already exists' FT.CREATE error as success, so a concurrent worker (or a create that races the FT.LIST check) reuses the index instead of making save/search raise. - A2A cancel watcher: the Valkey poll now swallows cache errors and keeps polling, and the decorator only treats the watcher as a cancel when it truly returned True. A connection error previously completed the watcher and looked like a real cancel, killing the running task. Tests updated for numeric created_at; added parse round-trip test. Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
ValkeyStorage accepted vectors without comparing their dimension to the index, so after an embedder change (e.g. text-embedding-3-small 1536 -> -large 3072) saves succeeded while Valkey Search silently dropped or mis-indexed the new vectors and recall quietly missed them. Add _check_and_set_vector_dim: reject batches with inconsistent embedding lengths, adopt the dimension on first use, and raise the shared EmbeddingDimensionMismatchError (a ValueError, so background-save plumbing does not swallow it as an executor shutdown) when a batch differs from the established dimension. Wire it into asave and _aupdate. Also correct the stale 'default 1536' docstring to reflect the None/auto-detect default. Mirrors the LanceDB and Qdrant behavior. Adds regression tests. Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
- FT.SEARCH now sets dialect=2 in FtSearchOptions. Valkey Search only parses the =>[KNN ...] vector clause under dialect 2, so without it a live module ignores or rejects the vector query (recall outage; mocked tests never caught it). - Records without an embedding no longer write the fixed-size FLOAT32 vector field as empty bytes, which made Valkey Search fail to index the hash. The field is omitted so the record is simply left out of the vector index. Updated the no-embedding save test to expect the field to be absent. Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
_vector_search injected scope_prefix into the FT.SEARCH tag clause without stripping a trailing slash, so a query for '/crew/' produced @scope:{/crew/*} and never matched a record stored as '/crew'. delete, list, count, and the boundary post-filter all rstrip('/') first, so recall silently diverged from those operations. Normalize the same way (rstrip('/') or '/'). Adds a regression test. Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
…stamps - _ensure_vector_index now reads an existing memory_index's vector dimension via FT.INFO and reconciles it with _vector_dim: adopts it when unset (so a process restart picks up the real size) and raises EmbeddingDimensionMismatch Error when the current embedder differs. Previously any existing index was trusted, so after a restart the first save/search adopted the new embedder size and wrote/queried against an index built at the old size (silent recall misses). The check is deliberately outside the broad FT.LIST except so the ValueError-based mismatch error is not swallowed. Verified against a live valkey-bundle (1536 index + 3072 embedder raises; matching dim does not). - created_at epochs are now UTC-consistent. MemoryRecord uses naive datetime.utcnow(); a naive .timestamp() is interpreted as local time, so stored NumericField/zset scores and age-filter thresholds were shifted by the host offset. Add _to_epoch (treat naive as UTC) and use it for the stored created_at, scope zset scores, and the older_than threshold; parse back as naive UTC. Adds parser, dimension, and timezone regression tests. Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
…s on update _aupdate removed the record from its scope/category/metadata indexes before running the dimension guard. When the guard raised EmbeddingDimensionMismatchError, the hash was left in place but the secondary indexes were already cleared, orphaning the record (invisible to list, count, scope, and filtered delete while still stored). Move the check to the top of _aupdate, before any read or index removal, and also ensure/validate the index there so the process-restart mismatch is caught on the update path too. Adds a regression test asserting no zrem/srem/hset on mismatch. Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
- _vector_search now compares the query embedding length to the established _vector_dim after ensuring the index, raising EmbeddingDimensionMismatchError instead of sending a wrong-sized blob to Valkey Search (which errors opaquely or returns empty/wrong hits, so recall looks like memory vanished). Matches LanceDB and Qdrant. - The FT.CREATE 'already exists' race branch now reads the existing index dimension via FT.INFO and reconciles/raises, instead of assuming success - a concurrent worker may have built memory_index at a different size. Factor the existing-index reconciliation into _reconcile_existing_index_dim and reuse it from both the FT.LIST-exists and create-race paths. Adds a search mismatch regression test. Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
…e age cutoff
- reset() now drops memory_index and clears _index_created/_vector_dim (restoring
any explicitly-configured dimension), so the documented 'reset then rebuild
with a new embedder' recovery path works instead of tripping
EmbeddingDimensionMismatchError on stale index state. [High]
- _escape_search_query now escapes whitespace, so a multi-word category/scope
stays a single TAG term in @categories:{tag} instead of splitting into
separate terms that miss or mis-select records. [Medium]
- _find_records_older_than uses an exclusive upper ScoreBoundary so a record
whose created_at equals older_than is not deleted, matching the 'created
before' contract and LanceDB's strict < comparison. [Low]
Adds regression tests for all three.
Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
… flag A full reset() drops the shared memory_index, but only the calling instance cleared _index_created. Other processes kept the flag set and skipped _ensure_vector_index, so their later search hit a missing index and save wrote hashes that were never vector-indexed. _ensure_vector_index no longer takes the _index_created fast exit; it always confirms the index exists via FT.LIST (cheap, batch/search-level, not per-record) and recreates it when missing. The expensive FT.INFO dimension reconciliation still runs only once per process (gated on _index_created). Verified live: a worker with a stale flag recreates the index on its next save after another worker's reset. Adds a regression test. Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
Remove unused '# noqa: BLE001' directives in task.py and valkey_storage.py flagged by RUF100 (BLE001 is not enabled in this project's ruff config). No behavior change. Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
- Metadata post-filter now requires the key to be present and compares by value instead of str(rec.metadata.get(k)) == str(v). The old coercion matched a missing key against 'None'/None and treated 5 == '5', diverging from LanceDB/ Qdrant and returning the wrong recall set. - _alist_records no longer hydrates every record in scope. It collects (id, score) pairs from the scope ZSETs, orders newest-first, slices to the requested page, then hgetalls only that page — so a limit=200 list reads ~200 records instead of the whole keyspace (which risked the 2s request timeout). Factored scope-key discovery into _find_scope_keys, reused by _find_records_by_scope. - ValkeyStorage honors an ACL username: new username param threaded into ServerCredentials and the Memory storage factory, matching the cache path. Adds pagination-efficiency and typed-metadata regression tests. Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
CI runs 'ruff format --check'; wrap the line ruff prefers in _alist_records. No behavior change. Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
…lank ACL user, empty query
- _vector_search: treat root scope '/' as no scope filter instead of injecting
a standalone '*'. Combined with a category clause the old code produced
'(* @categories:{...})', which Valkey Search rejects ('*' is only valid as a
standalone match-all), so category-filtered recall at root returned nothing.
- _vector_search: short-circuit to [] when the query embedding is empty rather
than issuing a KNN with a zero-length blob.
- _aupdate: HDEL the embedding field when a content-only update carries no
embedding. HSET never removes omitted fields, so the hash kept the old vector
and Search kept ranking the record by a stale embedding.
- ServerCredentials: normalize an empty username to None so a password-only URL
does password auth instead of authenticating as a blank ACL user.
Signed-off-by: MatthiasHowellYopp <matthias.howell@improving.com>
MatthiasHowellYopp
force-pushed
the
feat/valkey-4-storage
branch
from
October 9, 2026 18:46
4d1c7fe to
e247705
Compare
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using high effort and found 3 potential issues.
There are 6 total unresolved issues (including 3 from previous reviews).
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
Want fixes drafted automatically? Bugbot Autofix can create code changes for findings. A team admin can enable Autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit e247705. Configure here.
| for score_key in ("__score", "score"): | ||
| if score_key in field_dict: | ||
| distance = float(field_dict[score_key]) | ||
| score = max(0.0, min(1.0, 1.0 - (distance / 2.0))) |
There was a problem hiding this comment.
Incorrect vector similarity conversion
Medium Severity
Valkey Search COSINE returns distance 1 - cosine_similarity, but _parse_search_result maps it with 1 - distance / 2 (i.e. (1 + sim) / 2). Encoding and consolidation compare that score to a true cosine threshold such as consolidation_threshold 0.85, so Valkey triggers merges around 0.70 similarity and ranks recall higher than LanceDB or in-batch cosine.
Reviewed by Cursor Bugbot for commit e247705. Configure here.
| for key, value in metadata_filter.items(): | ||
| value_str = str(value) | ||
| metadata_key = self._metadata_key(key, value_str) | ||
| members = await client.smembers(metadata_key) |
There was a problem hiding this comment.
Delete metadata ignores value types
Medium Severity
adelete resolves metadata_filter only through metadata:{key}:{str(value)} sets. Search post-filters with typed equality (5 is not "5"). The same filter can therefore delete records that search would not return, including collisions like True vs "True" or None vs "None".
Additional Locations (1)
Reviewed by Cursor Bugbot for commit e247705. Configure here.
| self._storage = LanceDBStorage() | ||
| elif self.storage == "valkey": | ||
| from crewai.memory.storage.valkey_storage import ValkeyStorage | ||
| from crewai.utilities.cache_config import parse_cache_url |
There was a problem hiding this comment.
Missing Valkey extra install hint
Medium Severity
Selecting storage="valkey" or the upload valkey backend imports glide through ValkeyStorage / ValkeyCache with no install hint. A missing crewai[valkey] extra surfaces as a raw ModuleNotFoundError for glide, unlike the A2A path which raises ImportError telling the user to install crewai[valkey].
Additional Locations (1)
Triggered by learned rule: Optional extras raise ImportError not uv add
Reviewed by Cursor Bugbot for commit e247705. Configure here.
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Title: feat(valkey): ValkeyStorage vector memory backend
Description:
Part 4/4 of adding Valkey as a storage backend for CrewAI. This PR adds the core vector storage implementation and wires it into the memory system. Depends on parts 1 and 3.
What changed:
valkey_storage.py (new) — Full StorageBackend implementation using Valkey-GLIDE. Supports vector similarity search via Valkey Search module (FLAT and HNSW indexes), scope/category/metadata filtering through tag and numeric indexes, and both sync and async interfaces. Handles lazy client and index initialization, automatic index creation, and batch save operations with JSON + binary embedding serialization.
unified_memory.py — Added "valkey" as a recognized storage option. When selected, reads connection details from VALKEY_URL/REDIS_URL via the cache_config utility from part 1 and instantiates ValkeyStorage.
lib/crewai-tools/pyproject.toml — Tightened the scrapegraph-py extra to >=1.46.0,<2 (was >=1.9.0,<2) so the 2.x release that dropped the Client class stays excluded. This pin used to sit in the root override-dependencies list; moving it into the crewai-tools extra keeps it out of the security-pin block upstream edits often, which was causing this branch to hit the same pyproject.toml conflict on every rebase.
uv.lock — The lock diff here is large but version-neutral. Pulling the entry out of override-dependencies makes uv re-express dependency markers across the graph, but the resolved version of every package is unchanged.
Fixes #5795 — metadata_filter in _vector_search was incorrectly injected into the FT.SEARCH query as @{key}:{value} field clauses, but these arbitrary metadata keys are not in the FT index schema. Fixed by applying metadata_filter as a post-filter over search results (with 3x over-fetch to compensate for filtering loss).
Testing:
test_valkey_storage.py — Core CRUD operations, batch saves, record retrieval, deletion, flushing, TTL, metadata handling, connection management
test_valkey_storage_errors.py — Connection failures, index creation errors, malformed data, graceful degradation
test_valkey_storage_scope.py — Hierarchical scope queries, scope listing, child scope enumeration, cross-scope isolation
test_valkey_storage_search.py — Vector similarity search, composite scoring, category filtering, metadata post-filtering, limit/offset, empty results
All tests use mocked Valkey clients and do not require a running Valkey instance.
Note
High Risk
Large new distributed storage/cache surface (vector index lifecycle, shared Valkey DB) plus security-sensitive A2A cancellation and a deliberate change away from remote pickle for agent cards; misconfiguration could affect task cancel behavior or memory recall correctness.
Overview
Adds Valkey as a first-class option across memory, caching, and A2A: a new
ValkeyStoragebackend implements vector CRUD/search (Valkey Search KNN, scope/category indexes, metadata post-filtering per #5795), andMemory(storage="valkey")connects via sharedcache_configURL parsing (VALKEY_URL/REDIS_URL).Introduces
ValkeyCache(JSON, TTL, namespace clear) and refactorsUploadCachebehind pluggable backends (memory/redis/valkey), including safer Redis clears (namespace-scoped), lazycrewaiimports, and a minimum 1s TTL whenexpires_atis in the past.A2A task cancellation can route cancel flags through Valkey when
VALKEY_URLis set (polling with error resilience; clearer error ifcrewai[valkey]is missing). Agent card fetching is pinned to in-processSimpleMemoryCacheso pickled values are never read from operator-controlled Redis/Valkey.Related memory hardening:
embed_textsoffloads blocking embedders with optional 30s fail-loud timeouts on save paths,drain_writestimeouts untrack stuck futures, and recall/save memory tools validate requiredqueries/contentsargs.Also adds optional
crewai[valkey](valkey-glide>=2.5.2), tightensscrapegraph-pyin crewai-tools, and extends test coverage for Valkey storage errors, upload cache, and A2A cache behavior.Reviewed by Cursor Bugbot for commit e247705. Bugbot is set up for automated code reviews on this repo. Configure here.
Valkey library-name tag
Sets the Valkey client's library-name tag to
crewai(via valkey-glideclient_info_tag), so the connection reportslib-name=GlidePy(crewai)inCLIENT INFO. This lets operators attribute Valkey usage to CrewAI. Metadata only — no behavioural change. Requiresvalkey-glide>=2.5.2, whereclient_info_tagwas introduced.client_nameis unchanged;CLIENT SETNAMEandCLIENT SETINFO LIB-NAMEare independent fields.