feat(pipecat): voice agent integration with persistent memory#54
Merged
Conversation
Adds a first-class Pipecat integration so developers can drop Vektori
long-term memory into any real-time voice pipeline in ~10 lines.
vektori/integrations/pipecat/processor.py
- VektoriMemoryProcessor: intercepts OpenAILLMContextFrame before the
LLM, searches Vektori (facts + episodes), injects a [Memory context]
block into the system prompt each turn
- VektoriStorageProcessor: buffers user transcription + streamed LLM
reply, stores the completed turn in Vektori on LLMFullResponseEndFrame
vektori/integrations/pipecat/__init__.py
- clean public API with helpful ImportError if pipecat-ai not installed
vektori/integrations/__init__.py
- namespace package for future integrations
examples/pipecat_voice_agent.py
- full FastAPI WebSocket server using Deepgram STT, GPT-4o-mini,
ElevenLabs TTS, Silero VAD, and both Vektori processors
- configurable via env vars; documented alternative STT/TTS providers
pyproject.toml
- new [pipecat] optional dependency group: pipecat-ai, fastapi, uvicorn
|
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 a Pipecat integration and examples for voice agents, introduces ExtractionConfig and agent-type extraction plumbing into Vektori, provides Pipecat FrameProcessors for memory injection and storage, and exposes a FastAPI WebSocket voice example with per-connection pipelines. Changes
Sequence DiagramsequenceDiagram
participant Client as WebSocket Client
participant FastAPI
participant Pipeline as Pipecat Pipeline
participant VAD as Silero VAD
participant STT as Deepgram STT
participant MemProc as VektoriMemoryProcessor
participant LLM as OpenAI LLM
participant StorProc as VektoriStorageProcessor
participant Vektori
participant TTS as ElevenLabs TTS
Client->>FastAPI: WebSocket connect (/ws/{user_id})
FastAPI->>Pipeline: create session pipeline (session_id)
Client->>VAD: stream audio
VAD->>STT: voice segments
STT->>Pipeline: TranscriptionFrame
Pipeline->>MemProc: OpenAILLMContextFrame
MemProc->>Vektori: search(user_id, depth, top_k)
Vektori-->>MemProc: search results (facts, episodes)
MemProc->>LLM: OpenAILLMContextFrame (with injected system memory)
LLM->>Pipeline: streaming TextFrame
Pipeline->>StorProc: buffer TranscriptionFrame + TextFrame
LLM->>Pipeline: LLMFullResponseEndFrame
StorProc->>Vektori: add(user_id, session_id, [user, assistant])
Vektori-->>StorProc: ack
Pipeline->>TTS: final assistant text
TTS->>Client: audio output
Client->>FastAPI: WebSocket disconnect
FastAPI->>Vektori: close(session)
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
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: 5
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@examples/pipecat_voice_agent.py`:
- Around line 98-101: The module currently reads DEEPGRAM_API_KEY and
ELEVENLABS_API_KEY/ELEVENLABS_VOICE_ID at import time which forces those
providers to be configured even when using alternatives (e.g.,
WhisperSTTService); move these os.environ/os.getenv lookups out of the top-level
scope and instead fetch them only when constructing the provider-specific
services (e.g., inside the Deepgram-related setup and the ElevenLabs TTS setup)
or guard them behind the chosen provider selection logic so DEEPGRAM_API_KEY,
ELEVENLABS_API_KEY and ELEVENLABS_VOICE_ID are only required when their
corresponding provider is actually used.
In `@vektori/integrations/pipecat/__init__.py`:
- Around line 12-19: The current except ImportError block around the from
.processor import VektoriMemoryProcessor, VektoriStorageProcessor masks any
ImportError as "Pipecat is not installed"; change it to catch only
ModuleNotFoundError, inspect the exception to ensure the missing module is the
pipecat dependency (e.g., check exc.name or "pipecat" in str(exc)), and raise
the user-friendly message only in that case; for any other
ImportError/ModuleNotFoundError that is not pipecat-related, re-raise the
original exception so real bugs (typos, missing symbols) are not hidden.
In `@vektori/integrations/pipecat/processor.py`:
- Around line 167-194: _inject_memory currently returns early on missing query
or search failure or empty memory while leaving any previously injected "[Memory
context]" in context; update _inject_memory to restore the system message to
self._base_system (or remove the memory block if no base exists) before each
early return (i.e., before returning when query is falsy, in the except block
for self._vektori.search, and when memory_text is empty) by calling
_set_system_message(context, self._base_system) or _set_system_message(context,
None) as appropriate so stale memory is not retained across turns.
- Around line 145-154: The constructor annotations use quoted "Vektori" which is
redundant due to from __future__ import annotations and causes Ruff UP037;
update the two __init__ signatures (the one at the top of the class and the
second constructor around line ~254) to use unquoted Vektori (vektori: Vektori)
instead of "Vektori", keeping all other parameters unchanged; this removes the
unnecessary string annotations while preserving the TYPE_CHECKING import usage.
- Around line 283-287: The current code reads the previous assistant message
from context and schedules self._store_turn() via asyncio.ensure_future() while
holding live references to self._pending_user (string) and
self._pending_assistant_parts (list), causing races when
allow_interruptions=True; fix by snapshotting the buffers into local immutable
copies (e.g. copy the string and make a shallow copy of the list) before the
asyncio.ensure_future(...) call so _store_turn() gets stable data, and also
remove the assistant-context fallback that uses _last_message(self._context,
"assistant") (or alternatively reorder the pipeline so ctx_agg.assistant() runs
before vektori_storage) to ensure you don't read the previous turn's assistant
text.
🪄 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: defaults
Review profile: CHILL
Plan: Pro
Run ID: 23cc36f8-740b-477e-a169-85ed7c8f3e46
📒 Files selected for processing (5)
examples/pipecat_voice_agent.pypyproject.tomlvektori/integrations/__init__.pyvektori/integrations/pipecat/__init__.pyvektori/integrations/pipecat/processor.py
Comment on lines
+98
to
+101
| DEEPGRAM_API_KEY = os.environ["DEEPGRAM_API_KEY"] | ||
|
|
||
| ELEVENLABS_API_KEY = os.environ["ELEVENLABS_API_KEY"] | ||
| ELEVENLABS_VOICE_ID = os.getenv("ELEVENLABS_VOICE_ID", "21m00Tcm4TlvDq8ikWAM") |
There was a problem hiding this comment.
Load provider-specific API keys only when that provider is selected.
Line 98 and Line 100 make Deepgram and ElevenLabs mandatory at import time, so the example still crashes after swapping to WhisperSTTService or another TTS as the docstring suggests. Resolve those env vars next to the concrete service setup, or gate them behind an explicit provider choice.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@examples/pipecat_voice_agent.py` around lines 98 - 101, The module currently
reads DEEPGRAM_API_KEY and ELEVENLABS_API_KEY/ELEVENLABS_VOICE_ID at import time
which forces those providers to be configured even when using alternatives
(e.g., WhisperSTTService); move these os.environ/os.getenv lookups out of the
top-level scope and instead fetch them only when constructing the
provider-specific services (e.g., inside the Deepgram-related setup and the
ElevenLabs TTS setup) or guard them behind the chosen provider selection logic
so DEEPGRAM_API_KEY, ELEVENLABS_API_KEY and ELEVENLABS_VOICE_ID are only
required when their corresponding provider is actually used.
Comment on lines
+12
to
+19
| try: | ||
| from .processor import VektoriMemoryProcessor, VektoriStorageProcessor | ||
|
|
||
| __all__ = ["VektoriMemoryProcessor", "VektoriStorageProcessor"] | ||
| except ImportError as exc: # pipecat not installed | ||
| raise ImportError( | ||
| "Pipecat is not installed. Run: pip install 'vektori[pipecat]'" | ||
| ) from exc |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
# First, locate and examine the processor.py file
find . -path "*vektori/integrations/pipecat*" -type f -name "*.py" | head -20Repository: vektori-ai/vektori
Length of output: 148
🏁 Script executed:
# Examine the __init__.py file in question
cat -n vektori/integrations/pipecat/__init__.pyRepository: vektori-ai/vektori
Length of output: 726
🏁 Script executed:
# Examine the processor.py file
cat -n vektori/integrations/pipecat/processor.pyRepository: vektori-ai/vektori
Length of output: 13870
Don't mask non-Pipecat import failures.
Line 16 catches every ImportError raised while importing from .processor, so a real bug (missing dependency, typo, etc.) will be rethrown as "Pipecat is not installed," which is misleading. Use ModuleNotFoundError instead to catch only missing-module errors, check that it's pipecat-related, and re-raise everything else unchanged.
🐛 Proposed fix
try:
from .processor import VektoriMemoryProcessor, VektoriStorageProcessor
__all__ = ["VektoriMemoryProcessor", "VektoriStorageProcessor"]
-except ImportError as exc: # pipecat not installed
- raise ImportError(
- "Pipecat is not installed. Run: pip install 'vektori[pipecat]'"
- ) from exc
+except ModuleNotFoundError as exc:
+ if exc.name and exc.name.startswith("pipecat"):
+ raise ImportError(
+ "Pipecat is not installed. Run: pip install 'vektori[pipecat]'"
+ ) from exc
+ raise🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/integrations/pipecat/__init__.py` around lines 12 - 19, The current
except ImportError block around the from .processor import
VektoriMemoryProcessor, VektoriStorageProcessor masks any ImportError as
"Pipecat is not installed"; change it to catch only ModuleNotFoundError, inspect
the exception to ensure the missing module is the pipecat dependency (e.g.,
check exc.name or "pipecat" in str(exc)), and raise the user-friendly message
only in that case; for any other ImportError/ModuleNotFoundError that is not
pipecat-related, re-raise the original exception so real bugs (typos, missing
symbols) are not hidden.
Comment on lines
+167
to
+194
| async def _inject_memory(self, context: OpenAILLMContext) -> None: | ||
| """Search Vektori and update the system message in *context* in-place.""" | ||
| query = _last_message(context, "user") | ||
| if not query: | ||
| return | ||
|
|
||
| try: | ||
| memory = await self._vektori.search( | ||
| query=query, | ||
| user_id=self._user_id, | ||
| depth=self._depth, | ||
| top_k=self._top_k, | ||
| ) | ||
| except Exception: | ||
| logger.exception("VektoriMemoryProcessor: search failed — skipping injection") | ||
| return | ||
|
|
||
| memory_text = _format_memory(memory) | ||
| if not memory_text: | ||
| return | ||
|
|
||
| new_system = self._base_system | ||
| if new_system: | ||
| new_system += f"\n\n[Memory context]\n{memory_text}" | ||
| else: | ||
| new_system = f"[Memory context]\n{memory_text}" | ||
|
|
||
| _set_system_message(context, new_system) |
There was a problem hiding this comment.
Reset stale memory when a lookup misses or fails.
context is reused across turns, so returning on Line 171, Line 182, or Line 186 leaves the previous [Memory context] block in place. After one successful lookup, a later miss/error will still bias unrelated prompts with stale memory. Restore self._base_system before those early exits, or remove the injected block entirely when there is no base prompt.
🧠 Proposed fix
async def _inject_memory(self, context: OpenAILLMContext) -> None:
"""Search Vektori and update the system message in *context* in-place."""
query = _last_message(context, "user")
if not query:
+ _set_system_message(context, self._base_system)
return
@@
except Exception:
logger.exception("VektoriMemoryProcessor: search failed — skipping injection")
+ _set_system_message(context, self._base_system)
return
memory_text = _format_memory(memory)
if not memory_text:
+ _set_system_message(context, self._base_system)
return🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/integrations/pipecat/processor.py` around lines 167 - 194,
_inject_memory currently returns early on missing query or search failure or
empty memory while leaving any previously injected "[Memory context]" in
context; update _inject_memory to restore the system message to
self._base_system (or remove the memory block if no base exists) before each
early return (i.e., before returning when query is falsy, in the except block
for self._vektori.search, and when memory_text is empty) by calling
_set_system_message(context, self._base_system) or _set_system_message(context,
None) as appropriate so stale memory is not retained across turns.
Comment on lines
+283
to
+287
| # Fall back to reading from context if frame buffers are empty | ||
| if not user_text and self._context: | ||
| user_text = _last_message(self._context, "user") or "" | ||
| if not assistant_text and self._context: | ||
| assistant_text = _last_message(self._context, "assistant") or "" |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
find . -type f -name "pipecat_voice_agent.py"Repository: vektori-ai/vektori
Length of output: 95
🏁 Script executed:
cd . && git ls-files | head -20Repository: vektori-ai/vektori
Length of output: 622
🏁 Script executed:
fd -e py | grep pipecatRepository: vektori-ai/vektori
Length of output: 176
🏁 Script executed:
cat -n vektori/integrations/pipecat/processor.py | sed -n '270,350p'Repository: vektori-ai/vektori
Length of output: 2991
🏁 Script executed:
cat -n examples/pipecat_voice_agent.py | sed -n '220,250p'Repository: vektori-ai/vektori
Length of output: 1695
🏁 Script executed:
rg "_last_message" vektori/integrations/pipecat/processor.py -B 2 -A 5Repository: vektori-ai/vektori
Length of output: 1090
🏁 Script executed:
rg "def _last_message" vektori/integrations/pipecat/processor.py -A 15Repository: vektori-ai/vektori
Length of output: 735
Snapshot buffers before scheduling async task and remove assistant-context fallback or reorder pipeline.
Line 287 reads the previous assistant message from context because ctx_agg.assistant() (line 231) runs downstream after vektori_storage in the pipeline—the current turn's assistant message hasn't been aggregated yet.
Additionally, line 331 schedules self._store_turn() asynchronously via asyncio.ensure_future() without snapshotting self._pending_user and self._pending_assistant_parts. These are live references: a string (line 322) and a mutable list (line 327). With allow_interruptions=True enabled, new frames can arrive during the async task execution, causing self._pending_user to be overwritten (line 322) or self._pending_assistant_parts to be mutated (line 327) before _store_turn() reads them at lines 280–281.
Snapshot both buffers before line 331, and either remove the assistant-context fallback at line 287 or reorder the pipeline so vektori_storage runs after ctx_agg.assistant().
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/integrations/pipecat/processor.py` around lines 283 - 287, The
current code reads the previous assistant message from context and schedules
self._store_turn() via asyncio.ensure_future() while holding live references to
self._pending_user (string) and self._pending_assistant_parts (list), causing
races when allow_interruptions=True; fix by snapshotting the buffers into local
immutable copies (e.g. copy the string and make a shallow copy of the list)
before the asyncio.ensure_future(...) call so _store_turn() gets stable data,
and also remove the assistant-context fallback that uses
_last_message(self._context, "assistant") (or alternatively reorder the pipeline
so ctx_agg.assistant() runs before vektori_storage) to ensure you don't read the
previous turn's assistant text.
…nConfig
Adds ExtractionConfig — a four-level customisation API for Vektori's LLM
fact and episode extraction. Designed for teams running specialised agents
(pre-sales, sales, support, HR) who need extraction biased toward their
domain without replacing the core prompt logic.
Level 1 — agent_type preset (zero effort):
Vektori(agent_type="presales") # focuses on pain points, budget, DMs, objections
Vektori(agent_type="sales") # focuses on deal stage, pricing, close date, blockers
Vektori(agent_type="support") # focuses on issue details, resolution, CSAT signals
Vektori(agent_type="hr") # focuses on feedback, performance, career goals
Level 2 — domain hints (low effort):
ExtractionConfig(agent_type="presales", focus_on=["ICP fit"], ignore=["small talk"])
Level 3 — prompt suffix (precise control without rewriting the prompt):
ExtractionConfig(facts_prompt_suffix="Always extract exact dollar amounts.")
Level 4 — full prompt override (escape hatch, must keep same JSON schema):
ExtractionConfig(custom_facts_prompt=MY_TEMPLATE)
Changes:
vektori/config.py — ExtractionConfig dataclass + VektoriConfig.extraction_config
vektori/__init__.py — export ExtractionConfig at top level
vektori/ingestion/extractor.py — _AGENT_FACTS_GUIDANCE/_AGENT_EPISODES_GUIDANCE presets,
_build_domain_guidance_* / _facts_prompt / _episodes_prompt
methods, {domain_guidance} slot in both prompt templates
vektori/client.py — agent_type + extraction_config constructor params,
wires ExtractionConfig into FactExtractor
examples/agent_type_extraction.py — runnable demo comparing presales/sales/general
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (2)
examples/agent_type_extraction.py (1)
140-142:asyncio.sleep(3)is a heuristic wait.For a demo script this is acceptable, but consider adding a comment noting that in production, callers should use
async_extraction=Falsefor deterministic behavior or poll for completion.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@examples/agent_type_extraction.py` around lines 140 - 142, The demo uses a heuristic asyncio.sleep(3) to wait for async extraction before calling presales_v.search("budget and decision makers", user_id="demo-user"); update the comment to explicitly state this is only for the demo and explain two production-safe alternatives: (1) call the extraction synchronously by setting async_extraction=False where extraction is invoked so presales_v.search is deterministic, or (2) implement a polling loop that checks extraction completion before calling presales_v.search; reference asyncio.sleep, presales_v.search and async_extraction=False in the comment so future readers know the preferred approaches.tests/unit/test_agent_type_presets.py (1)
40-44: Good edge case: verifies "general" returns empty guidance.This confirms the default behavior doesn't inject any domain bias.
Consider adding a test for an unknown/invalid
agent_typeto verify graceful fallback (e.g.,agent_type="nonexistent"should behave like"general").💡 Optional: test for unknown agent_type
def test_unknown_agent_type_has_no_domain_guidance(): extractor = _build_extractor("nonexistent_type") assert extractor._build_domain_guidance_facts() == "" assert extractor._build_domain_guidance_episodes() == ""🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tests/unit/test_agent_type_presets.py` around lines 40 - 44, Add a test that verifies an unknown/invalid agent_type falls back to the general behavior: call _build_extractor with a nonexistent agent type (e.g., "nonexistent_type") and assert that extractor._build_domain_guidance_facts() and extractor._build_domain_guidance_episodes() both return empty strings, mirroring the existing test_general_agent_type_has_no_domain_guidance and using the same helper _build_extractor and methods _build_domain_guidance_facts/_build_domain_guidance_episodes to locate the behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@examples/agent_type_extraction.py`:
- Line 39: Remove the unused import "json" from the top-level imports in
examples/agent_type_extraction.py; locate the import statement that reads
"import json" and delete it so there are no unused imports in the module.
- Line 133: Remove the unused variable declaration captured: list[dict] = []
from agent_type_extraction.py; either delete that line entirely or instead wire
the intended capture by passing the _capture_out collector into the add(...)
call where the output hook is registered (refer to the _capture_out and add(...)
symbols) so there are no unused locals left.
In `@vektori/ingestion/extractor.py`:
- Line 436: The type annotation for the parameter extraction_config is currently
a quoted string ("ExtractionConfig | None") which triggers ruff UP037; change it
to a real annotation (ExtractionConfig | None) on the function/method signature
and ensure the ExtractionConfig symbol is imported unconditionally (remove it
from any TYPE_CHECKING-only import block or add a normal import) so the name is
available at runtime and the annotation no longer needs to be quoted.
---
Nitpick comments:
In `@examples/agent_type_extraction.py`:
- Around line 140-142: The demo uses a heuristic asyncio.sleep(3) to wait for
async extraction before calling presales_v.search("budget and decision makers",
user_id="demo-user"); update the comment to explicitly state this is only for
the demo and explain two production-safe alternatives: (1) call the extraction
synchronously by setting async_extraction=False where extraction is invoked so
presales_v.search is deterministic, or (2) implement a polling loop that checks
extraction completion before calling presales_v.search; reference asyncio.sleep,
presales_v.search and async_extraction=False in the comment so future readers
know the preferred approaches.
In `@tests/unit/test_agent_type_presets.py`:
- Around line 40-44: Add a test that verifies an unknown/invalid agent_type
falls back to the general behavior: call _build_extractor with a nonexistent
agent type (e.g., "nonexistent_type") and assert that
extractor._build_domain_guidance_facts() and
extractor._build_domain_guidance_episodes() both return empty strings, mirroring
the existing test_general_agent_type_has_no_domain_guidance and using the same
helper _build_extractor and methods
_build_domain_guidance_facts/_build_domain_guidance_episodes to locate the
behavior.
🪄 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: defaults
Review profile: CHILL
Plan: Pro
Run ID: eb13804f-b3c2-4157-b905-9c4f959f847e
📒 Files selected for processing (6)
examples/agent_type_extraction.pytests/unit/test_agent_type_presets.pyvektori/__init__.pyvektori/client.pyvektori/config.pyvektori/ingestion/extractor.py
✅ Files skipped from review due to trivial changes (1)
- vektori/init.py
| """ | ||
|
|
||
| import asyncio | ||
| import json |
There was a problem hiding this comment.
Remove unused import.
json is imported but never used in the file.
🧹 Proposed fix
import asyncio
-import json
from vektori import ExtractionConfig, Vektori🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@examples/agent_type_extraction.py` at line 39, Remove the unused import
"json" from the top-level imports in examples/agent_type_extraction.py; locate
the import statement that reads "import json" and delete it so there are no
unused imports in the module.
| print("\n[1] agent_type='presales' — built-in preset\n") | ||
|
|
||
| presales_v = Vektori(agent_type="presales") | ||
| captured: list[dict] = [] |
There was a problem hiding this comment.
Remove unused variable.
captured list is declared but never used. The comment mentions _capture_out but it's not passed to add().
🧹 Proposed fix
print("\n[1] agent_type='presales' — built-in preset\n")
presales_v = Vektori(agent_type="presales")
- captured: list[dict] = []
await presales_v.add(
messages=PRESALES_CONVERSATION,
session_id="demo-presales-001",
user_id="demo-user",
- # _capture_out is an internal debug hook used in tests
)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@examples/agent_type_extraction.py` at line 133, Remove the unused variable
declaration captured: list[dict] = [] from agent_type_extraction.py; either
delete that line entirely or instead wire the intended capture by passing the
_capture_out collector into the add(...) call where the output hook is
registered (refer to the _capture_out and add(...) symbols) so there are no
unused locals left.
There was a problem hiding this comment.
Actionable comments posted: 9
🧹 Nitpick comments (4)
vektori/models/openai.py (1)
105-112: Consider aligning type hints with the base class.The method signature uses
dict[str, object]while the abstract base class (ChatModelProvider.complete()) usesdict[str, Any]. While functionally equivalent, usingAnywould be more consistent with the interface contract.♻️ Suggested type alignment
async def complete( self, - messages: list[dict[str, object]], + messages: list[dict[str, Any]], *, - tools: list[dict[str, object]] | None = None, + tools: list[dict[str, Any]] | None = None, max_tokens: int | None = None, temperature: float | None = None, ) -> ChatCompletionResult:You'll also need to add
Anyto the imports fromtyping.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@vektori/models/openai.py` around lines 105 - 112, The complete method in openai.py uses dict[str, object] for message/tool types which differs from the abstract ChatModelProvider.complete() signature; update the type hints in vektori.models.openai.openAI.complete (messages and tools) to use dict[str, Any] to match the base class, and add Any to the typing imports in that module.vektori/models/litellm_provider.py (1)
67-74: Consider aligning type hints with the base class.Same type hint inconsistency as
OpenAIChatModel— usingdict[str, object]instead ofdict[str, Any]from the abstract interface.♻️ Suggested type alignment
async def complete( self, - messages: list[dict[str, object]], + messages: list[dict[str, Any]], *, - tools: list[dict[str, object]] | None = None, + tools: list[dict[str, Any]] | None = None, max_tokens: int | None = None, temperature: float | None = None, ) -> ChatCompletionResult:Add
Anyto imports:from typing import Any🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@vektori/models/litellm_provider.py` around lines 67 - 74, The complete method in litellm_provider.py uses dict[str, object] for messages and tools which mismatches the abstract interface; update the type hints to dict[str, Any] by importing Any from typing (e.g., add "from typing import Any") and replace both occurrences of dict[str, object] in the async def complete signature with dict[str, Any] so the method aligns with ChatCompletionResult's base interface and matches OpenAIChatModel.vektori/cli.py (2)
810-812: Silent exception swallowing may hide parsing errors.The bare
except Exception: passat the file-level silently discards all errors, making it difficult to debug malformed session files. Consider logging a warning with the file path.♻️ Suggested improvement
- except Exception: - pass + except Exception as e: + logger.debug("Failed to parse Claude session %s: %s", path, e) return cwd, messagesYou'll need to add a logger at the top of the file:
import logging logger = logging.getLogger(__name__)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@vektori/cli.py` around lines 810 - 812, The bare except in the block that ends with "return cwd, messages" silently swallows parsing errors; replace it with catching Exception as e and log a warning including the session file path and the exception details instead of pass. Add a module-level logger (import logging; logger = logging.getLogger(__name__)) and call logger.warning or logger.exception with context (e.g., the session file variable name used in this function) so malformed session files are reported while still returning cwd, messages.
855-857: Same issue: silent exception swallowing in Codex parser.Consider logging a debug message for parse failures here as well.
♻️ Suggested improvement
- except Exception: - pass + except Exception as e: + logger.debug("Failed to parse Codex session %s: %s", path, e) return cwd, messages🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@vektori/cli.py` around lines 855 - 857, The except block that swallows all exceptions before returning cwd, messages should log the failure instead of silently passing; change the bare "except Exception: pass" to "except Exception as e:" and emit a debug-level log (e.g., logger.debug("Codex parse failed", exc_info=True) or logger.debug(f"Codex parse failed: {e}", exc_info=True)) so parse failures in the Codex parser are recorded; use the existing logger or import logging if none is available and keep returning cwd, messages after logging.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@examples/openai_agent.py`:
- Around line 28-30: The print call unconditionally accesses result.content
(from agent.chat) which is optional and may be None for tool-only responses;
update the finally block to guard before printing by checking the result object
and its content (e.g., verify result is not None and result.content is truthy or
not None/empty) and only call print(f"Assistant: {result.content}\n") when
content exists (otherwise skip printing or print a clear placeholder), referring
to the agent.chat result and the result.content access to locate the change.
In `@vektori/agent.py`:
- Line 35: The persist_assistant_messages flag is ignored because exchange
always receives the assistant reply; update the message handling so that when
persist_assistant_messages is False you do not include assistant role messages
in the payload sent to exchange and you do not write them to persistence.
Concretely, modify the code paths that build the messages list for the exchange
call (the exchange function) to filter out messages with role "assistant" when
persist_assistant_messages is False, and add the same check before any
persistence/save routine (e.g., any persist_messages/save_messages helper) so
assistant replies are not stored.
- Around line 5-18: Ruff I001 flags the import block ordering; reorder and group
imports into standard library, third-party, and local modules and sort them
alphabetically within each group so the import list in vektori/agent.py complies
with Ruff: group asyncio, dataclass/dataclasses, Path/pathlib, re, typing.Any,
uuid4 into the stdlib block; keep third-party (none here) separate; then list
local project imports (Vektori, AgentContextLoader, InMemoryProfileStore,
ProfilePatch, ProfileStore, SQLiteProfileStore, MessageWindow,
ChatModelProvider, build_messages, should_retrieve) alphabetically and remove
any unused imports if present so symbols like asyncio, dataclass, Path, re, Any,
uuid4, Vektori, AgentContextLoader, InMemoryProfileStore, ProfilePatch,
ProfileStore, SQLiteProfileStore, MessageWindow, ChatModelProvider,
build_messages, and should_retrieve are correctly grouped and ordered.
- Around line 88-121: The code mutates self.window before any awaited work
(calls like _should_retrieve, memory.search, context_loader.load,
profile_store.list_active, build_messages, model.complete) so failures leave a
dangling user-only turn; instead, keep the incoming user_message in a local
variable and only call self.window.add("user", user_message) (and later
self.window.add("assistant", assistant_content) and await
self.window.compact(self.model)) after model.complete returns successfully; if
model.complete raises, do not modify self.window and let the caller retry, and
ensure the retrieval logic (_should_retrieve / memory.search) and message
construction use the local user_message rather than mutating the window early.
In `@vektori/context.py`:
- Around line 27-33: The load() method currently returns an empty
LoadedAgentContext when _resolve_path() returns None, hiding errors when an
explicit context_path was provided; change load() to detect when a user-supplied
context path is invalid by inspecting the object's context_path (or equivalent
constructor arg) and, if present but _resolve_path() returned None, raise a
clear exception (e.g. FileNotFoundError or ValueError) instead of returning an
empty LoadedAgentContext; keep the existing behavior only for the case where no
explicit context_path was given, and apply the same change to the other
context-loading variants referenced in the file (the branches that call
_load_yaml and _load_markdown) so invalid explicit paths fail fast.
- Around line 72-85: The _load_yaml function should guard against yaml.safe_load
returning non-dict values and ensure `agent` is a dict before using .get or
.items; update _load_yaml to treat `raw = yaml.safe_load(...) or {}` as before
but coerce/validate `agent = raw.get("agent", {})` to `agent = agent if
isinstance(agent, dict) else {}` (or use a dict(...) conversion) so all
subsequent accesses for persona, instructions, response_style, memory and the
extra_sections comprehension over agent.items() are safe; ensure the return
constructs for LoadedAgentContext still default to empty values when keys are
missing or when agent was not a dict.
In `@vektori/memory/profile.py`:
- Around line 53-55: The current save implementation appends every ProfilePatch
to self._patches[key], causing multiple active patches for the same
(observer_id, observed_id) and making list_active() return stale duplicates;
change save(self, patch: ProfilePatch) to supersede prior active patches for
that key by replacing the stored entry instead of appending (i.e., set
self._patches[key] = [patch] or otherwise remove/mark previous patches as
inactive), and apply the same replacement logic to the other save implementation
referenced in the comment (the save block around lines 86-109) so list_active()
only returns the single most recent active patch per key.
- Around line 5-12: The import block is misordered and duplicates typing
imports; reorder and consolidate imports to satisfy Ruff I001: group
standard-library imports alphabetically and combine typing names into one line.
Specifically, keep standard-library imports only, sort them (for example: from
dataclasses import dataclass, field; from datetime import datetime, timezone;
import json; from pathlib import Path; import sqlite3; from typing import Any,
Protocol), and remove the separate "from typing import Protocol" line so "Any"
and "Protocol" are imported together.
In `@vektori/memory/window.py`:
- Around line 71-90: The compaction logic in compact() can overwrite prior
context and drop messages when the summarizer returns empty/weak output; change
compact() so it only updates self._rolling_summary when result.content is
non-empty (and instead appends or merges new summary into the existing
self._rolling_summary rather than replacing it), avoid trimming/losing
self._recent_messages if the summary is empty (i.e., abort the removal/keep
original kept_messages), and only increment self._compaction_count on a
successful non-empty summary; use the existing symbols summarizer.complete,
self.summary_max_tokens, result.content, self._rolling_summary,
self._recent_messages and self._compaction_count to locate and implement these
checks/merge behavior.
---
Nitpick comments:
In `@vektori/cli.py`:
- Around line 810-812: The bare except in the block that ends with "return cwd,
messages" silently swallows parsing errors; replace it with catching Exception
as e and log a warning including the session file path and the exception details
instead of pass. Add a module-level logger (import logging; logger =
logging.getLogger(__name__)) and call logger.warning or logger.exception with
context (e.g., the session file variable name used in this function) so
malformed session files are reported while still returning cwd, messages.
- Around line 855-857: The except block that swallows all exceptions before
returning cwd, messages should log the failure instead of silently passing;
change the bare "except Exception: pass" to "except Exception as e:" and emit a
debug-level log (e.g., logger.debug("Codex parse failed", exc_info=True) or
logger.debug(f"Codex parse failed: {e}", exc_info=True)) so parse failures in
the Codex parser are recorded; use the existing logger or import logging if none
is available and keep returning cwd, messages after logging.
In `@vektori/models/litellm_provider.py`:
- Around line 67-74: The complete method in litellm_provider.py uses dict[str,
object] for messages and tools which mismatches the abstract interface; update
the type hints to dict[str, Any] by importing Any from typing (e.g., add "from
typing import Any") and replace both occurrences of dict[str, object] in the
async def complete signature with dict[str, Any] so the method aligns with
ChatCompletionResult's base interface and matches OpenAIChatModel.
In `@vektori/models/openai.py`:
- Around line 105-112: The complete method in openai.py uses dict[str, object]
for message/tool types which differs from the abstract
ChatModelProvider.complete() signature; update the type hints in
vektori.models.openai.openAI.complete (messages and tools) to use dict[str, Any]
to match the base class, and add Any to the typing imports in that module.
🪄 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: defaults
Review profile: CHILL
Plan: Pro
Run ID: 8fc61fa8-804e-4b15-adc4-e0774834c166
📒 Files selected for processing (22)
README.mdexamples/openai_agent.pyexamples/vektori_agent_demo.pypyproject.tomltests/integration/test_agent_chat_flow.pytests/unit/test_agent_context.pytests/unit/test_cli_agent.pytests/unit/test_message_window.pytests/unit/test_profile_store.pytests/unit/test_prompt_builder.pyvektori/__init__.pyvektori/agent.pyvektori/cli.pyvektori/context.pyvektori/memory/__init__.pyvektori/memory/profile.pyvektori/memory/window.pyvektori/models/base.pyvektori/models/factory.pyvektori/models/litellm_provider.pyvektori/models/openai.pyvektori/prompts.py
✅ Files skipped from review due to trivial changes (2)
- examples/vektori_agent_demo.py
- tests/unit/test_profile_store.py
🚧 Files skipped from review as they are similar to previous changes (2)
- pyproject.toml
- vektori/init.py
Comment on lines
+5
to
+18
| import asyncio | ||
| from dataclasses import dataclass, field | ||
| from pathlib import Path | ||
| import re | ||
| from typing import Any | ||
| from uuid import uuid4 | ||
|
|
||
| from vektori.client import Vektori | ||
| from vektori.context import AgentContextLoader | ||
| from vektori.memory.profile import InMemoryProfileStore, ProfilePatch, ProfileStore, SQLiteProfileStore | ||
| from vektori.memory.window import MessageWindow | ||
| from vektori.models.base import ChatModelProvider | ||
| from vektori.prompts import build_messages | ||
| from vektori.retrieval.gate import should_retrieve |
There was a problem hiding this comment.
Fix the import block so CI passes.
Ruff I001 is failing here as well.
🛠️ Suggested fix
import asyncio
+import re
from dataclasses import dataclass, field
from pathlib import Path
-import re
from typing import Any
from uuid import uuid4
from vektori.client import Vektori
from vektori.context import AgentContextLoader
-from vektori.memory.profile import InMemoryProfileStore, ProfilePatch, ProfileStore, SQLiteProfileStore
+from vektori.memory.profile import (
+ InMemoryProfileStore,
+ ProfilePatch,
+ ProfileStore,
+ SQLiteProfileStore,
+)
from vektori.memory.window import MessageWindow
from vektori.models.base import ChatModelProvider
from vektori.prompts import build_messages📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| import asyncio | |
| from dataclasses import dataclass, field | |
| from pathlib import Path | |
| import re | |
| from typing import Any | |
| from uuid import uuid4 | |
| from vektori.client import Vektori | |
| from vektori.context import AgentContextLoader | |
| from vektori.memory.profile import InMemoryProfileStore, ProfilePatch, ProfileStore, SQLiteProfileStore | |
| from vektori.memory.window import MessageWindow | |
| from vektori.models.base import ChatModelProvider | |
| from vektori.prompts import build_messages | |
| from vektori.retrieval.gate import should_retrieve | |
| import asyncio | |
| import re | |
| from dataclasses import dataclass, field | |
| from pathlib import Path | |
| from typing import Any | |
| from uuid import uuid4 | |
| from vektori.client import Vektori | |
| from vektori.context import AgentContextLoader | |
| from vektori.memory.profile import ( | |
| InMemoryProfileStore, | |
| ProfilePatch, | |
| ProfileStore, | |
| SQLiteProfileStore, | |
| ) | |
| from vektori.memory.window import MessageWindow | |
| from vektori.models.base import ChatModelProvider | |
| from vektori.prompts import build_messages | |
| from vektori.retrieval.gate import should_retrieve |
🧰 Tools
🪛 GitHub Actions: CI
[error] 3-18: ruff check: I001 Import block is un-sorted or un-formatted. help: Organize imports
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/agent.py` around lines 5 - 18, Ruff I001 flags the import block
ordering; reorder and group imports into standard library, third-party, and
local modules and sort them alphabetically within each group so the import list
in vektori/agent.py complies with Ruff: group asyncio, dataclass/dataclasses,
Path/pathlib, re, typing.Any, uuid4 into the stdlib block; keep third-party
(none here) separate; then list local project imports (Vektori,
AgentContextLoader, InMemoryProfileStore, ProfilePatch, ProfileStore,
SQLiteProfileStore, MessageWindow, ChatModelProvider, build_messages,
should_retrieve) alphabetically and remove any unused imports if present so
symbols like asyncio, dataclass, Path, re, Any, uuid4, Vektori,
AgentContextLoader, InMemoryProfileStore, ProfilePatch, ProfileStore,
SQLiteProfileStore, MessageWindow, ChatModelProvider, build_messages, and
should_retrieve are correctly grouped and ordered.
| compaction_trigger_ratio: float = 0.8 | ||
| keep_last_n_turns: int = 6 | ||
| summary_max_tokens: int = 400 | ||
| persist_assistant_messages: bool = True |
There was a problem hiding this comment.
Honor persist_assistant_messages.
exchange always includes the assistant reply, so this config flag never changes behavior.
🛠️ Suggested fix
- exchange = [
- {"role": "user", "content": user_message},
- {"role": "assistant", "content": assistant_content},
- ]
+ exchange = [{"role": "user", "content": user_message}]
+ if self.config.persist_assistant_messages:
+ exchange.append({"role": "assistant", "content": assistant_content})Also applies to: 124-145
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/agent.py` at line 35, The persist_assistant_messages flag is ignored
because exchange always receives the assistant reply; update the message
handling so that when persist_assistant_messages is False you do not include
assistant role messages in the payload sent to exchange and you do not write
them to persistence. Concretely, modify the code paths that build the messages
list for the exchange call (the exchange function) to filter out messages with
role "assistant" when persist_assistant_messages is False, and add the same
check before any persistence/save routine (e.g., any
persist_messages/save_messages helper) so assistant replies are not stored.
Comment on lines
+88
to
+121
| self.window.add("user", user_message) | ||
|
|
||
| memories: dict[str, list[dict[str, Any]]] = {"facts": [], "episodes": [], "sentences": []} | ||
| if await self._should_retrieve(user_message): | ||
| memories = await self.memory.search( | ||
| query=user_message, | ||
| user_id=self.user_id, | ||
| agent_id=self.agent_id, | ||
| depth=self.config.retrieval_depth, | ||
| top_k=self.config.retrieval_top_k, | ||
| ) | ||
|
|
||
| context = self.context_loader.load() | ||
| profile_patches = await self.profile_store.list_active( | ||
| observer_id=self.agent_id or "default-agent", | ||
| observed_id=self.user_id, | ||
| ) | ||
| messages = build_messages( | ||
| context=context, | ||
| profile_patches=profile_patches, | ||
| memories=memories, | ||
| window_state=self.window.snapshot(), | ||
| runtime_overrides=self.config.runtime_overrides, | ||
| ) | ||
|
|
||
| completion = await self.model.complete( | ||
| messages, | ||
| max_tokens=self.config.reserve_response_tokens, | ||
| temperature=0.2, | ||
| ) | ||
| assistant_content = completion.content or "" | ||
| self.window.add("assistant", assistant_content) | ||
|
|
||
| summary_updated = await self.window.compact(self.model) |
There was a problem hiding this comment.
Stage the user turn until the model call succeeds.
The live window is mutated before any awaited work. If retrieval or completion fails, the next retry will build on a dangling user-only turn.
🛠️ Suggested fix
- self.window.add("user", user_message)
+ window_state = self.window.snapshot()
+ window_state.recent_messages.append({"role": "user", "content": user_message})
memories: dict[str, list[dict[str, Any]]] = {"facts": [], "episodes": [], "sentences": []}
if await self._should_retrieve(user_message):
@@
messages = build_messages(
context=context,
profile_patches=profile_patches,
memories=memories,
- window_state=self.window.snapshot(),
+ window_state=window_state,
runtime_overrides=self.config.runtime_overrides,
)
@@
completion = await self.model.complete(
messages,
max_tokens=self.config.reserve_response_tokens,
temperature=0.2,
)
assistant_content = completion.content or ""
+ self.window.add("user", user_message)
self.window.add("assistant", assistant_content)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/agent.py` around lines 88 - 121, The code mutates self.window before
any awaited work (calls like _should_retrieve, memory.search,
context_loader.load, profile_store.list_active, build_messages, model.complete)
so failures leave a dangling user-only turn; instead, keep the incoming
user_message in a local variable and only call self.window.add("user",
user_message) (and later self.window.add("assistant", assistant_content) and
await self.window.compact(self.model)) after model.complete returns
successfully; if model.complete raises, do not modify self.window and let the
caller retry, and ensure the retrieval logic (_should_retrieve / memory.search)
and message construction use the local user_message rather than mutating the
window early.
Comment on lines
+27
to
+33
| def load(self) -> LoadedAgentContext: | ||
| path = self._resolve_path() | ||
| if path is None: | ||
| return LoadedAgentContext() | ||
| if path.suffix in {".yaml", ".yml"}: | ||
| return self._load_yaml(path) | ||
| return self._load_markdown(path) |
There was a problem hiding this comment.
Fail fast when an explicit context path is invalid.
If context_path is provided but missing, load() currently returns an empty context, which hides config mistakes and makes behavior non-obvious.
💡 Suggested fix
def load(self) -> LoadedAgentContext:
path = self._resolve_path()
if path is None:
+ if self.context_path is not None:
+ raise FileNotFoundError(f"Context file not found: {self.context_path}")
return LoadedAgentContext()
if path.suffix in {".yaml", ".yml"}:
return self._load_yaml(path)
return self._load_markdown(path)Also applies to: 35-70
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/context.py` around lines 27 - 33, The load() method currently returns
an empty LoadedAgentContext when _resolve_path() returns None, hiding errors
when an explicit context_path was provided; change load() to detect when a
user-supplied context path is invalid by inspecting the object's context_path
(or equivalent constructor arg) and, if present but _resolve_path() returned
None, raise a clear exception (e.g. FileNotFoundError or ValueError) instead of
returning an empty LoadedAgentContext; keep the existing behavior only for the
case where no explicit context_path was given, and apply the same change to the
other context-loading variants referenced in the file (the branches that call
_load_yaml and _load_markdown) so invalid explicit paths fail fast.
Comment on lines
+72
to
+85
| def _load_yaml(self, path: Path) -> LoadedAgentContext: | ||
| raw = yaml.safe_load(path.read_text()) or {} | ||
| agent = raw.get("agent", {}) | ||
| return LoadedAgentContext( | ||
| persona=agent.get("persona", "") or "", | ||
| instructions=[str(item) for item in (agent.get("instructions", []) or [])], | ||
| response_style=agent.get("response_style", {}) or {}, | ||
| memory_policy=agent.get("memory", {}) or {}, | ||
| extra_sections={ | ||
| key: value | ||
| for key, value in agent.items() | ||
| if key not in {"name", "persona", "instructions", "response_style", "memory"} | ||
| }, | ||
| ) |
There was a problem hiding this comment.
Guard YAML structure before mapping access.
yaml.safe_load() can return non-dicts; agent can also be non-dict. Current .get calls can raise at runtime on malformed but syntactically valid YAML.
💡 Suggested fix
def _load_yaml(self, path: Path) -> LoadedAgentContext:
- raw = yaml.safe_load(path.read_text()) or {}
- agent = raw.get("agent", {})
+ raw = yaml.safe_load(path.read_text()) or {}
+ if not isinstance(raw, dict):
+ raw = {}
+ agent_obj = raw.get("agent", {})
+ agent = agent_obj if isinstance(agent_obj, dict) else {}
+
+ response_style = agent.get("response_style", {}) or {}
+ if not isinstance(response_style, dict):
+ response_style = {}
+ memory_policy = agent.get("memory", {}) or {}
+ if not isinstance(memory_policy, dict):
+ memory_policy = {}
+
return LoadedAgentContext(
persona=agent.get("persona", "") or "",
instructions=[str(item) for item in (agent.get("instructions", []) or [])],
- response_style=agent.get("response_style", {}) or {},
- memory_policy=agent.get("memory", {}) or {},
+ response_style=response_style,
+ memory_policy=memory_policy,
extra_sections={
- key: value
+ key: str(value)
for key, value in agent.items()
if key not in {"name", "persona", "instructions", "response_style", "memory"}
},
)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/context.py` around lines 72 - 85, The _load_yaml function should
guard against yaml.safe_load returning non-dict values and ensure `agent` is a
dict before using .get or .items; update _load_yaml to treat `raw =
yaml.safe_load(...) or {}` as before but coerce/validate `agent =
raw.get("agent", {})` to `agent = agent if isinstance(agent, dict) else {}` (or
use a dict(...) conversion) so all subsequent accesses for persona,
instructions, response_style, memory and the extra_sections comprehension over
agent.items() are safe; ensure the return constructs for LoadedAgentContext
still default to empty values when keys are missing or when agent was not a
dict.
Comment on lines
+5
to
+12
| import json | ||
| import sqlite3 | ||
| from dataclasses import dataclass, field | ||
| from datetime import datetime, timezone | ||
| from pathlib import Path | ||
| from typing import Protocol | ||
| from typing import Any | ||
|
|
There was a problem hiding this comment.
Fix the import block so CI passes.
Ruff is already failing on I001 here.
🛠️ Suggested fix
import json
import sqlite3
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
-from typing import Protocol
-from typing import Any
+from typing import Any, Protocol📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| import json | |
| import sqlite3 | |
| from dataclasses import dataclass, field | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| from typing import Protocol | |
| from typing import Any | |
| import json | |
| import sqlite3 | |
| from dataclasses import dataclass, field | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| from typing import Any, Protocol |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/memory/profile.py` around lines 5 - 12, The import block is
misordered and duplicates typing imports; reorder and consolidate imports to
satisfy Ruff I001: group standard-library imports alphabetically and combine
typing names into one line. Specifically, keep standard-library imports only,
sort them (for example: from dataclasses import dataclass, field; from datetime
import datetime, timezone; import json; from pathlib import Path; import
sqlite3; from typing import Any, Protocol), and remove the separate "from typing
import Protocol" line so "Any" and "Protocol" are imported together.
Comment on lines
+53
to
+55
| async def save(self, patch: ProfilePatch) -> None: | ||
| key = (patch.observer_id, patch.observed_id) | ||
| self._patches.setdefault(key, []).append(patch) |
There was a problem hiding this comment.
Supersede the previous active patch for the same key.
Both stores only append. A second preferences.name, preferences.units, or response_style.verbosity update leaves multiple active values for one singular key, and list_active() will return all of them.
🛠️ Suggested fix
async def save(self, patch: ProfilePatch) -> None:
key = (patch.observer_id, patch.observed_id)
+ for existing in self._patches.get(key, []):
+ if existing.key == patch.key and existing.active:
+ existing.active = False
self._patches.setdefault(key, []).append(patch) async def save(self, patch: ProfilePatch) -> None:
await self._ensure_initialized()
assert self._connection is not None
+ self._connection.execute(
+ """
+ UPDATE profile_patches
+ SET active = 0
+ WHERE observer_id = ? AND observed_id = ? AND key = ? AND active = 1
+ """,
+ (patch.observer_id, patch.observed_id, patch.key),
+ )
self._connection.execute(
"""
INSERT INTO profile_patches (Also applies to: 86-109
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/memory/profile.py` around lines 53 - 55, The current save
implementation appends every ProfilePatch to self._patches[key], causing
multiple active patches for the same (observer_id, observed_id) and making
list_active() return stale duplicates; change save(self, patch: ProfilePatch) to
supersede prior active patches for that key by replacing the stored entry
instead of appending (i.e., set self._patches[key] = [patch] or otherwise
remove/mark previous patches as inactive), and apply the same replacement logic
to the other save implementation referenced in the comment (the save block
around lines 86-109) so list_active() only returns the single most recent active
patch per key.
Comment on lines
+71
to
+90
| prompt = ( | ||
| "Summarize the conversation state in this format:\n" | ||
| "Conversation Summary\n" | ||
| "- Active goals:\n" | ||
| "- Confirmed preferences:\n" | ||
| "- Open questions:\n" | ||
| "- Constraints:\n" | ||
| "- Recent commitments:\n\n" | ||
| f"Conversation:\n{old_text}" | ||
| ) | ||
| result = await summarizer.complete( | ||
| [{"role": "user", "content": prompt}], | ||
| max_tokens=self.summary_max_tokens, | ||
| temperature=0.0, | ||
| ) | ||
| if result.content: | ||
| self._rolling_summary = result.content.strip() | ||
| self._recent_messages = kept_messages | ||
| self._compaction_count += 1 | ||
| return True |
There was a problem hiding this comment.
Compaction can silently lose memory state across turns.
compact() replaces the rolling summary using only the latest removed chunk and still trims history even when summary text is empty. This can forget earlier context and drop data on weak/empty summarizer outputs.
💡 Suggested fix
- prompt = (
- "Summarize the conversation state in this format:\n"
+ existing_summary = self._rolling_summary.strip()
+ prompt = (
+ "Update the rolling conversation summary in this format:\n"
"Conversation Summary\n"
"- Active goals:\n"
"- Confirmed preferences:\n"
"- Open questions:\n"
"- Constraints:\n"
"- Recent commitments:\n\n"
- f"Conversation:\n{old_text}"
+ f"Existing summary:\n{existing_summary or '(none)'}\n\n"
+ f"New conversation:\n{old_text}"
)
- result = await summarizer.complete(
- [{"role": "user", "content": prompt}],
- max_tokens=self.summary_max_tokens,
- temperature=0.0,
- )
- if result.content:
- self._rolling_summary = result.content.strip()
+ try:
+ result = await summarizer.complete(
+ [{"role": "user", "content": prompt}],
+ max_tokens=self.summary_max_tokens,
+ temperature=0.0,
+ )
+ except Exception:
+ return False
+
+ new_summary = (result.content or "").strip()
+ if not new_summary:
+ return False
+
+ self._rolling_summary = new_summary
self._recent_messages = kept_messages
self._compaction_count += 1
return True🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/memory/window.py` around lines 71 - 90, The compaction logic in
compact() can overwrite prior context and drop messages when the summarizer
returns empty/weak output; change compact() so it only updates
self._rolling_summary when result.content is non-empty (and instead appends or
merges new summary into the existing self._rolling_summary rather than replacing
it), avoid trimming/losing self._recent_messages if the summary is empty (i.e.,
abort the removal/keep original kept_messages), and only increment
self._compaction_count on a successful non-empty summary; use the existing
symbols summarizer.complete, self.summary_max_tokens, result.content,
self._rolling_summary, self._recent_messages and self._compaction_count to
locate and implement these checks/merge behavior.
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
1 participant
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.
Summary
vektori/integrations/pipecat/— two drop-in PipecatFrameProcessorsubclasses that give any voice pipeline Vektori long-term memoryexamples/pipecat_voice_agent.py— a complete, runnable FastAPI WebSocket server (Deepgram → GPT-4o-mini → ElevenLabs)pipecatoptional dependency group inpyproject.tomlProcessors
VektoriMemoryProcessor(place before LLM)OpenAILLMContextFrameeach turnvektori.search()on the latest user utterance (depth="l1",top_k=5— voice-tuned defaults)[Memory context]block into the LLM system prompt before inferenceVektoriStorageProcessor(place after LLM)TranscriptionFrame+ streamedTextFramechunksvektori.add()onLLMFullResponseEndFrameOpenAILLMContextobject if frame buffers are emptyUsage (10 lines)
Test plan
pip install "vektori[pipecat]"installs cleanlyVektoriMemoryProcessorinjects memory into system prompt on second turnVektoriStorageProcessorstores conversation;vektori.search()returns it on the next turnvektori.close()without errorsImportErrorfromvektori.integrations.pipecat🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Tests