Feat/native agent harness#55
Merged
Merged
Conversation
|
Warning Rate limit exceeded
To keep reviews running without waiting, you can enable usage-based add-on for your organization. This allows additional reviews beyond the hourly cap. Account admins can enable it under billing. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: ⛔ Files ignored due to path filters (1)
📒 Files selected for processing (15)
📝 WalkthroughWalkthroughThis PR introduces a comprehensive agent harness framework for Vektori, including a stateful Changes
Sequence Diagram(s)sequenceDiagram
participant User as User/Caller
participant Agent as VektoriAgent
participant Memory as Vektori Memory
participant Context as ContextLoader
participant Profile as ProfileStore
participant Window as MessageWindow
participant Prompt as PromptBuilder
participant Chat as ChatModelProvider
participant Tools as Tool Executor
User->>Agent: chat(user_message)
Agent->>Window: add(role='user', content)
Agent->>Memory: search() [if gate allows]
Memory-->>Agent: facts, episodes, sentences
Agent->>Context: load()
Context-->>Agent: LoadedAgentContext
Agent->>Profile: list_active(observer_id, observed_id)
Profile-->>Agent: active ProfilePatches
Agent->>Prompt: build_prompt_result(context, patches, memories, window)
Prompt-->>Agent: messages, memories_used, debug
Agent->>Chat: complete(messages, tools?)
Chat-->>Agent: ChatCompletionResult
alt Tool Calls Present
Agent->>Tools: handle_tool_call()
Tools->>Memory: search() / update_profile()
Tools-->>Agent: result
Agent->>Chat: complete() [tool response loop]
Chat-->>Agent: ChatCompletionResult (final)
end
Agent->>Window: add(role='assistant', content)
Agent->>Window: compact() [if needed]
Agent->>Agent: extract_profile_patches(user_message)
Agent->>Profile: save(patch)
Agent->>Memory: add(user+assistant) [sync or background]
Agent-->>User: AgentTurnResult
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 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. Review rate limit: 0/1 reviews remaining, refill in 44 minutes and 51 seconds.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: 15
🧹 Nitpick comments (16)
README.md (1)
153-162: Usetry/finallyin the README loop to guarantee cleanup.In the snippet, failures during chat/input can bypass cleanup on Lines 161-162. The docs should mirror the safe pattern used in
examples/openai_agent.py.♻️ Proposed fix for the README snippet
- print("Chat with memory (type 'quit' to exit)\n") - while True: - user_input = input("You: ").strip() - if user_input.lower() == "quit": - break - result = await agent.chat(user_input) - print(f"Assistant: {result.content}\n") - - await agent.close() - await memory.close() + print("Chat with memory (type 'quit' to exit)\n") + try: + while True: + user_input = input("You: ").strip() + if user_input.lower() == "quit": + break + result = await agent.chat(user_input) + print(f"Assistant: {result.content}\n") + finally: + await agent.close() + await memory.close()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@README.md` around lines 153 - 162, The README's interactive loop can skip cleanup if an exception occurs; wrap the chat loop that calls agent.chat(user_input) in a try/finally so that await agent.close() and await memory.close() always run; specifically, put the while True input/chat sequence (and any awaiting of agent.chat) inside a try block and call await agent.close() and await memory.close() in the finally block to mirror the safe pattern used in examples/openai_agent.py and ensure proper async cleanup.examples/vektori_agent_demo.py (1)
23-26: Ensure resources are always closed in the demo flow.If
agent.chat(...)fails on Line 23, cleanup on Lines 25-26 is skipped. Wrap the turn intry/finallyso the demo doesn’t leak connections.♻️ Proposed fix
- result = await agent.chat("What do you remember about how I like answers?") - print(result.content) - await agent.close() - await memory.close() + try: + result = await agent.chat("What do you remember about how I like answers?") + print(result.content) + finally: + await agent.close() + await memory.close()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@examples/vektori_agent_demo.py` around lines 23 - 26, The demo currently calls agent.chat(...) and then closes resources, but if agent.chat raises an exception the cleanup (agent.close() and memory.close()) is skipped; wrap the interaction in a try/finally so that agent.close() and memory.close() are always executed (place the result = await agent.chat(...) and print(...) inside the try block and call await agent.close() and await memory.close() in the finally block), referencing the agent.chat, agent.close, and memory.close symbols to locate the change.vektori/models/litellm_provider.py (1)
67-74: Type hints differ from base class contract.The method signature uses
list[dict[str, object]]formessagesandtools, but the abstractChatModelProviderbase class (invektori/models/base.py:70-90) defines these aslist[dict[str, Any]]. While functionally equivalent at runtime, aligning with the base class improves consistency and IDE tooling.Suggested fix
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 need to add
Anyto the import:-from vektori.models.base import ChatCompletionResult, ChatModelProvider, LLMProvider +from typing import Any +from vektori.models.base import ChatCompletionResult, ChatModelProvider, LLMProvider🤖 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's type hints (messages and tools) in LitellmProvider use list[dict[str, object]] which mismatch the ChatModelProvider base signature that uses list[dict[str, Any]]; update the complete method signature to use Any for those dict value types, add Any to the module imports, and ensure the method signature in class LitellmProvider matches the base ChatModelProvider exactly so IDE/type checkers see the override correctly.vektori/models/openai.py (2)
105-112: Type hints differ from base class contract.Same issue as
LiteLLMChatModel: the signature useslist[dict[str, object]]but the abstract base class useslist[dict[str, Any]].Suggested fix
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 to 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/openai.py` around lines 105 - 112, The method signature of openai.ChatModel.complete uses list[dict[str, object]] which mismatches the abstract base contract that expects list[dict[str, Any]]; update the signature in the complete method (and any similar signatures such as in LiteLLMChatModel) to use typing.Any for message dict values, add Any to the imports, and ensure type annotations across the class match the base class contract (i.e., messages: list[dict[str, Any]]) so the method properly implements the base class.
98-103: MissingImportErrorhandling for consistency.
OpenAIEmbedder._get_client(lines 34-41) wraps theopenaiimport in atry/exceptand raises a helpful error message.OpenAIChatModel._get_clientdoesn't follow this pattern, leading to inconsistent behavior when the package is missing.Add ImportError handling
def _get_client(self): if self._client is None: - from openai import AsyncOpenAI - - self._client = AsyncOpenAI(api_key=self._api_key) + try: + from openai import AsyncOpenAI + except ImportError as e: + raise ImportError("openai package required: pip install openai") from e + self._client = AsyncOpenAI(api_key=self._api_key) return self._client🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@vektori/models/openai.py` around lines 98 - 103, OpenAIChatModel._get_client currently imports AsyncOpenAI without ImportError handling; update this method to mirror OpenAIEmbedder._get_client by wrapping "from openai import AsyncOpenAI" in a try/except ImportError and raise a clear, actionable ImportError that mentions the missing "openai" package and how to install it (include the original exception or message for debugging), then proceed to instantiate self._client = AsyncOpenAI(api_key=self._api_key) as before.examples/agent_type_extraction.py (2)
38-41: Unused import:json.The
jsonmodule is imported but never used in this file.Remove unused import
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` around lines 38 - 41, The file imports the json module but never uses it; remove the unused import statement ("import json") from examples/agent_type_extraction.py so only needed imports (asyncio and from vektori import ExtractionConfig, Vektori) remain, ensuring no other references to json exist in functions like any extraction setup or main coroutine.
140-142: Fixed sleep may be fragile for varying network/load conditions.The 3-second
asyncio.sleepassumes extraction completes within that window. For a demo this is acceptable, but consider documenting that results may be incomplete if extraction takes longer, or using a more robust wait mechanism if this example is expected to be reliable.🤖 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 fixed 3-second asyncio.sleep is fragile because it may not wait long enough for async extraction; replace the hard sleep with a polling/wait loop that repeatedly calls presales_v.search (or a dedicated completion API if available) until expected results are present or a timeout is reached, e.g., poll presales_v.search("budget and decision makers", user_id="demo-user") with short await asyncio.sleep intervals and a max timeout, and document the timeout behavior so the demo handles variable network/load conditions gracefully.examples/pipecat_voice_agent.py (1)
136-141: CORS allows all origins — acceptable for example, not for production.The
allow_origins=["*"]configuration is fine for local development and examples, but should be restricted in production deployments.🤖 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 136 - 141, The current CORSMiddleware setup in the app.add_middleware call uses allow_origins=["*"], which is fine for examples but unsafe in production; update the code that configures CORSMiddleware (the app.add_middleware(..., CORSMiddleware, allow_origins=... ) call) to read allowed origins from configuration or an environment variable (e.g., ALLOWED_ORIGINS or a settings object) and only use ["*"] when explicitly running in a local/dev mode, otherwise pass a restricted list of origins or raise an error if none are configured.vektori/context.py (1)
35-70: Path resolution logic is correct but verbose.The method works correctly. Consider simplifying the candidate list construction:
Optional simplification
def _resolve_path(self) -> Path | None: candidates: list[Path] = [] if self.context_path is not None: candidates.append(Path(self.context_path)) else: cwd = Path.cwd() - candidates.extend( - [ - cwd / "agents.md", - cwd / "vektori.yaml", - cwd / "vektori.yml", - ] - ) - candidates.extend( - [ - path / "agents.md" - for path in [*cwd.parents][:3] - ] - ) - candidates.extend( - [ - path / "vektori.yaml" - for path in [*cwd.parents][:3] - ] - ) - candidates.extend( - [ - path / "vektori.yml" - for path in [*cwd.parents][:3] - ] - ) + search_dirs = [cwd, *cwd.parents[:3]] + filenames = ["agents.md", "vektori.yaml", "vektori.yml"] + candidates = [d / f for d in search_dirs for f in filenames] for candidate in candidates: if candidate.exists(): return candidate return None🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@vektori/context.py` around lines 35 - 70, The _resolve_path method builds the same candidate paths in a verbose, repeated way; simplify by consolidating the filenames list (e.g., ["agents.md","vektori.yaml","vektori.yml"]) and the search roots (cwd plus first three cwd.parents) then generate candidates via a nested loop/comprehension combining each root with each filename (while still honoring an explicit self.context_path if provided), keep the existing behavior of returning the first existing candidate or None and reference the function name _resolve_path, the candidates variable, and context_path when making the change.vektori/prompts.py (2)
116-117: Repeated prompt assembly in budget check loop is inefficient.
_within_budget()calls_assemble()on every iteration of the trim loop (lines 147-159). For conversations with many messages or memory items, this rebuilds the entire message list repeatedly. Consider caching the assembled result or computing token deltas incrementally.💡 Optional: Cache token estimate during trimming
+ # Pre-compute initial estimate + current_tokens = estimate_tokens_for_messages(_assemble()) + def _within_budget() -> bool: - return estimate_tokens_for_messages(_assemble()) <= budget + nonlocal current_tokens + return current_tokens <= budget + + def _recompute_tokens() -> None: + nonlocal current_tokens + current_tokens = estimate_tokens_for_messages(_assemble())Then call
_recompute_tokens()after each trim operation.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@vektori/prompts.py` around lines 116 - 117, The _within_budget function repeatedly calls _assemble() causing expensive full message rebuilds during the trim loop; change the logic to cache the assembled messages and their token estimate once (use a local variable for assembled = _assemble() and tokens = estimate_tokens_for_messages(assembled)), then have _within_budget reference that cached tokens (or better, replace _within_budget usage with a direct check against the cached tokens), and after each trim operation update the cached tokens by calling a small helper _recompute_tokens() (or decrement tokens by the known token delta of the removed item) rather than re-calling _assemble(); update references to _within_budget, _assemble, estimate_tokens_for_messages, budget and add or use _recompute_tokens accordingly.
34-51: LoadedAgentContext fieldsresponse_style,memory_policy, andextra_sectionsare not used.Per
vektori/context.py:12-18,LoadedAgentContextdefinesresponse_style,memory_policy, andextra_sectionsfields, but_render_system_promptonly usespersonaandinstructions. If these fields are intended for future use, consider adding a TODO comment. Otherwise, they could be integrated here.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@vektori/prompts.py` around lines 34 - 51, _render_system_prompt currently ignores LoadedAgentContext fields response_style, memory_policy, and extra_sections; update the function to include these fields in the generated system prompt (e.g., add sections like "Response style:" showing context.response_style, "Memory policy:" showing context.memory_policy, and render each item in context.extra_sections) so the prompt reflects all context data, or if they are intentionally unused add a short TODO comment inside _render_system_prompt referencing response_style, memory_policy, and extra_sections to clarify intended future use.vektori/integrations/pipecat/processor.py (2)
329-331: Fire-and-forget task may cause issues on shutdown.Using
asyncio.ensure_future()creates an untracked background task. If the pipeline shuts down before_store_turn()completes, the turn may be lost. Consider tracking these tasks for graceful shutdown, similar to howVektoriAgenttracks_background_tasks.💡 Optional: Track background tasks for graceful shutdown
def __init__( self, vektori: "Vektori", user_id: str, session_id: str, context: OpenAILLMContext | None = None, ) -> None: super().__init__() self._vektori = vektori self._user_id = user_id self._session_id = session_id self._context = context self._pending_user: str = "" self._pending_assistant_parts: list[str] = [] + self._background_tasks: set[asyncio.Task] = set() + + def _schedule_background(self, coro) -> None: + task = asyncio.create_task(coro) + self._background_tasks.add(task) + task.add_done_callback(self._background_tasks.discard)Then in
process_frame:elif isinstance(frame, LLMFullResponseEndFrame): # LLM finished — store the completed turn asynchronously - asyncio.ensure_future(self._store_turn()) + self._schedule_background(self._store_turn())🤖 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 329 - 331, The current use of asyncio.ensure_future(self._store_turn()) in the LLMFullResponseEndFrame branch creates an untracked fire-and-forget task that can be lost on shutdown; replace this with a tracked background task pattern: create the task via asyncio.create_task(self._store_turn()), append it to a per-instance task list (e.g., self._background_tasks) and ensure the processor exposes shutdown/cleanup logic that cancels and awaits tasks (similar to VektoriAgent's _background_tasks handling) so _store_turn() is awaited or cancelled cleanly during shutdown.
319-322: TranscriptionFrame handling may overwrite previous transcription.If multiple
TranscriptionFrameevents arrive before anLLMFullResponseEndFrame, only the last transcription is retained. This might be intentional for "final" transcriptions, but consider whether you need to accumulate partial transcriptions.🤖 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 319 - 322, The current handling in the block checking FrameDirection.DOWNSTREAM and isinstance(frame, TranscriptionFrame) assigns self._pending_user = frame.text or "" which discards earlier partial transcriptions; change this to accumulate partials or only replace on final markers depending on desired behavior. Update the TranscriptionFrame handling in the processor to either append non-final transcription chunks to self._pending_user (e.g., concatenate with a separator and guard against duplicates) or check a "is_final" flag on TranscriptionFrame (or the arrival of LLMFullResponseEndFrame) and only set/replace self._pending_user when the frame is final; ensure references to FrameDirection.DOWNSTREAM, TranscriptionFrame, self._pending_user and LLMFullResponseEndFrame are used so the logic correctly preserves and finalizes the transcript.vektori/agent.py (1)
193-196:close()should handle case whereprofile_storelacksclose()method.If a custom
ProfileStoreimplementation doesn't have aclose()method, this will raise anAttributeError. Consider checking if the method exists or using a try/except.💡 Optional: Defensive close handling
async def close(self) -> None: if self._background_tasks: await asyncio.gather(*self._background_tasks, return_exceptions=True) - await self.profile_store.close() + if hasattr(self.profile_store, 'close'): + await self.profile_store.close()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@vektori/agent.py` around lines 193 - 196, The close() method may raise AttributeError if self.profile_store lacks a close() method; update Agent.close to defensively call or await profile_store.close only when available — e.g., check hasattr(self.profile_store, "close") or wrap the call in try/except Exception to swallow AttributeError (and optionally log it); ensure the existing await asyncio.gather(*self._background_tasks, return_exceptions=True) logic remains unchanged and that any async close is awaited correctly when present.vektori/memory/window.py (1)
57-90: Compaction logic looks correct with one minor observation.The compaction flow properly:
- Checks token threshold before proceeding
- Preserves last N turns (multiplied by 2 for user+assistant pairs)
- Builds a structured summary prompt
- Updates state atomically after successful summarization
One consideration: if
summarizer.complete()returnscontent=None, the_rolling_summaryis not updated but messages are still truncated on line 88. This means old context is lost without replacement. Consider whether this is intentional behavior or if you should preserve messages when summarization fails.💡 Optional: Preserve messages if summarization returns no content
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 + if not result.content: + # Summarization failed - don't discard messages + return False + self._rolling_summary = result.content.strip() + 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 57 - 90, The compaction currently discards old messages even when summarizer.complete(...) returns no content; update the compact method so it only replaces _recent_messages, updates _rolling_summary, and increments _compaction_count when result.content is truthy; if result.content is falsy, leave _recent_messages and _rolling_summary intact and return False (or propagate a failure), ensuring the summarizer.complete call is checked before truncating state in compact.vektori/cli.py (1)
767-769:_slug_to_pathconversion logic may not accurately reconstruct paths.The function replaces only the first hyphen with
/and all others with spaces. This seems fragile—if the original path was/home/user/my-project, the slughome-user-my-projectwould becomehome/user my project(with spaces). Consider whether this is the intended behavior or if it should preserve hyphens after the first/.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@vektori/cli.py` around lines 767 - 769, _slug_to_path currently replaces the first hyphen with "/" then turns all remaining hyphens into spaces, which mangles legitimate hyphenated path segments; update the function (_slug_to_path) to only convert the first hyphen into a path separator and leave subsequent hyphens unchanged (i.e., remove the second replace that turns "-" into " "), then strip and return the result so a slug like "home-user-my-project" becomes "home/user-my-project".
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@demo.sh`:
- Around line 45-47: The demo hardcodes the user flag as "dev" so the exported
VEKTORI_USER_ID variable is ignored; update the script to use the
VEKTORI_USER_ID environment variable everywhere the user/namespace is passed
(replace any literal "-u dev" or similar with "-u ${VEKTORI_USER_ID}" or the
shell variable usage pattern used elsewhere), including all command invocations
in the block around VEKTORI_USER_ID and the later repeated section (the area
covering the commands referenced in the comment), so callers who set
VEKTORI_USER_ID get their value applied.
In `@examples/pipecat_voice_agent.py`:
- Around line 248-258: The disconnect handler (on_disconnected) may not run due
to a Pipecat transport race, risking that external resource vektori isn't
closed; wrap the call to PipelineRunner().run(task) in a try/finally so
vektori.close() is always awaited on exit (safe because Vektori.close() is
idempotent), leaving the existing await vektori.close() in on_disconnected
as-is; locate PipelineRunner, runner.run(task), and vektori.close to implement
the finally cleanup.
In `@tests/unit/test_harness_spec_checklist.py`:
- Around line 5-7: The test currently uses
Path("docs/AGENT_HARNESS_CHECKLIST.md") which depends on the process CWD; change
it to resolve the docs path relative to the test file instead (use
Path(__file__).resolve().parent/.. or .parent.parent then /
"docs/AGENT_HARNESS_CHECKLIST.md") so the path lookup in the
test_harness_spec_checklist.py file does not break when pytest is run from an
IDE or other working directory; update the references to the
Path("docs/AGENT_HARNESS_CHECKLIST.md") variable (e.g., the path variable) to
this resolved path before calling exists() and read_text().
In `@vektori/cli.py`:
- Around line 1083-1085: The import of the datetime module is inside the loop
"for i, s in enumerate(all_sessions, 1):" which causes repeated imports; move
the import statement out of the loop (either to the top of the function or
module) and keep the existing usage at "ts =
datetime.datetime.fromtimestamp(s['mtime']).strftime(...)" unchanged so the loop
reuses the already-imported module.
- Around line 1213-1218: The exception handler currently both echoes
{"continue": True} when hook is truthy and then raises typer.Exit(1), making the
subsequent return unreachable; decide the intended behavior and implement it: if
hook mode should silently continue, remove the raise typer.Exit(1) (keep
typer.echo(json.dumps({"continue": True})) and return) in the exception block;
if hook mode should signal error and exit, remove the json echo and keep raising
typer.Exit(1). Update the exception handling around the hook variable in the
function in vektori/cli.py accordingly.
In `@vektori/config.py`:
- Around line 96-109: The config currently silently ignores typos in agent_type
because the extractor uses .get(..., ""), so add explicit validation in
__post_init__: define a shared constant set (e.g., ALLOWED_AGENT_TYPES)
containing the valid preset names and in __post_init__ check that
self.agent_type is in that set; if not, raise a ValueError with a clear message
listing allowed values. Ensure the same ALLOWED_AGENT_TYPES constant is
referenced by any preset lookup logic so validation and lookup stay in sync
(referencing agent_type, __post_init__, and the preset extractor logic).
In `@vektori/ingestion/extractor.py`:
- Around line 473-488: The _build_domain_guidance_episodes function currently
only appends the preset from _AGENT_EPISODES_GUIDANCE and
cfg.episodes_prompt_suffix, so it omits the ExtractionConfig hints cfg.focus_on
and cfg.ignore; update _build_domain_guidance_episodes to include these hints
(formatted consistently with the preset and suffix) when present by appending
cfg.focus_on and cfg.ignore (or a small, clear directive built from them) into
the parts list alongside the preset and episodes_prompt_suffix so episode
generation receives the same domain steering as fact extraction.
In `@vektori/ingestion/filter.py`:
- Line 101: The junk-filter regex
r"^\d[\d,\.]+\s*(views|likes|retweets|replies|reposts|followers|following)\s*$"
requires at least two numeric characters, so single-digit counts like "1 like"
bypass the filter; update that pattern (where it's defined in filter.py) to
allow a single digit by changing the quantifier after the first \d from + to *
(so the class after the initial digit can be empty), preserving the rest of the
anchors and the engagement-word group.
- Around line 105-106: The current pipe-delimited pattern
r"^[^|]{1,40}\|[^|]{1,40}\|" is too broad and matches normal table
headers/enumerations; update the filter entry that contains that regex so it
only flags true footer navs by (1) requiring known footer keywords (e.g., terms,
privacy, cookie, contact, about, legal) in a case-insensitive alternative OR (2)
enforcing stricter pipe formatting (no surrounding spaces around pipes and
anchoring the entire line) plus a minimum field count, so the pattern only
matches when either a footer keyword is present or the line is a tightly-packed,
multi-field nav line rather than a typical markdown table/header (replace the
existing r"^[^|]{1,40}\|[^|]{1,40}\|" pattern accordingly).
In `@vektori/integrations/pipecat/__init__.py`:
- Around line 12-19: The current broad except ImportError in
vektori.integrations.pipecat.__init__ masks real errors raised inside
processor.py; change the guard so we only translate failures that are truly due
to the external "pipecat" package being missing. Either (A) explicitly attempt
import pipecat before importing .processor and raise the friendly message only
when that import fails, or (B) inspect the caught
ImportError/ModuleNotFoundError (check exc.name or isinstance(exc,
ModuleNotFoundError) and exc.name == "pipecat") and re-raise the original
exception for any other import failure; keep the exported symbols
VektoriMemoryProcessor and VektoriStorageProcessor unchanged.
In `@vektori/memory/profile.py`:
- Around line 18-25: ProfilePatch.value is typed as Any but SQLiteProfileStore
serializes with json.dumps, causing backend-specific failures; narrow the type
to a JSON-safe alias (e.g., JSONValue = Union[dict, list, str, int, float, bool,
None]) and update ProfilePatch.value to that type, or add a validator on
ProfilePatch (e.g., __post_init__ or a validate_json_safe method) that rejects
non-serializable types, and invoke the same validation from both
InMemoryProfileStore and SQLiteProfileStore before persisting so unsupported
values are consistently rejected; update the other identical definition
referenced (lines 167-176) the same way.
- Around line 181-184: The InMemoryProfileStore currently leaves
Patch.last_confirmed_at as None for new inserts while SQLiteProfileStore sets it
to now, causing inconsistent semantics; update InMemoryProfileStore (the method
that creates/inserts new Patch instances) to initialize last_confirmed_at to the
same timestamp used for created_at (now) for brand-new patches so both stores
treat a fresh insert as "confirmed at creation" (also ensure any code that reads
patch.last_confirmed_at uses isoformat consistently as in the SQLite path).
- Around line 88-120: The blocking sqlite3 calls in SQLiteProfileStore (methods
list_active, list_all, save, _ensure_initialized and helpers like _row_to_patch)
run on the event loop and must be made asynchronous: either replace the sync
sqlite3 usage with aiosqlite (open an aiosqlite Connection and use await
connection.execute/await cursor.fetchall) or wrap all synchronous DB
interactions inside asyncio.to_thread calls (e.g., move the
execute/fetchall/commit/close sequences into functions and call them via
asyncio.to_thread) so the event loop isn’t blocked; update connection handling
accordingly so the async methods await the DB work and preserve existing return
types.
In `@vektori/prompts.py`:
- Around line 134-135: The code calls _render_system_prompt(context,
profile_patches, runtime_overrides) twice and assigns both results to
system_prompt, causing a redundant duplicate; remove the second identical call
so system_prompt is only assigned once—update the occurrence that repeats the
assignment of system_prompt (the duplicate line) and keep the single call to
_render_system_prompt(context, profile_patches, runtime_overrides).
In `@vektori/retrieval/scoring.py`:
- Around line 118-124: The JSON-parsing branch for metadata can produce non-dict
values (e.g., list, string, null) and then calling raw_meta.get(...) causes
AttributeError; after the json.loads() in the block that handles raw_meta =
fact.get("metadata") or {} (and the try/except importing json), add a type check
to ensure raw_meta is a dict before using .get(), and if it's not a dict replace
it with an empty dict (or safely extract a "source" key if you want to support
other types); update the variables referenced (raw_meta and source) so source =
raw_meta.get("source", "user") is only executed when raw_meta is a dict.
---
Nitpick comments:
In `@examples/agent_type_extraction.py`:
- Around line 38-41: The file imports the json module but never uses it; remove
the unused import statement ("import json") from
examples/agent_type_extraction.py so only needed imports (asyncio and from
vektori import ExtractionConfig, Vektori) remain, ensuring no other references
to json exist in functions like any extraction setup or main coroutine.
- Around line 140-142: The fixed 3-second asyncio.sleep is fragile because it
may not wait long enough for async extraction; replace the hard sleep with a
polling/wait loop that repeatedly calls presales_v.search (or a dedicated
completion API if available) until expected results are present or a timeout is
reached, e.g., poll presales_v.search("budget and decision makers",
user_id="demo-user") with short await asyncio.sleep intervals and a max timeout,
and document the timeout behavior so the demo handles variable network/load
conditions gracefully.
In `@examples/pipecat_voice_agent.py`:
- Around line 136-141: The current CORSMiddleware setup in the
app.add_middleware call uses allow_origins=["*"], which is fine for examples but
unsafe in production; update the code that configures CORSMiddleware (the
app.add_middleware(..., CORSMiddleware, allow_origins=... ) call) to read
allowed origins from configuration or an environment variable (e.g.,
ALLOWED_ORIGINS or a settings object) and only use ["*"] when explicitly running
in a local/dev mode, otherwise pass a restricted list of origins or raise an
error if none are configured.
In `@examples/vektori_agent_demo.py`:
- Around line 23-26: The demo currently calls agent.chat(...) and then closes
resources, but if agent.chat raises an exception the cleanup (agent.close() and
memory.close()) is skipped; wrap the interaction in a try/finally so that
agent.close() and memory.close() are always executed (place the result = await
agent.chat(...) and print(...) inside the try block and call await agent.close()
and await memory.close() in the finally block), referencing the agent.chat,
agent.close, and memory.close symbols to locate the change.
In `@README.md`:
- Around line 153-162: The README's interactive loop can skip cleanup if an
exception occurs; wrap the chat loop that calls agent.chat(user_input) in a
try/finally so that await agent.close() and await memory.close() always run;
specifically, put the while True input/chat sequence (and any awaiting of
agent.chat) inside a try block and call await agent.close() and await
memory.close() in the finally block to mirror the safe pattern used in
examples/openai_agent.py and ensure proper async cleanup.
In `@vektori/agent.py`:
- Around line 193-196: The close() method may raise AttributeError if
self.profile_store lacks a close() method; update Agent.close to defensively
call or await profile_store.close only when available — e.g., check
hasattr(self.profile_store, "close") or wrap the call in try/except Exception to
swallow AttributeError (and optionally log it); ensure the existing await
asyncio.gather(*self._background_tasks, return_exceptions=True) logic remains
unchanged and that any async close is awaited correctly when present.
In `@vektori/cli.py`:
- Around line 767-769: _slug_to_path currently replaces the first hyphen with
"/" then turns all remaining hyphens into spaces, which mangles legitimate
hyphenated path segments; update the function (_slug_to_path) to only convert
the first hyphen into a path separator and leave subsequent hyphens unchanged
(i.e., remove the second replace that turns "-" into " "), then strip and return
the result so a slug like "home-user-my-project" becomes "home/user-my-project".
In `@vektori/context.py`:
- Around line 35-70: The _resolve_path method builds the same candidate paths in
a verbose, repeated way; simplify by consolidating the filenames list (e.g.,
["agents.md","vektori.yaml","vektori.yml"]) and the search roots (cwd plus first
three cwd.parents) then generate candidates via a nested loop/comprehension
combining each root with each filename (while still honoring an explicit
self.context_path if provided), keep the existing behavior of returning the
first existing candidate or None and reference the function name _resolve_path,
the candidates variable, and context_path when making the change.
In `@vektori/integrations/pipecat/processor.py`:
- Around line 329-331: The current use of
asyncio.ensure_future(self._store_turn()) in the LLMFullResponseEndFrame branch
creates an untracked fire-and-forget task that can be lost on shutdown; replace
this with a tracked background task pattern: create the task via
asyncio.create_task(self._store_turn()), append it to a per-instance task list
(e.g., self._background_tasks) and ensure the processor exposes shutdown/cleanup
logic that cancels and awaits tasks (similar to VektoriAgent's _background_tasks
handling) so _store_turn() is awaited or cancelled cleanly during shutdown.
- Around line 319-322: The current handling in the block checking
FrameDirection.DOWNSTREAM and isinstance(frame, TranscriptionFrame) assigns
self._pending_user = frame.text or "" which discards earlier partial
transcriptions; change this to accumulate partials or only replace on final
markers depending on desired behavior. Update the TranscriptionFrame handling in
the processor to either append non-final transcription chunks to
self._pending_user (e.g., concatenate with a separator and guard against
duplicates) or check a "is_final" flag on TranscriptionFrame (or the arrival of
LLMFullResponseEndFrame) and only set/replace self._pending_user when the frame
is final; ensure references to FrameDirection.DOWNSTREAM, TranscriptionFrame,
self._pending_user and LLMFullResponseEndFrame are used so the logic correctly
preserves and finalizes the transcript.
In `@vektori/memory/window.py`:
- Around line 57-90: The compaction currently discards old messages even when
summarizer.complete(...) returns no content; update the compact method so it
only replaces _recent_messages, updates _rolling_summary, and increments
_compaction_count when result.content is truthy; if result.content is falsy,
leave _recent_messages and _rolling_summary intact and return False (or
propagate a failure), ensuring the summarizer.complete call is checked before
truncating state in compact.
In `@vektori/models/litellm_provider.py`:
- Around line 67-74: The complete method's type hints (messages and tools) in
LitellmProvider use list[dict[str, object]] which mismatch the ChatModelProvider
base signature that uses list[dict[str, Any]]; update the complete method
signature to use Any for those dict value types, add Any to the module imports,
and ensure the method signature in class LitellmProvider matches the base
ChatModelProvider exactly so IDE/type checkers see the override correctly.
In `@vektori/models/openai.py`:
- Around line 105-112: The method signature of openai.ChatModel.complete uses
list[dict[str, object]] which mismatches the abstract base contract that expects
list[dict[str, Any]]; update the signature in the complete method (and any
similar signatures such as in LiteLLMChatModel) to use typing.Any for message
dict values, add Any to the imports, and ensure type annotations across the
class match the base class contract (i.e., messages: list[dict[str, Any]]) so
the method properly implements the base class.
- Around line 98-103: OpenAIChatModel._get_client currently imports AsyncOpenAI
without ImportError handling; update this method to mirror
OpenAIEmbedder._get_client by wrapping "from openai import AsyncOpenAI" in a
try/except ImportError and raise a clear, actionable ImportError that mentions
the missing "openai" package and how to install it (include the original
exception or message for debugging), then proceed to instantiate self._client =
AsyncOpenAI(api_key=self._api_key) as before.
In `@vektori/prompts.py`:
- Around line 116-117: The _within_budget function repeatedly calls _assemble()
causing expensive full message rebuilds during the trim loop; change the logic
to cache the assembled messages and their token estimate once (use a local
variable for assembled = _assemble() and tokens =
estimate_tokens_for_messages(assembled)), then have _within_budget reference
that cached tokens (or better, replace _within_budget usage with a direct check
against the cached tokens), and after each trim operation update the cached
tokens by calling a small helper _recompute_tokens() (or decrement tokens by the
known token delta of the removed item) rather than re-calling _assemble();
update references to _within_budget, _assemble, estimate_tokens_for_messages,
budget and add or use _recompute_tokens accordingly.
- Around line 34-51: _render_system_prompt currently ignores LoadedAgentContext
fields response_style, memory_policy, and extra_sections; update the function to
include these fields in the generated system prompt (e.g., add sections like
"Response style:" showing context.response_style, "Memory policy:" showing
context.memory_policy, and render each item in context.extra_sections) so the
prompt reflects all context data, or if they are intentionally unused add a
short TODO comment inside _render_system_prompt referencing response_style,
memory_policy, and extra_sections to clarify intended future use.
🪄 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: 00a472ef-54cb-4c84-8ba0-c8580d15aa18
📒 Files selected for processing (38)
.gitignoreREADME.mddemo.shdocs/AGENT_HARNESS_CHECKLIST.mddocs/AGENT_HARNESS_SPEC.mdexamples/agent_type_extraction.pyexamples/openai_agent.pyexamples/pipecat_voice_agent.pyexamples/vektori_agent_demo.pypyproject.tomltests/integration/test_agent_chat_flow.pytests/unit/test_agent_context.pytests/unit/test_agent_type_presets.pytests/unit/test_cli_agent.pytests/unit/test_harness_spec_checklist.pytests/unit/test_message_window.pytests/unit/test_profile_store.pytests/unit/test_prompt_builder.pyvektori/__init__.pyvektori/agent.pyvektori/cli.pyvektori/client.pyvektori/config.pyvektori/context.pyvektori/ingestion/extractor.pyvektori/ingestion/filter.pyvektori/integrations/__init__.pyvektori/integrations/pipecat/__init__.pyvektori/integrations/pipecat/processor.pyvektori/memory/__init__.pyvektori/memory/profile.pyvektori/memory/window.pyvektori/models/base.pyvektori/models/factory.pyvektori/models/litellm_provider.pyvektori/models/openai.pyvektori/prompts.pyvektori/retrieval/scoring.py
Comment on lines
+45
to
+47
| export VEKTORI_USER_ID="${VEKTORI_USER_ID:-dev}" | ||
| export VEKTORI_EXTRACTION_MODEL="${VEKTORI_EXTRACTION_MODEL:-litellm:groq/llama-3.3-70b-versatile}" | ||
| export VEKTORI_EMBEDDING_MODEL="${VEKTORI_EMBEDDING_MODEL:-sentence-transformers:all-MiniLM-L6-v2}" |
There was a problem hiding this comment.
The script ignores caller-supplied VEKTORI_USER_ID.
Every command still passes -u dev, so the exported VEKTORI_USER_ID never actually takes effect. That makes the demo silently read/write the wrong memory namespace for anyone who sets a different user in the environment.
Also applies to: 68-116
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@demo.sh` around lines 45 - 47, The demo hardcodes the user flag as "dev" so
the exported VEKTORI_USER_ID variable is ignored; update the script to use the
VEKTORI_USER_ID environment variable everywhere the user/namespace is passed
(replace any literal "-u dev" or similar with "-u ${VEKTORI_USER_ID}" or the
shell variable usage pattern used elsewhere), including all command invocations
in the block around VEKTORI_USER_ID and the later repeated section (the area
covering the commands referenced in the comment), so callers who set
VEKTORI_USER_ID get their value applied.
Comment on lines
+248
to
+258
| @transport.event_handler("on_client_disconnected") | ||
| async def on_disconnected(t, client): | ||
| logger.info("Client disconnected: user=%s — closing session", user_id) | ||
| await task.queue_frames([EndFrame()]) | ||
| await vektori.close() | ||
|
|
||
| # ------------------------------------------------------------------ | ||
| # Run | ||
| # ------------------------------------------------------------------ | ||
| runner = PipelineRunner() | ||
| await runner.run(task) |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check if Vektori.close() is idempotent (safe to call multiple times)
rg -n "async def close" --type=py -A 10 | head -50Repository: vektori-ai/vektori
Length of output: 3025
🏁 Script executed:
# Search for other on_client_disconnected handler patterns in the codebase
rg -n "on_client_disconnected" --type=py -B 2 -A 5Repository: vektori-ai/vektori
Length of output: 661
🏁 Script executed:
# Check if there are other voice agent examples with similar cleanup patterns
fd -e "*.py" -path "*/examples/*" -name "*voice*" -o -path "*/examples/*" -name "*agent*" | head -10Repository: vektori-ai/vektori
Length of output: 230
🏁 Script executed:
# Look for try/finally patterns in other examples
rg -n "try:" --type=py -A 3 examples/ | grep -A 3 "close()" | head -40Repository: vektori-ai/vektori
Length of output: 44
🏁 Script executed:
# Find all Python files in examples directory
find examples -name "*.py" -type fRepository: vektori-ai/vektori
Length of output: 367
🏁 Script executed:
# Search for async with or context manager patterns in examples
rg -n "async with" examples/ --type=py -B 1 -A 3 | head -50Repository: vektori-ai/vektori
Length of output: 417
🏁 Script executed:
# Look for how transport cleanup is handled in examples
rg -n "transport" examples/ --type=py | head -20Repository: vektori-ai/vektori
Length of output: 650
🏁 Script executed:
# Read the complete voice_endpoint function to see full cleanup context
sed -n '148,258p' examples/pipecat_voice_agent.py | cat -nRepository: vektori-ai/vektori
Length of output: 5193
🏁 Script executed:
# Check the full scope of the voice_endpoint function for any outer try/finally
sed -n '140,260p' examples/pipecat_voice_agent.py | cat -nRepository: vektori-ai/vektori
Length of output: 5510
🏁 Script executed:
# Search for PipelineRunner implementation or documentation
rg -n "class PipelineRunner" --type=py -A 10Repository: vektori-ai/vektori
Length of output: 44
🏁 Script executed:
# Check if there are any try/finally patterns in pipecat-related code
rg -n "try:" examples/pipecat_voice_agent.py -A 5Repository: vektori-ai/vektori
Length of output: 44
🏁 Script executed:
# Look for how FastAPIWebsocketTransport handles cleanup or disconnects
rg -n "FastAPIWebsocketTransport\|on_client_disconnected" --type=py -B 2 -A 5Repository: vektori-ai/vektori
Length of output: 44
🌐 Web query:
Pipecat PipelineRunner cleanup patterns best practices
💡 Result:
Pipecat PipelineRunner cleanup patterns and best practices focus on graceful shutdown, automatic resource management, and preventing leaks in voice AI pipelines. ## Core Cleanup Mechanism PipelineRunner automatically handles cleanup through its lifecycle: - After task completion or cancellation, calls await self.cleanup on the base object. - Supports optional force_gc=True to run gc.collect post-run for memory cleanup. - Signal handling: handle_sigint=True (default) and handle_sigterm=False (opt-in) trigger cancel on SIGINT/SIGTERM for graceful interruption. Example initialization: runner = PipelineRunner(force_gc=True, handle_sigint=True) await runner.run(task) ## Termination Patterns Use frame-based termination for clean shutdowns (from docs.pipecat.ai/guides/learn/pipeline-termination): 1. Graceful (EndFrame): await task.queue_frame(EndFrame) or await task.stop_when_done - processes pending frames before cleanup. 2. Immediate (CancelFrame): await task.cancel - discards pending frames, sends CancelFrame downstream. Tie to transport events (quickstart example): @transport.event_handler("on_client_disconnected") async def on_client_disconnected(transport, client): await task.cancel # Triggers pipeline cleanup ## PipelineTask Cleanup Sequence PipelineTask orchestrates processor cleanup (pipecat.pipeline.task source): 1. Handles CancelledError, calls _cleanup(cleanup_pipeline=True). 2. Cleans base objects, observers (await self._observer.cleanup), ends tracing. 3. Calls await self._pipeline.cleanup to propagate cleanup through processors. Processors implement async def cleanup(self) for resource release (e.g., VAD analyzers added in CHANGELOG 0.0.108). ## Best Practices 1. Event Handlers: Use @task.event_handler("on_pipeline_finished") for post-termination logic: @task.event_handler("on_pipeline_finished") async def on_pipeline_finished(task, frame): if isinstance(frame, EndFrame): logger.info("Normal completion") 2. Idle Detection: Enable idle_timeout_secs in PipelineTaskParams; auto-cancels on timeout (default 300s). 3. Custom Processors: Always propagate termination frames and implement cleanup: async def process_frame(self, frame: Frame, direction: FrameDirection): if isinstance(frame, (EndFrame, CancelFrame)): await self.cleanup await self.push_frame(frame) async def cleanup(self): # Release resources (files, connections) await super.cleanup 4. Error Handling: Wrap runner.run(task) in try/except, call task.cancel on errors. 5. Observers: Stop proxy tasks before observer cleanup to avoid races (GitHub issue #4195). 6. Deployment: Use handle_sigint=False when main runner handles signals; containerize for scaling. These ensure no hanging processes, resource leaks, or lost metrics. Prioritize official docs (reference-server.pipecat.ai) and GitHub CHANGELOG for updates.
Citations:
- 1: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/pipeline/runner.html
- 2: https://docs.pipecat.ai/pipecat/get-started/quickstart
- 3: TaskObserver.cleanup() closes observer resources before stopping proxy tasks pipecat-ai/pipecat#4195
- 4: https://reference-server.pipecat.ai/en/stable/api/pipecat.pipeline.runner.html
- 5: https://reference-server.pipecat.ai/en/stable/_modules/pipecat/pipeline/runner.html
- 6: https://docs.pipecat.ai/server/pipeline/pipeline-task
- 7: https://reference-server.pipecat.ai/en/stable/_modules/pipecat/pipeline/task.html
- 8: https://reference-server.pipecat.ai/en/stable/api/pipecat.pipeline.task.html
- 9: https://docs.pipecat.ai/guides/learn/pipeline-termination
🌐 Web query:
Pipecat transport on_client_disconnected handler reliability fastapi websocket
💡 Result:
The on_client_disconnected handler in Pipecat's FastAPIWebsocketTransport is generally reliable for detecting client WebSocket disconnections. It is triggered in the receive message loop when a receive exception occurs (indicating client disconnect) and the connection is not already closing due to server-initiated disconnect. Official docs confirm the event fires on client WebSocket disconnection, with parameters (transport, websocket). Source code shows it calls the callback only if not self._client.is_closing, distinguishing client vs server disconnects. However, there is a known potential reliability issue (GitHub issue #3710, reported 2026-02-11): If a send exception occurs first (e.g., due to client hangup), it sets _closing=True, preventing the receive exception from triggering on_client_disconnected if receive fails afterward due to race conditions. A fix was proposed via PR #3712 (remove _closing set in send handler), but status as of 2026-04-14 is unclear from available data. Similar issues exist in other transports like Daily (fixed in 2025). Usage example: @transport.event_handler("on_client_disconnected") async def on_client_disconnected(transport, websocket): print("Client disconnected") Primarily designed for telephony WebSockets (Twilio etc.), but works generally with proper serializer config (e.g., ProtobufFrameSerializer for non-telephony). No widespread reports of unreliability beyond the race condition bug.
Citations:
- 1: https://docs.pipecat.ai/api-reference/server/services/transport/daily
- 2: https://docs.pipecat.ai/api-reference/server/events/overview
- 3: https://docs.pipecat.ai/server/services/transport/fastapi-websocket
- 4: https://docs.pipecat.ai/api-reference/server/services/transport/fastapi-websocket
- 5: https://reference-server.pipecat.ai/en/latest/_modules/pipecat/transports/websocket/fastapi.html
- 6: Potential bug in FastAPIWebsocketClient pipecat-ai/pipecat#3710
- 7: https://github.com/pipecat-ai/pipecat/blob/dc97ffc9/src/pipecat/transports/websocket/fastapi.py
🌐 Web query:
FastAPI WebSocket handler cleanup guarantees network disconnect
💡 Result:
FastAPI does not provide automatic cleanup guarantees for WebSocket handlers on network disconnects. Developers must manually handle cleanup by catching the WebSocketDisconnect exception in their handler code. Key points from official FastAPI/Starlette documentation and sources: - When the ASGI server (e.g., Uvicorn) detects a client disconnect, it sends a "websocket.disconnect" message to the WebSocket endpoint. - In Starlette's WebSocket implementation (used by FastAPI), calling receive_* methods (like receive_text) after receiving this message raises WebSocketDisconnect exception. This allows catching it in a try-except block around the receive loop for cleanup (e.g., removing from active connections list). - Example from official docs: @app.websocket("/ws/{client_id}") async def websocket_endpoint(websocket: WebSocket, client_id: int): await manager.connect(websocket) try: while True: data = await websocket.receive_text # ... process data except WebSocketDisconnect: manager.disconnect(websocket) - No automatic on_disconnect callback in function-style endpoints (most common). WebSocketEndpoint class provides on_disconnect, but it's less used. - For network disconnects (e.g., client crash, network failure), the server detects TCP close or timeout, triggers the disconnect message, and the exception is raised on next receive/send attempt. - If handler is not actively receiving (e.g., awaiting other tasks), disconnect may not be detected immediately. Workarounds: concurrent tasks for receive_text and other operations using asyncio.wait(FIRST_COMPLETED), or heartbeats/periodic sends with try-except. - Server shutdown requires manual handling (e.g., @app.on_event("shutdown") to close active sockets), as it doesn't automatically trigger per-connection cleanup. - No built-in guarantees for resource cleanup (memory, DB connections, tasks) without explicit developer code. Common issues: "zombie" connections if not removed from lists, tasks not cancelled. In summary, cleanup is guaranteed only if you implement exception handling and connection management properly. FastAPI/Starlette provide the mechanism (WebSocketDisconnect exception), but not automatic execution.
Citations:
- 1: https://fastapi.tiangolo.com/advanced/websockets/
- 2: https://github.com/encode/starlette/blob/master/starlette/websockets.py
- 3: https://hexshift.medium.com/handling-websocket-disconnections-gracefully-in-fastapi-9f0a1de365da
- 4: WebSocketDisconnect not caught while waiting on async tasks fastapi/fastapi#3934
- 5: Disconnect Events without calling receive_text Kludex/starlette#1121
- 6: Handle websocket disconnect in event loop Kludex/starlette#759
Potential resource leak if disconnect handler doesn't fire due to known Pipecat transport race condition.
Pipecat's FastAPIWebsocketTransport has a known race condition (GitHub #3710) where a send exception can prevent on_client_disconnected from firing if it occurs before the receive exception. Since vektori is an external service (not a Pipecat processor), it won't be automatically cleaned up by PipelineRunner. Consider wrapping runner.run(task) in try/finally to ensure cleanup.
Ensure vektori cleanup with try/finally
# ------------------------------------------------------------------
# Run
# ------------------------------------------------------------------
runner = PipelineRunner()
- await runner.run(task)
+ try:
+ await runner.run(task)
+ finally:
+ await vektori.close()Note: Vektori.close() is idempotent (checks if connection exists before closing), so calling it twice (once in the handler, once in finally) is safe.
📝 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
| @transport.event_handler("on_client_disconnected") | |
| async def on_disconnected(t, client): | |
| logger.info("Client disconnected: user=%s — closing session", user_id) | |
| await task.queue_frames([EndFrame()]) | |
| await vektori.close() | |
| # ------------------------------------------------------------------ | |
| # Run | |
| # ------------------------------------------------------------------ | |
| runner = PipelineRunner() | |
| await runner.run(task) | |
| `@transport.event_handler`("on_client_disconnected") | |
| async def on_disconnected(t, client): | |
| logger.info("Client disconnected: user=%s — closing session", user_id) | |
| await task.queue_frames([EndFrame()]) | |
| await vektori.close() | |
| # ------------------------------------------------------------------ | |
| # Run | |
| # ------------------------------------------------------------------ | |
| runner = PipelineRunner() | |
| try: | |
| await runner.run(task) | |
| finally: | |
| await vektori.close() |
🤖 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 248 - 258, The disconnect
handler (on_disconnected) may not run due to a Pipecat transport race, risking
that external resource vektori isn't closed; wrap the call to
PipelineRunner().run(task) in a try/finally so vektori.close() is always awaited
on exit (safe because Vektori.close() is idempotent), leaving the existing await
vektori.close() in on_disconnected as-is; locate PipelineRunner,
runner.run(task), and vektori.close to implement the finally cleanup.
Comment on lines
+5
to
+7
| path = Path("docs/AGENT_HARNESS_CHECKLIST.md") | ||
| assert path.exists() | ||
| text = path.read_text() |
There was a problem hiding this comment.
Resolve the docs path from the test file, not the process CWD.
Path("docs/AGENT_HARNESS_CHECKLIST.md") only works when pytest is launched from the repository root. Running this test from an IDE or a subdirectory will fail even though the file exists.
🧭 Suggested fix
def test_harness_spec_checklist_exists():
- path = Path("docs/AGENT_HARNESS_CHECKLIST.md")
+ path = Path(__file__).resolve().parents[2] / "docs" / "AGENT_HARNESS_CHECKLIST.md"
assert path.exists()
- text = path.read_text()
+ text = path.read_text(encoding="utf-8")🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@tests/unit/test_harness_spec_checklist.py` around lines 5 - 7, The test
currently uses Path("docs/AGENT_HARNESS_CHECKLIST.md") which depends on the
process CWD; change it to resolve the docs path relative to the test file
instead (use Path(__file__).resolve().parent/.. or .parent.parent then /
"docs/AGENT_HARNESS_CHECKLIST.md") so the path lookup in the
test_harness_spec_checklist.py file does not break when pytest is run from an
IDE or other working directory; update the references to the
Path("docs/AGENT_HARNESS_CHECKLIST.md") variable (e.g., the path variable) to
this resolved path before calling exists() and read_text().
Comment on lines
+18
to
+25
| key: str | ||
| value: Any | ||
| reason: str | ||
| source: str | ||
| observer_id: str | ||
| observed_id: str | ||
| confidence: float | ||
| created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) |
There was a problem hiding this comment.
Make ProfilePatch.value backend-portable.
ProfilePatch.value is public Any, but the SQLite backend persists it via json.dumps(...). A patch that works with InMemoryProfileStore can therefore start failing with SQLiteProfileStore purely because the backend changed. Please narrow this field to a JSON-safe type, or validate/reject unsupported values consistently before storage.
Also applies to: 167-176
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/memory/profile.py` around lines 18 - 25, ProfilePatch.value is typed
as Any but SQLiteProfileStore serializes with json.dumps, causing
backend-specific failures; narrow the type to a JSON-safe alias (e.g., JSONValue
= Union[dict, list, str, int, float, bool, None]) and update ProfilePatch.value
to that type, or add a validator on ProfilePatch (e.g., __post_init__ or a
validate_json_safe method) that rejects non-serializable types, and invoke the
same validation from both InMemoryProfileStore and SQLiteProfileStore before
persisting so unsupported values are consistently rejected; update the other
identical definition referenced (lines 167-176) the same way.
Comment on lines
+88
to
+120
| async def list_active(self, observer_id: str, observed_id: str) -> list[ProfilePatch]: | ||
| await self._ensure_initialized() | ||
| assert self._connection is not None | ||
| cursor = self._connection.execute( | ||
| """ | ||
| SELECT key, value_json, reason, source, observer_id, observed_id, | ||
| confidence, created_at, last_confirmed_at, active | ||
| FROM profile_patches | ||
| WHERE observer_id = ? AND observed_id = ? AND active = 1 | ||
| ORDER BY created_at ASC | ||
| """, | ||
| (observer_id, observed_id), | ||
| ) | ||
| rows = cursor.fetchall() | ||
| cursor.close() | ||
| return [self._row_to_patch(row) for row in rows] | ||
|
|
||
| async def list_all(self, observer_id: str, observed_id: str) -> list[ProfilePatch]: | ||
| await self._ensure_initialized() | ||
| assert self._connection is not None | ||
| cursor = self._connection.execute( | ||
| """ | ||
| SELECT key, value_json, reason, source, observer_id, observed_id, | ||
| confidence, created_at, last_confirmed_at, active | ||
| FROM profile_patches | ||
| WHERE observer_id = ? AND observed_id = ? | ||
| ORDER BY created_at ASC | ||
| """, | ||
| (observer_id, observed_id), | ||
| ) | ||
| rows = cursor.fetchall() | ||
| cursor.close() | ||
| return [self._row_to_patch(row) for row in rows] |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
find . -name "profile.py" -path "*/memory/*" | head -5Repository: vektori-ai/vektori
Length of output: 89
🏁 Script executed:
cat -n ./vektori/memory/profile.pyRepository: vektori-ai/vektori
Length of output: 9862
🏁 Script executed:
rg -l "SQLiteProfileStore|ProfileStore" --type py | head -20Repository: vektori-ai/vektori
Length of output: 164
🏁 Script executed:
rg -B 2 -A 2 "ProfileStore\|SQLiteProfileStore" ./vektori/agent.py | head -50Repository: vektori-ai/vektori
Length of output: 44
🏁 Script executed:
wc -l ./vektori/agent.py && head -100 ./vektori/agent.pyRepository: vektori-ai/vektori
Length of output: 3594
🏁 Script executed:
grep -n "profile_store" ./vektori/agent.pyRepository: vektori-ai/vektori
Length of output: 582
🏁 Script executed:
sed -n '100,120p' ./vektori/agent.py && echo "---" && sed -n '220,230p' ./vektori/agent.pyRepository: vektori-ai/vektori
Length of output: 1215
Switch this profile store to aiosqlite or move the DB work behind asyncio.to_thread.
The SQLiteProfileStore uses blocking sqlite3 operations in async methods (list_active() at line 104, save(), and _ensure_initialized()). These methods are called directly in the chat path during message handling, where blocking I/O on the event loop will stall other concurrent sessions instead of yielding.
Also applies to: 122-218
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/memory/profile.py` around lines 88 - 120, The blocking sqlite3 calls
in SQLiteProfileStore (methods list_active, list_all, save, _ensure_initialized
and helpers like _row_to_patch) run on the event loop and must be made
asynchronous: either replace the sync sqlite3 usage with aiosqlite (open an
aiosqlite Connection and use await connection.execute/await cursor.fetchall) or
wrap all synchronous DB interactions inside asyncio.to_thread calls (e.g., move
the execute/fetchall/commit/close sequences into functions and call them via
asyncio.to_thread) so the event loop isn’t blocked; update connection handling
accordingly so the async methods await the DB work and preserve existing return
types.
Comment on lines
+181
to
+184
| patch.confidence, | ||
| patch.created_at.isoformat(), | ||
| patch.last_confirmed_at.isoformat() if patch.last_confirmed_at else now, | ||
| 1 if patch.active else 0, |
There was a problem hiding this comment.
Keep last_confirmed_at semantics consistent across stores.
A brand-new patch inserted into SQLiteProfileStore gets last_confirmed_at=now, while InMemoryProfileStore leaves it as None until the value is explicitly confirmed again. That makes the same field mean different things depending on backend and collapses “created” vs “re-confirmed”.
🩹 Suggested fix
self._connection.execute(
"""
INSERT INTO profile_patches (
key, value_json, reason, source, observer_id, observed_id,
confidence, created_at, last_confirmed_at, active
@@
patch.observed_id,
patch.confidence,
patch.created_at.isoformat(),
- patch.last_confirmed_at.isoformat() if patch.last_confirmed_at else now,
+ patch.last_confirmed_at.isoformat() if patch.last_confirmed_at else None,
1 if patch.active else 0,
),
)📝 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
| patch.confidence, | |
| patch.created_at.isoformat(), | |
| patch.last_confirmed_at.isoformat() if patch.last_confirmed_at else now, | |
| 1 if patch.active else 0, | |
| patch.confidence, | |
| patch.created_at.isoformat(), | |
| patch.last_confirmed_at.isoformat() if patch.last_confirmed_at else None, | |
| 1 if patch.active else 0, |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/memory/profile.py` around lines 181 - 184, The InMemoryProfileStore
currently leaves Patch.last_confirmed_at as None for new inserts while
SQLiteProfileStore sets it to now, causing inconsistent semantics; update
InMemoryProfileStore (the method that creates/inserts new Patch instances) to
initialize last_confirmed_at to the same timestamp used for created_at (now) for
brand-new patches so both stores treat a fresh insert as "confirmed at creation"
(also ensure any code that reads patch.last_confirmed_at uses isoformat
consistently as in the SQLite path).
Comment on lines
+134
to
+135
| system_prompt = _render_system_prompt(context, profile_patches, runtime_overrides) | ||
| system_prompt = _render_system_prompt(context, profile_patches, runtime_overrides) |
There was a problem hiding this comment.
Duplicate line: _render_system_prompt called twice.
Line 134 and 135 are identical. The second call overwrites the first with the same result. Remove the duplicate.
🔧 Proposed fix
- system_prompt = _render_system_prompt(context, profile_patches, runtime_overrides)
system_prompt = _render_system_prompt(context, profile_patches, runtime_overrides)
if estimate_tokens_for_text(system_prompt) > budget and system_prompt:📝 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
| system_prompt = _render_system_prompt(context, profile_patches, runtime_overrides) | |
| system_prompt = _render_system_prompt(context, profile_patches, runtime_overrides) | |
| system_prompt = _render_system_prompt(context, profile_patches, runtime_overrides) | |
| if estimate_tokens_for_text(system_prompt) > budget and system_prompt: |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/prompts.py` around lines 134 - 135, The code calls
_render_system_prompt(context, profile_patches, runtime_overrides) twice and
assigns both results to system_prompt, causing a redundant duplicate; remove the
second identical call so system_prompt is only assigned once—update the
occurrence that repeats the assignment of system_prompt (the duplicate line) and
keep the single call to _render_system_prompt(context, profile_patches,
runtime_overrides).
Comment on lines
+118
to
+124
| raw_meta = fact.get("metadata") or {} | ||
| if isinstance(raw_meta, str): | ||
| try: | ||
| import json as _json; raw_meta = _json.loads(raw_meta) | ||
| except Exception: | ||
| raw_meta = {} | ||
| source = raw_meta.get("source", "user") |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Verify the concern:
# 1) Confirm current parsing/get flow in scoring.
# 2) Check whether tests cover non-object JSON metadata.
# 3) Demonstrate json.loads return types for valid non-object JSON.
rg -n -C3 'raw_meta|source = raw_meta\.get' vektori/retrieval/scoring.py
rg -n -C3 'score_and_rank|metadata' tests vektori/retrieval
python - <<'PY'
import json
samples = ['{"source":"user"}', '[]', '"assistant"', 'null', '1']
for s in samples:
v = json.loads(s)
print(f"{s:20} -> {type(v).__name__}")
PYRepository: vektori-ai/vektori
Length of output: 16428
🏁 Script executed:
# Look at the _fact() helper definition and test metadata patterns
rg -n -B2 -A10 'def _fact' tests/unit/test_scoring.py
# Check if score_and_rank is ever tested with edge-case metadata
rg -n 'metadata.*\[|metadata.*null|metadata.*"[^{]' tests/Repository: vektori-ai/vektori
Length of output: 1421
🏁 Script executed:
# Find all calls to score_and_rank with metadata argument
rg -n -B3 -A3 'score_and_rank\(' vektori/retrieval/ | head -50
# Check what MilvusBackend._to_fact does to metadata (where it parses JSON)
rg -n -B5 -A15 '_to_fact' vektori/storage/Repository: vektori-ai/vektori
Length of output: 30827
🏁 Script executed:
# Find the _parse_metadata function in milvus.py
rg -n -B2 -A10 'def _parse_metadata' vektori/storage/
# Also check if metadata is ever pre-parsed or comes as non-dict to scoring
rg -n 'metadata.*=' vektori/storage/chroma_backend.py | head -20Repository: vektori-ai/vektori
Length of output: 1351
🏁 Script executed:
# Check how qdrant and lancedb backends actually store/return metadata
rg -n -B5 -A5 'metadata.*json|json.*metadata' vektori/storage/qdrant_backend.py vektori/storage/lancedb_backend.py
# Check the fact structure more carefully to understand if metadata can be a string at scoring time
rg -n 'def.*_payload_to_fact|def.*_row_to_fact' vektori/storage/ -A 20 | head -80Repository: vektori-ai/vektori
Length of output: 7849
Handle non-object JSON metadata before calling .get()
Lines 121–124 can decode valid JSON that is not a dict (e.g., [], "string", null), causing .get() to crash with AttributeError. This occurs because Qdrant and LanceDB backends store metadata as JSON strings without parsing them first. Add a type check after json.loads() to ensure metadata is a dict before calling .get().
Proposed fix
+import json
import math
from datetime import datetime, timezone
from typing import Any
@@
- raw_meta = fact.get("metadata") or {}
+ raw_meta = fact.get("metadata") or {}
if isinstance(raw_meta, str):
try:
- import json as _json; raw_meta = _json.loads(raw_meta)
- except Exception:
+ parsed_meta = json.loads(raw_meta)
+ raw_meta = parsed_meta if isinstance(parsed_meta, dict) else {}
+ except (json.JSONDecodeError, TypeError, ValueError):
raw_meta = {}
+ elif not isinstance(raw_meta, dict):
+ raw_meta = {}
source = raw_meta.get("source", "user")🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/retrieval/scoring.py` around lines 118 - 124, The JSON-parsing branch
for metadata can produce non-dict values (e.g., list, string, null) and then
calling raw_meta.get(...) causes AttributeError; after the json.loads() in the
block that handles raw_meta = fact.get("metadata") or {} (and the try/except
importing json), add a type check to ensure raw_meta is a dict before using
.get(), and if it's not a dict replace it with an empty dict (or safely extract
a "source" key if you want to support other types); update the variables
referenced (raw_meta and source) so source = raw_meta.get("source", "user") is
only executed when raw_meta is a dict.
There was a problem hiding this comment.
Actionable comments posted: 7
🧹 Nitpick comments (1)
vektori/agent.py (1)
216-219: Consider cancelling background tasks before awaiting them inclose().Currently
close()waits on all tasks but never cancels pending ones. Long-running writes can block shutdown indefinitely.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@vektori/agent.py` around lines 216 - 219, In close(), cancel any pending background tasks in self._background_tasks before awaiting them so long-running tasks don't block shutdown: iterate over self._background_tasks, call task.cancel() for non-done tasks, then await them with asyncio.gather(..., return_exceptions=True) as currently done, and finally call await self.profile_store.close(); ensure you still handle/ignore CancelledError and other exceptions from the gather result so close completes reliably.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@scripts/test_agent_e2e.py`:
- Around line 241-243: The code unconditionally requires OPENAI_API_KEY; change
the check so the script only enforces OPENAI_API_KEY when OpenAI is actually
selected as a provider. Replace the unconditional
os.environ.get("OPENAI_API_KEY") check with a conditional that inspects the
relevant provider env vars (e.g., VEKTORI_MODEL, EMBEDDING_PROVIDER,
EXTRACTION_PROVIDER, MODEL_PROVIDER) and only exits if any of those indicate
OpenAI (or are unset and default to OpenAI in your app logic); keep the existing
error message but only call sys.exit(1) when an OpenAI provider is in use.
- Around line 125-127: The test assumes model.complete is a mock by accessing
model.complete.call_args, which breaks with real providers; update the logic to
branch: if hasattr(model.complete, "call_args") (or isinstance(model.complete,
unittest.mock.Mock)) then build joined from model.complete.call_args as before,
otherwise build joined from the actual messages variable passed to the model
(e.g., use " ".join(m["content"] for m in messages if m["role"] == "system")).
Change the code that assigns joined to handle both cases so
model.complete.call_args is only accessed for mock objects.
In `@vektori/agent.py`:
- Around line 284-326: The loop currently allows one extra tool round
(range(self.config.max_tool_round_trips + 1)) which can return a completion that
still contains tool calls; change the loop to
range(self.config.max_tool_round_trips) to limit attempts, and after the loop
check if completion.tool_calls (or completion.tool_calls or raw_calls) is
non-empty — if so, request a final plain assistant completion by calling
self.model.complete(working_messages, tools=[],
max_tokens=self.config.reserve_response_tokens, temperature=0.2) (or tools=None
depending on provider) to obtain a non-tool completion before returning; update
references in this flow (max_tool_round_trips, model.complete, MEMORY_TOOLS,
working_messages, completion, tool_calls_log, handle_tool_call) accordingly so
the method never returns unresolved tool calls.
In `@vektori/memory/window.py`:
- Around line 91-100: The current compaction unconditionally replaces
_recent_messages and increments _compaction_count even when summarizer.complete
fails or returns empty; change the flow so you call summarizer.complete inside a
try/except, check that result.content is non-empty before mutating state, and
only then set _rolling_summary = result.content.strip(), set _recent_messages =
kept_messages, increment _compaction_count and return True; if an exception is
raised or result.content is falsy, do not modify _recent_messages or
_compaction_count (leave existing summary intact) and return False (or propagate
a controlled error) so compaction is non-destructive. Ensure you reference the
existing summarizer.complete call, the prompt variable, result.content,
kept_messages, and the attributes _rolling_summary, _recent_messages, and
_compaction_count when making these changes.
In `@vektori/tools/memory.py`:
- Line 190: The debug log currently emits sensitive profile values via
logger.debug("Tool: saved profile patch key=%s value=%s", key, value); change
this to avoid logging raw values by either logging only the key (e.g., "Tool:
saved profile patch key=%s") or replacing value with a redacted placeholder or
truncated/hashed representation; update the call in the same location (the
logger.debug invocation in vektori/tools/memory.py) so only non-sensitive
identifiers are logged and ensure any tests or callers expecting the old message
are adjusted accordingly.
- Around line 176-178: The runtime validation in update_profile currently only
checks key and value but the tool schema requires reason as well; update the
validation in the function (the block that returns "update_profile: 'key' and
'value' are required.") to also reject empty or missing reason (e.g., check if
not reason or reason.strip() == ""), and change the returned error text to
mention 'key', 'value', and 'reason' so it matches the declared schema and
prevents empty reasons from passing.
- Around line 132-143: The int(...) conversion for top_k happens outside the try
and can raise before the memory.search call; move the top_k parsing into the
same try block (or add its own guarded try/except) and handle
ValueError/TypeError by falling back to the default (6) and/or clamping to a
safe range; update the code around the agent.memory.search call in memory.py
(the top_k parsing and the try/except surrounding agent.memory.search and its
return) so that malformed args["top_k"] do not raise out of the tool but instead
produce a controlled warning and use the default top_k before calling
agent.memory.search.
---
Nitpick comments:
In `@vektori/agent.py`:
- Around line 216-219: In close(), cancel any pending background tasks in
self._background_tasks before awaiting them so long-running tasks don't block
shutdown: iterate over self._background_tasks, call task.cancel() for non-done
tasks, then await them with asyncio.gather(..., return_exceptions=True) as
currently done, and finally call await self.profile_store.close(); ensure you
still handle/ignore CancelledError and other exceptions from the gather result
so close completes reliably.
🪄 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: 74f95b7c-92fe-47e3-9052-0c00ea9f03f4
📒 Files selected for processing (11)
.gitignoredocs/AGENT_HARNESS_CHECKLIST.mdscripts/test_agent_e2e.pytests/unit/test_agent.pytests/unit/test_tools.pytests/unit/test_window_persistence.pyvektori/agent.pyvektori/memory/__init__.pyvektori/memory/window.pyvektori/tools/__init__.pyvektori/tools/memory.py
✅ Files skipped from review due to trivial changes (3)
- vektori/tools/init.py
- .gitignore
- docs/AGENT_HARNESS_CHECKLIST.md
🚧 Files skipped from review as they are similar to previous changes (1)
- vektori/memory/init.py
Comment on lines
+125
to
+127
| joined = " ".join( | ||
| m["content"] for m in model.complete.call_args[0][0] if m["role"] == "system" | ||
| ) |
There was a problem hiding this comment.
Phase 3 uses mock-only introspection on a real model object.
On Lines 125-127, model.complete.call_args will fail with real providers because this is not a mock in this script.
Suggested fix direction
- joined = " ".join(
- m["content"] for m in model.complete.call_args[0][0] if m["role"] == "system"
- )
- ok &= _check(
- "preferences.name" in joined or "Laxman" in joined,
- "profile patch appears in system prompt",
- )
+ # Verify persistence directly from profile store (provider-agnostic)
+ patches = await agent.profile_store.list_active(
+ observer_id=agent.agent_id or "default-agent",
+ observed_id=agent.user_id,
+ )
+ ok &= _check(
+ any(p.key == "preferences.name" and str(p.value).lower() == "laxman" for p in patches),
+ "name profile patch persisted",
+ )🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@scripts/test_agent_e2e.py` around lines 125 - 127, The test assumes
model.complete is a mock by accessing model.complete.call_args, which breaks
with real providers; update the logic to branch: if hasattr(model.complete,
"call_args") (or isinstance(model.complete, unittest.mock.Mock)) then build
joined from model.complete.call_args as before, otherwise build joined from the
actual messages variable passed to the model (e.g., use " ".join(m["content"]
for m in messages if m["role"] == "system")). Change the code that assigns
joined to handle both cases so model.complete.call_args is only accessed for
mock objects.
Comment on lines
+241
to
+243
| if not os.environ.get("OPENAI_API_KEY"): | ||
| print(f"{_RED}ERROR{_RESET}: OPENAI_API_KEY not set. Export it and re-run.") | ||
| sys.exit(1) |
There was a problem hiding this comment.
Credential gating is hardcoded to OpenAI despite configurable providers.
Lines 241-243 always require OPENAI_API_KEY, which incorrectly blocks runs when VEKTORI_MODEL/embedding/extraction are non-OpenAI providers.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@scripts/test_agent_e2e.py` around lines 241 - 243, The code unconditionally
requires OPENAI_API_KEY; change the check so the script only enforces
OPENAI_API_KEY when OpenAI is actually selected as a provider. Replace the
unconditional os.environ.get("OPENAI_API_KEY") check with a conditional that
inspects the relevant provider env vars (e.g., VEKTORI_MODEL,
EMBEDDING_PROVIDER, EXTRACTION_PROVIDER, MODEL_PROVIDER) and only exits if any
of those indicate OpenAI (or are unset and default to OpenAI in your app logic);
keep the existing error message but only call sys.exit(1) when an OpenAI
provider is in use.
Comment on lines
+284
to
+326
| for _ in range(self.config.max_tool_round_trips + 1): | ||
| completion = await self.model.complete( | ||
| working_messages, | ||
| tools=MEMORY_TOOLS, | ||
| max_tokens=self.config.reserve_response_tokens, | ||
| temperature=0.2, | ||
| ) | ||
|
|
||
| raw_calls = completion.tool_calls or [] | ||
| if not raw_calls: | ||
| break | ||
|
|
||
| # Append assistant's tool-calling turn | ||
| working_messages.append({ | ||
| "role": "assistant", | ||
| "content": completion.content or "", | ||
| "tool_calls": raw_calls, | ||
| }) | ||
|
|
||
| # Execute each tool and append results | ||
| for call in raw_calls: | ||
| # Normalize across providers: OpenAI object vs dict | ||
| if hasattr(call, "function"): | ||
| name = call.function.name | ||
| import json as _json | ||
| args = _json.loads(call.function.arguments or "{}") | ||
| call_id = getattr(call, "id", name) | ||
| else: | ||
| name = call.get("function", {}).get("name", "") | ||
| import json as _json | ||
| args = _json.loads(call.get("function", {}).get("arguments", "{}")) | ||
| call_id = call.get("id", name) | ||
|
|
||
| result_text = await handle_tool_call(name, args, self) | ||
| tool_calls_log.append({"name": name, "args": args, "result": result_text}) | ||
| working_messages.append({ | ||
| "role": "tool", | ||
| "tool_call_id": call_id, | ||
| "content": result_text, | ||
| }) | ||
|
|
||
| assert completion is not None | ||
| return completion, tool_calls_log |
There was a problem hiding this comment.
Tool loop bound allows an extra tool round and may return unresolved tool calls.
Using range(self.config.max_tool_round_trips + 1) (Line 284) permits one extra tool-call iteration, and if calls persist, the method can return a completion that still contains tool calls (no final plain completion).
Suggested fix pattern
- for _ in range(self.config.max_tool_round_trips + 1):
+ for _ in range(self.config.max_tool_round_trips):
completion = await self.model.complete(
working_messages,
tools=MEMORY_TOOLS,
max_tokens=self.config.reserve_response_tokens,
temperature=0.2,
)
@@
- if not raw_calls:
- break
+ if not raw_calls:
+ return completion, tool_calls_log
@@
- assert completion is not None
- return completion, tool_calls_log
+ # Final non-tool pass after capped tool rounds
+ completion = await self.model.complete(
+ working_messages,
+ max_tokens=self.config.reserve_response_tokens,
+ temperature=0.2,
+ )
+ return completion, tool_calls_log📝 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
| for _ in range(self.config.max_tool_round_trips + 1): | |
| completion = await self.model.complete( | |
| working_messages, | |
| tools=MEMORY_TOOLS, | |
| max_tokens=self.config.reserve_response_tokens, | |
| temperature=0.2, | |
| ) | |
| raw_calls = completion.tool_calls or [] | |
| if not raw_calls: | |
| break | |
| # Append assistant's tool-calling turn | |
| working_messages.append({ | |
| "role": "assistant", | |
| "content": completion.content or "", | |
| "tool_calls": raw_calls, | |
| }) | |
| # Execute each tool and append results | |
| for call in raw_calls: | |
| # Normalize across providers: OpenAI object vs dict | |
| if hasattr(call, "function"): | |
| name = call.function.name | |
| import json as _json | |
| args = _json.loads(call.function.arguments or "{}") | |
| call_id = getattr(call, "id", name) | |
| else: | |
| name = call.get("function", {}).get("name", "") | |
| import json as _json | |
| args = _json.loads(call.get("function", {}).get("arguments", "{}")) | |
| call_id = call.get("id", name) | |
| result_text = await handle_tool_call(name, args, self) | |
| tool_calls_log.append({"name": name, "args": args, "result": result_text}) | |
| working_messages.append({ | |
| "role": "tool", | |
| "tool_call_id": call_id, | |
| "content": result_text, | |
| }) | |
| assert completion is not None | |
| return completion, tool_calls_log | |
| for _ in range(self.config.max_tool_round_trips): | |
| completion = await self.model.complete( | |
| working_messages, | |
| tools=MEMORY_TOOLS, | |
| max_tokens=self.config.reserve_response_tokens, | |
| temperature=0.2, | |
| ) | |
| raw_calls = completion.tool_calls or [] | |
| if not raw_calls: | |
| return completion, tool_calls_log | |
| # Append assistant's tool-calling turn | |
| working_messages.append({ | |
| "role": "assistant", | |
| "content": completion.content or "", | |
| "tool_calls": raw_calls, | |
| }) | |
| # Execute each tool and append results | |
| for call in raw_calls: | |
| # Normalize across providers: OpenAI object vs dict | |
| if hasattr(call, "function"): | |
| name = call.function.name | |
| import json as _json | |
| args = _json.loads(call.function.arguments or "{}") | |
| call_id = getattr(call, "id", name) | |
| else: | |
| name = call.get("function", {}).get("name", "") | |
| import json as _json | |
| args = _json.loads(call.get("function", {}).get("arguments", "{}")) | |
| call_id = call.get("id", name) | |
| result_text = await handle_tool_call(name, args, self) | |
| tool_calls_log.append({"name": name, "args": args, "result": result_text}) | |
| working_messages.append({ | |
| "role": "tool", | |
| "tool_call_id": call_id, | |
| "content": result_text, | |
| }) | |
| # Final non-tool pass after capped tool rounds | |
| completion = await self.model.complete( | |
| working_messages, | |
| max_tokens=self.config.reserve_response_tokens, | |
| temperature=0.2, | |
| ) | |
| return completion, tool_calls_log |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/agent.py` around lines 284 - 326, The loop currently allows one extra
tool round (range(self.config.max_tool_round_trips + 1)) which can return a
completion that still contains tool calls; change the loop to
range(self.config.max_tool_round_trips) to limit attempts, and after the loop
check if completion.tool_calls (or completion.tool_calls or raw_calls) is
non-empty — if so, request a final plain assistant completion by calling
self.model.complete(working_messages, tools=[],
max_tokens=self.config.reserve_response_tokens, temperature=0.2) (or tools=None
depending on provider) to obtain a non-tool completion before returning; update
references in this flow (max_tool_round_trips, model.complete, MEMORY_TOOLS,
working_messages, completion, tool_calls_log, handle_tool_call) accordingly so
the method never returns unresolved tool calls.
Comment on lines
+91
to
+100
| 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.
Avoid destructive compaction when summarization fails or returns empty content.
On Line 91 and Lines 96-100, compaction can still drop old_messages even if summarization errors or returns empty text. That can silently lose context.
Suggested fix
- 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
+ try:
+ result = await summarizer.complete(
+ [{"role": "user", "content": prompt}],
+ max_tokens=self.summary_max_tokens,
+ temperature=0.0,
+ )
+ except Exception:
+ return False
+
+ summary = (result.content or "").strip()
+ if not summary:
+ return False
+
+ self._rolling_summary = summary
+ self._recent_messages = kept_messages
+ self._compaction_count += 1
+ return True📝 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
| 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 | |
| try: | |
| result = await summarizer.complete( | |
| [{"role": "user", "content": prompt}], | |
| max_tokens=self.summary_max_tokens, | |
| temperature=0.0, | |
| ) | |
| except Exception: | |
| return False | |
| summary = (result.content or "").strip() | |
| if not summary: | |
| return False | |
| self._rolling_summary = 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 91 - 100, The current compaction
unconditionally replaces _recent_messages and increments _compaction_count even
when summarizer.complete fails or returns empty; change the flow so you call
summarizer.complete inside a try/except, check that result.content is non-empty
before mutating state, and only then set _rolling_summary =
result.content.strip(), set _recent_messages = kept_messages, increment
_compaction_count and return True; if an exception is raised or result.content
is falsy, do not modify _recent_messages or _compaction_count (leave existing
summary intact) and return False (or propagate a controlled error) so compaction
is non-destructive. Ensure you reference the existing summarizer.complete call,
the prompt variable, result.content, kept_messages, and the attributes
_rolling_summary, _recent_messages, and _compaction_count when making these
changes.
Comment on lines
+132
to
+143
| top_k = int(args.get("top_k", 6)) | ||
| try: | ||
| results = await agent.memory.search( | ||
| query=query, | ||
| user_id=agent.user_id, | ||
| agent_id=agent.agent_id, | ||
| depth=depth, | ||
| top_k=top_k, | ||
| ) | ||
| except Exception as e: | ||
| logger.warning("search_memory tool call failed: %s", e) | ||
| return f"Memory search failed: {e}" |
There was a problem hiding this comment.
Harden top_k parsing for malformed tool arguments.
On Line 132, int(args.get("top_k", 6)) can raise before your try block, which can fail the whole tool round instead of returning a tool error string.
Suggested fix
- top_k = int(args.get("top_k", 6))
+ raw_top_k = args.get("top_k", 6)
+ try:
+ top_k = int(raw_top_k)
+ except (TypeError, ValueError):
+ return "search_memory: 'top_k' must be an integer."
+ top_k = max(1, min(top_k, 20))
try:
results = await agent.memory.search(🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/tools/memory.py` around lines 132 - 143, The int(...) conversion for
top_k happens outside the try and can raise before the memory.search call; move
the top_k parsing into the same try block (or add its own guarded try/except)
and handle ValueError/TypeError by falling back to the default (6) and/or
clamping to a safe range; update the code around the agent.memory.search call in
memory.py (the top_k parsing and the try/except surrounding agent.memory.search
and its return) so that malformed args["top_k"] do not raise out of the tool but
instead produce a controlled warning and use the default top_k before calling
agent.memory.search.
Comment on lines
+176
to
+178
| if not key or value is None: | ||
| return "update_profile: 'key' and 'value' are required." | ||
|
|
There was a problem hiding this comment.
Enforce reason to match the declared required schema.
On Lines 176-178, runtime validation only checks key and value, but tool schema marks reason as required. Empty/missing reasons currently pass.
Suggested fix
- if not key or value is None:
- return "update_profile: 'key' and 'value' are required."
+ if not key or value is None or not reason.strip():
+ return "update_profile: 'key', 'value', and 'reason' are required."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/tools/memory.py` around lines 176 - 178, The runtime validation in
update_profile currently only checks key and value but the tool schema requires
reason as well; update the validation in the function (the block that returns
"update_profile: 'key' and 'value' are required.") to also reject empty or
missing reason (e.g., check if not reason or reason.strip() == ""), and change
the returned error text to mention 'key', 'value', and 'reason' so it matches
the declared schema and prevents empty reasons from passing.
| created_at=datetime.now(timezone.utc), | ||
| ) | ||
| await agent.profile_store.save(patch) | ||
| logger.debug("Tool: saved profile patch key=%s value=%s", key, value) |
There was a problem hiding this comment.
Avoid logging raw profile values in debug logs.
Line 190 logs value directly; profile values may include user-identifying or sensitive preferences. Prefer key-only logging or redaction.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/tools/memory.py` at line 190, The debug log currently emits sensitive
profile values via logger.debug("Tool: saved profile patch key=%s value=%s",
key, value); change this to avoid logging raw values by either logging only the
key (e.g., "Tool: saved profile patch key=%s") or replacing value with a
redacted placeholder or truncated/hashed representation; update the call in the
same location (the logger.debug invocation in vektori/tools/memory.py) so only
non-sensitive identifiers are logged and ensure any tests or callers expecting
the old message are adjusted accordingly.
…ple scripts - Add Gemini agent integration demo and single-file web UI example - Add Hermes and OpenClaw memory adapter integrations under vektori/integrations/ - Add real-world support case and web transcript example scripts - Update README to link all new examples
There was a problem hiding this comment.
Actionable comments posted: 9
🧹 Nitpick comments (5)
tests/unit/test_factory.py (1)
70-84: ⚡ Quick winAdd a provider-only input test to cover default-model fallback.
Current cases only pass
provider:model; they don’t exercise themodel_name or Nonebranch increate_chat_model.Proposed test addition
+def test_create_chat_model_openai_default_model(): + from vektori.models.openai import OpenAIChatModel, DEFAULT_LLM_MODEL + + model = create_chat_model("openai") + assert isinstance(model, OpenAIChatModel) + assert model.model == DEFAULT_LLM_MODEL🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tests/unit/test_factory.py` around lines 70 - 84, Add tests that call create_chat_model with provider-only strings to exercise the branch where model_name is None: call create_chat_model("litellm") and assert isinstance(..., LiteLLMChatModel) and that model.model is a non-empty string (truthy) to verify the provider default is used; do the same for create_chat_model("openai") asserting isinstance(..., OpenAIChatModel) and that model.model is truthy.examples/web_transcript_case.py (2)
42-42: ⚡ Quick winAvoid blocking I/O directly inside the coroutine.
Line 42 calls synchronous network I/O from
async def main(), which blocks the event loop.Proposed patch
- messages = fetch_hn_discussion() + messages = await asyncio.to_thread(fetch_hn_discussion)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@examples/web_transcript_case.py` at line 42, The call to synchronous network I/O fetch_hn_discussion() from async def main() blocks the event loop; either convert fetch_hn_discussion to an async function and await it from main, or keep it synchronous but call it off the event loop using asyncio.to_thread or loop.run_in_executor (e.g., replace messages = fetch_hn_discussion() with messages = await asyncio.to_thread(fetch_hn_discussion) and import asyncio). Update the function signature and all call sites if you choose to make fetch_hn_discussion async, or add the to_thread/run_in_executor wrapper in main to offload blocking I/O.
33-36: ⚡ Quick winUse transcript-accurate roles for imported HN replies.
Lines 34-35 mark every HN reply as
"assistant", but these are third-party commenters. This can skew role-aware extraction/retrieval behavior in the agent example.Proposed patch
messages.append({ - "role": "assistant", - "content": kid_data['text'][:500] + "role": "user", + "content": kid_data['text'][:500] })🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@examples/web_transcript_case.py` around lines 33 - 36, The code currently tags every imported HN reply as "assistant" in the messages.append call; change the role to reflect third‑party commenters (e.g., "user") so imported replies are transcript‑accurate; update the messages.append call that constructs the dict (refer to messages.append and kid_data['text'][:500]) to set "role": "user" (or another appropriate non-assistant role) while keeping the existing content truncation.examples/real_world_support_case.py (2)
56-58: ⚡ Quick winHarden memory debug access against missing metadata.
Line 57 and Line 58 assume
result.memories_usedis always a dict. A small defensive fallback avoids example crashes when retrieval metadata is absent.Proposed change
def _print_memories(result) -> None: - facts = result.memories_used.get("facts", []) - episodes = result.memories_used.get("episodes", []) + memories_used = getattr(result, "memories_used", None) or {} + facts = memories_used.get("facts", []) + episodes = memories_used.get("episodes", [])🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@examples/real_world_support_case.py` around lines 56 - 58, The _print_memories function assumes result.memories_used is always a dict; guard against missing or None metadata by coercing result.memories_used to a dict before accessing keys (e.g., use memories = result.memories_used or {} / dict(getattr(result, "memories_used", {}) ) ), then read facts = memories.get("facts", []) and episodes = memories.get("episodes", []) so the example won’t crash when metadata is absent; update references in _print_memories accordingly.
129-140: ⚡ Quick winEnsure
agent.close()always executes on failures.Line 140 is on the happy path only. Wrap the seeded run sequence in
try/finallyso cleanup is guaranteed.Proposed change
- print("Seeding the support case transcript...") - await agent.add_messages(SUPPORT_CASE_MESSAGES) - - await _run_turn(agent, "Call me Amina.") - await _run_turn(agent, "Keep your answers short.") - await _run_turn( - agent, - "What is the customer's preferred contact channel, and what is still blocked?", - ) - await _run_turn(agent, "What should I tell them next?") - - await agent.close() + try: + print("Seeding the support case transcript...") + await agent.add_messages(SUPPORT_CASE_MESSAGES) + + await _run_turn(agent, "Call me Amina.") + await _run_turn(agent, "Keep your answers short.") + await _run_turn( + agent, + "What is the customer's preferred contact channel, and what is still blocked?", + ) + await _run_turn(agent, "What should I tell them next?") + finally: + await agent.close()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@examples/real_world_support_case.py` around lines 129 - 140, The seeded run sequence that calls agent.add_messages(SUPPORT_CASE_MESSAGES) and multiple _run_turn(...) calls must ensure agent.close() runs even on errors; wrap the block that seeds and runs the turns in a try/finally where agent.close() is called in the finally so cleanup always occurs, keeping existing calls to agent.add_messages and _run_turn intact and not changing their order.
🤖 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/gemini_agent_ui.py`:
- Around line 331-333: The run method currently blocks forever on
future.result(); change run(self, coro, timeout=30) to call future =
asyncio.run_coroutine_threadsafe(coro, self.loop) and use
future.result(timeout=timeout) so the wait is bounded, catch
concurrent.futures.TimeoutError, call future.cancel() to stop the task, and
re-raise or return an appropriate error/exception so HTTP request threads aren’t
pinned; reference the run method and the future variable when making these
changes.
- Around line 521-523: The except block currently returns raw exception text to
clients by calling self._json({"error": str(e)},
HTTPStatus.INTERNAL_SERVER_ERROR); change this to log the full exception
(including stack/traceback) on the server using the existing logger or traceback
module and return a generic error payload to the client (e.g.,
{"error":"internal server error"}) with HTTPStatus.INTERNAL_SERVER_ERROR; keep
the exception variable e and the self._json call but replace the returned
message with a non-sensitive generic string and add an internal log statement
that captures the exception details.
- Around line 484-488: The _read_json method can raise json.JSONDecodeError and
currently bubbles up into a 500 handler; wrap the json.loads call in a
try/except that catches json.JSONDecodeError, call self.send_error(400, "Invalid
JSON") (or otherwise send a 400 response) and return an empty dict (or abort
processing) so malformed JSON returns 400 instead of propagating as a 500;
reference _read_json, json.loads, json.JSONDecodeError and self.send_error when
making the change.
In `@examples/web_transcript_case.py`:
- Around line 9-13: The get_hacker_news_item function makes a blocking external
call without timeout or error handling; wrap the urllib.request.urlopen call in
a try/except, pass a reasonable timeout (e.g., timeout=5) to urlopen, catch
urllib.error.URLError/HTTPError, socket.timeout and json.JSONDecodeError, log or
handle the error, and return None (or re-raise a clearer exception) on failure
so the caller won’t hang or crash; update references inside get_hacker_news_item
to use the timeout and explicit exception handling around
response.read()/json.loads.
In `@vektori/integrations/hermes.py`:
- Around line 47-48: The loop in hermes.py currently only reads
results.get("episodes", []) and drops hits when that key is missing; update the
parsing in the for loop that builds memories (the block iterating over episodes
and appending {"type": "episode", "content": ep["text"], "score": 1.0}) to fall
back to results.get("sentences", []) (or iterate episodes if present else
sentences) so sentence-level hits are included; ensure you account for the same
"text" key shape when using sentences and keep the memory type as "episode" for
compatibility.
- Around line 27-31: The memory write call to Vektori is missing the required
session_id argument: update the call to self.memory.add in the Hermes
integration (the line with result = await self.memory.add(...)) to pass the
current session identifier (e.g., session_id=self.session_id or the appropriate
session variable on the Hermes instance), leaving the existing messages, user_id
and metadata arguments intact so the call matches the Vektori client API.
- Around line 52-55: The clear() coroutine currently no-ops and therefore
incorrectly reports success; update hermes.py's async def clear(self) to either
perform the actual deletion by invoking the integration's deletion API/storage
routine (e.g., call self.client.delete_user_memory(self.user_id) or
self.storage.delete_user_data/clear for the current user) and await it, handling
and logging errors, or if no delete endpoint exists, explicitly raise
NotImplementedError (or a clear RuntimeError) and update the docstring to state
that deletion is unsupported; ensure you reference the clear method and the
integration client/storage attribute used to perform the deletion so callers no
longer get a silent success.
In `@vektori/integrations/openclaw.py`:
- Around line 41-44: search_vektori is hard-coded to "default_openclaw_user"
which breaks per-user isolation; modify search_vektori to accept or derive the
caller's user id and pass that to self.memory.search instead of the fixed string
— e.g., add a user_id parameter (user_id: Optional[str] = None) or read the same
context user used when writing (self.context.user_id or context.user.id) and
call self.memory.search(query=query, user_id=user_id_to_use, depth=depth);
update any callers to pass the current user where needed so retrieval uses the
same user-scoped memory as the writes.
- Around line 47-48: The current rendering only includes results["episodes"] and
drops sentence-level hits; update the block that builds the "Insights" output
(the code using results, episodes and output.append) to fall back to
results.get("sentences") when episodes is empty (or include both if desired):
detect if results.get("episodes") is truthy, otherwise use
results.get("sentences") and join each sentence's text into the same "- {text}"
list before appending the "Insights:\n" string; reference the same results
variable and the output.append call so the change is localized to that rendering
logic.
---
Nitpick comments:
In `@examples/real_world_support_case.py`:
- Around line 56-58: The _print_memories function assumes result.memories_used
is always a dict; guard against missing or None metadata by coercing
result.memories_used to a dict before accessing keys (e.g., use memories =
result.memories_used or {} / dict(getattr(result, "memories_used", {}) ) ), then
read facts = memories.get("facts", []) and episodes = memories.get("episodes",
[]) so the example won’t crash when metadata is absent; update references in
_print_memories accordingly.
- Around line 129-140: The seeded run sequence that calls
agent.add_messages(SUPPORT_CASE_MESSAGES) and multiple _run_turn(...) calls must
ensure agent.close() runs even on errors; wrap the block that seeds and runs the
turns in a try/finally where agent.close() is called in the finally so cleanup
always occurs, keeping existing calls to agent.add_messages and _run_turn intact
and not changing their order.
In `@examples/web_transcript_case.py`:
- Line 42: The call to synchronous network I/O fetch_hn_discussion() from async
def main() blocks the event loop; either convert fetch_hn_discussion to an async
function and await it from main, or keep it synchronous but call it off the
event loop using asyncio.to_thread or loop.run_in_executor (e.g., replace
messages = fetch_hn_discussion() with messages = await
asyncio.to_thread(fetch_hn_discussion) and import asyncio). Update the function
signature and all call sites if you choose to make fetch_hn_discussion async, or
add the to_thread/run_in_executor wrapper in main to offload blocking I/O.
- Around line 33-36: The code currently tags every imported HN reply as
"assistant" in the messages.append call; change the role to reflect third‑party
commenters (e.g., "user") so imported replies are transcript‑accurate; update
the messages.append call that constructs the dict (refer to messages.append and
kid_data['text'][:500]) to set "role": "user" (or another appropriate
non-assistant role) while keeping the existing content truncation.
In `@tests/unit/test_factory.py`:
- Around line 70-84: Add tests that call create_chat_model with provider-only
strings to exercise the branch where model_name is None: call
create_chat_model("litellm") and assert isinstance(..., LiteLLMChatModel) and
that model.model is a non-empty string (truthy) to verify the provider default
is used; do the same for create_chat_model("openai") asserting isinstance(...,
OpenAIChatModel) and that model.model is truthy.
🪄 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: 0835ea16-25d9-4eb8-8772-8489d8831750
📒 Files selected for processing (11)
README.mdexamples/gemini_agent_integration_demo.pyexamples/gemini_agent_ui.pyexamples/integrating_hermes_openclaw.pyexamples/real_world_support_case.pyexamples/vektori_agent_demo.pyexamples/web_transcript_case.pyscripts/test_agent_e2e.pytests/unit/test_factory.pyvektori/integrations/hermes.pyvektori/integrations/openclaw.py
✅ Files skipped from review due to trivial changes (1)
- examples/vektori_agent_demo.py
🚧 Files skipped from review as they are similar to previous changes (2)
- README.md
- scripts/test_agent_e2e.py
Comment on lines
+331
to
+333
| def run(self, coro): | ||
| future = asyncio.run_coroutine_threadsafe(coro, self.loop) | ||
| return future.result() |
There was a problem hiding this comment.
Request thread can block indefinitely on model/retrieval work.
Line 333 waits forever on future.result(); a slow/downstream hang can pin HTTP worker threads and degrade availability.
Proposed fix
+from concurrent.futures import TimeoutError as FutureTimeoutError
@@
- def run(self, coro):
+ def run(self, coro, *, timeout_s: float = 60):
future = asyncio.run_coroutine_threadsafe(coro, self.loop)
- return future.result()
+ try:
+ return future.result(timeout=timeout_s)
+ except FutureTimeoutError:
+ future.cancel()
+ raise TimeoutError(f"Operation timed out after {timeout_s}s")📝 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
| def run(self, coro): | |
| future = asyncio.run_coroutine_threadsafe(coro, self.loop) | |
| return future.result() | |
| from concurrent.futures import TimeoutError as FutureTimeoutError | |
| def run(self, coro, *, timeout_s: float = 60): | |
| future = asyncio.run_coroutine_threadsafe(coro, self.loop) | |
| try: | |
| return future.result(timeout=timeout_s) | |
| except FutureTimeoutError: | |
| future.cancel() | |
| raise TimeoutError(f"Operation timed out after {timeout_s}s") |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@examples/gemini_agent_ui.py` around lines 331 - 333, The run method currently
blocks forever on future.result(); change run(self, coro, timeout=30) to call
future = asyncio.run_coroutine_threadsafe(coro, self.loop) and use
future.result(timeout=timeout) so the wait is bounded, catch
concurrent.futures.TimeoutError, call future.cancel() to stop the task, and
re-raise or return an appropriate error/exception so HTTP request threads aren’t
pinned; reference the run method and the future variable when making these
changes.
Comment on lines
+484
to
+488
| def _read_json(self) -> dict: | ||
| length = int(self.headers.get("Content-Length", "0")) | ||
| raw = self.rfile.read(length) if length > 0 else b"{}" | ||
| return json.loads(raw.decode("utf-8")) | ||
|
|
There was a problem hiding this comment.
Invalid JSON should return 400 instead of bubbling as 500.
Line 487 can raise JSONDecodeError; currently this lands in generic 500 handling.
Proposed fix
def _read_json(self) -> dict:
length = int(self.headers.get("Content-Length", "0"))
raw = self.rfile.read(length) if length > 0 else b"{}"
- return json.loads(raw.decode("utf-8"))
+ try:
+ return json.loads(raw.decode("utf-8"))
+ except json.JSONDecodeError as exc:
+ raise ValueError("invalid JSON body") from exc
@@
- except TimeoutError:
+ except ValueError as exc:
+ self._json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
+ except TimeoutError:
self._json({"error": "request timed out"}, HTTPStatus.GATEWAY_TIMEOUT)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@examples/gemini_agent_ui.py` around lines 484 - 488, The _read_json method
can raise json.JSONDecodeError and currently bubbles up into a 500 handler; wrap
the json.loads call in a try/except that catches json.JSONDecodeError, call
self.send_error(400, "Invalid JSON") (or otherwise send a 400 response) and
return an empty dict (or abort processing) so malformed JSON returns 400 instead
of propagating as a 500; reference _read_json, json.loads, json.JSONDecodeError
and self.send_error when making the change.
Comment on lines
+521
to
+523
| except Exception as e: | ||
| self._json({"error": str(e)}, HTTPStatus.INTERNAL_SERVER_ERROR) | ||
|
|
There was a problem hiding this comment.
Internal exception details are exposed to clients.
Line 522 sends raw exception text in API responses, which can leak internal state and stack-context clues.
Proposed fix
+import logging
@@
+logger = logging.getLogger(__name__)
@@
- except Exception as e:
- self._json({"error": str(e)}, HTTPStatus.INTERNAL_SERVER_ERROR)
+ except TimeoutError:
+ self._json({"error": "request timed out"}, HTTPStatus.GATEWAY_TIMEOUT)
+ except Exception:
+ logger.exception("Unhandled error in %s", self.path)
+ self._json({"error": "internal server error"}, HTTPStatus.INTERNAL_SERVER_ERROR)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@examples/gemini_agent_ui.py` around lines 521 - 523, The except block
currently returns raw exception text to clients by calling self._json({"error":
str(e)}, HTTPStatus.INTERNAL_SERVER_ERROR); change this to log the full
exception (including stack/traceback) on the server using the existing logger or
traceback module and return a generic error payload to the client (e.g.,
{"error":"internal server error"}) with HTTPStatus.INTERNAL_SERVER_ERROR; keep
the exception variable e and the self._json call but replace the returned
message with a non-sensitive generic string and add an internal log statement
that captures the exception details.
Comment on lines
+9
to
+13
| def get_hacker_news_item(item_id): | ||
| url = f"https://hacker-news.firebaseio.com/v0/item/{item_id}.json" | ||
| req = urllib.request.Request(url) | ||
| with urllib.request.urlopen(req) as response: | ||
| return json.loads(response.read().decode()) |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
# Check if the file exists and examine the specific lines
cd "$(git rev-parse --show-toplevel 2>/dev/null || pwd)" && \
if [ -f "examples/web_transcript_case.py" ]; then
echo "=== File content (lines 1-50) ==="; \
head -50 "examples/web_transcript_case.py" | cat -n
else
echo "File not found. Checking if examples directory exists:"
find . -type d -name "examples" 2>/dev/null | head -5
find . -name "*transcript*" -type f 2>/dev/null | head -10
fiRepository: vektori-ai/vektori
Length of output: 2099
🌐 Web query:
urllib.request.urlopen exception handling timeout URLError TimeoutError
💡 Result:
To handle exceptions from urllib.request.urlopen related to timeouts, use a try-except block catching URLError (which wraps timeout conditions) and optionally check its reason attribute for socket.timeout. Key points: - Pass a timeout parameter to urlopen to set seconds before timeout (e.g., timeout=10). This applies to HTTP/HTTPS/FTP connections; uses global default if omitted. - urlopen raises URLError on protocol errors, including timeouts. - Timeout manifests as URLError where error.reason is a socket.timeout instance (import socket). - Also catch HTTPError for HTTP status errors. - No separate TimeoutError from concurrent.futures; it's socket.timeout wrapped in URLError. Example code: import urllib.request import urllib.error import socket try: with urllib.request.urlopen('https://example.com', timeout=10) as response: data = response.read except urllib.error.HTTPError as e: print(f'HTTP Error {e.code}: {e.reason}') except urllib.error.URLError as e: if isinstance(e.reason, socket.timeout): print('Request timed out') else: print(f'URL error: {e.reason}') except Exception as e: print(f'Unexpected error: {e}') This handles timeouts gracefully without crashing. For global timeout, use socket.setdefaulttimeout(10).
Citations:
- 1: https://docs.python.org/3/library/urllib.request.html?highlight=urlopen
- 2: https://docs.python.org/3.12/library/urllib.request.html
- 3: https://stackoverflow.com/questions/8763451/how-to-handle-urllibs-timeout-in-python-3
- 4: https://docs.python.org/3/library/urllib.error.html
- 5: https://stackoverflow.com/questions/43652951/urllib-request-fails-when-page-takes-too-long-to-respond
- 6: https://docs.python.org/3/library/urllib.request.html
Add timeout + error handling for HN API calls.
Lines 12–13 perform an external HTTP request without a timeout and without handling network/parsing errors, risking hangs or crashes.
Proposed patch
import asyncio
import urllib.request
+import urllib.error
import json
import os
@@
def get_hacker_news_item(item_id):
url = f"https://hacker-news.firebaseio.com/v0/item/{item_id}.json"
req = urllib.request.Request(url)
- with urllib.request.urlopen(req) as response:
- return json.loads(response.read().decode())
+ try:
+ with urllib.request.urlopen(req, timeout=10) as response:
+ return json.loads(response.read().decode())
+ except (urllib.error.URLError, json.JSONDecodeError):
+ return None🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@examples/web_transcript_case.py` around lines 9 - 13, The
get_hacker_news_item function makes a blocking external call without timeout or
error handling; wrap the urllib.request.urlopen call in a try/except, pass a
reasonable timeout (e.g., timeout=5) to urlopen, catch
urllib.error.URLError/HTTPError, socket.timeout and json.JSONDecodeError, log or
handle the error, and return None (or re-raise a clearer exception) on failure
so the caller won’t hang or crash; update references inside get_hacker_news_item
to use the timeout and explicit exception handling around
response.read()/json.loads.
Comment on lines
+27
to
+31
| result = await self.memory.add( | ||
| messages=[{"role": "user", "content": text}], | ||
| user_id=self.user_id, | ||
| metadata=metadata | ||
| ) |
There was a problem hiding this comment.
Missing required session_id in memory write path.
Line 27 calls Vektori.add(...) without session_id, but that argument is required by the current client API. This will fail at runtime and drop Hermes memory writes.
Proposed fix
+from uuid import uuid4
@@
async def add_memory(self, text: str, metadata: Dict[str, Any] = None) -> str:
"""Store a new fact/episode from the Hermes agent into Vektori."""
+ metadata = metadata or {}
+ session_id = str(metadata.get("session_id", f"hermes-{uuid4()}"))
result = await self.memory.add(
messages=[{"role": "user", "content": text}],
+ session_id=session_id,
user_id=self.user_id,
metadata=metadata
)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/integrations/hermes.py` around lines 27 - 31, The memory write call
to Vektori is missing the required session_id argument: update the call to
self.memory.add in the Hermes integration (the line with result = await
self.memory.add(...)) to pass the current session identifier (e.g.,
session_id=self.session_id or the appropriate session variable on the Hermes
instance), leaving the existing messages, user_id and metadata arguments intact
so the call matches the Vektori client API.
Comment on lines
+47
to
+48
| for ep in results.get("episodes", []): | ||
| memories.append({"type": "episode", "content": ep["text"], "score": 1.0}) |
There was a problem hiding this comment.
Episode parsing should fall back to sentences for compatibility.
Line 47 only reads episodes; if that list is absent/empty, relevant sentence hits are dropped from Hermes context.
Proposed fix
- for ep in results.get("episodes", []):
+ for ep in (results.get("episodes", []) or results.get("sentences", [])):
memories.append({"type": "episode", "content": ep["text"], "score": 1.0})📝 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
| for ep in results.get("episodes", []): | |
| memories.append({"type": "episode", "content": ep["text"], "score": 1.0}) | |
| for ep in (results.get("episodes", []) or results.get("sentences", [])): | |
| memories.append({"type": "episode", "content": ep["text"], "score": 1.0}) |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/integrations/hermes.py` around lines 47 - 48, The loop in hermes.py
currently only reads results.get("episodes", []) and drops hits when that key is
missing; update the parsing in the for loop that builds memories (the block
iterating over episodes and appending {"type": "episode", "content": ep["text"],
"score": 1.0}) to fall back to results.get("sentences", []) (or iterate episodes
if present else sentences) so sentence-level hits are included; ensure you
account for the same "text" key shape when using sentences and keep the memory
type as "episode" for compatibility.
Comment on lines
+52
to
+55
| async def clear(self) -> None: | ||
| """Clear memory for the current user.""" | ||
| # Vektori users would need an explicit delete endpoint, | ||
| pass |
There was a problem hiding this comment.
clear() currently reports success but does nothing.
Line 52 advertises clearing user memory, but Line 55 is a no-op. This can create retention/compliance surprises for callers.
Proposed fix
async def clear(self) -> None:
"""Clear memory for the current user."""
- # Vektori users would need an explicit delete endpoint,
- pass
+ raise NotImplementedError(
+ "clear() is not implemented for VektoriHermesMemory; add a delete endpoint first."
+ )🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/integrations/hermes.py` around lines 52 - 55, The clear() coroutine
currently no-ops and therefore incorrectly reports success; update hermes.py's
async def clear(self) to either perform the actual deletion by invoking the
integration's deletion API/storage routine (e.g., call
self.client.delete_user_memory(self.user_id) or
self.storage.delete_user_data/clear for the current user) and await it, handling
and logging errors, or if no delete endpoint exists, explicitly raise
NotImplementedError (or a clear RuntimeError) and update the docstring to state
that deletion is unsupported; ensure you reference the clear method and the
integration client/storage attribute used to perform the deletion so callers no
longer get a silent success.
Comment on lines
+41
to
+44
| async def search_vektori(query: str, depth: str = "l1") -> str: | ||
| """Search the Vektori long-term graph memory.""" | ||
| results = await self.memory.search(query=query, user_id="default_openclaw_user", depth=depth) | ||
| output = [] |
There was a problem hiding this comment.
Retrieval is pinned to a fixed user, breaking per-user memory isolation.
Line 43 always searches default_openclaw_user, while writes in Line 29 are scoped by context user. In multi-user runs, tool retrieval will show the wrong memory set.
Proposed fix
class VektoriOpenClawPlugin(OpenClawPlugin):
@@
- def __init__(self, vektori_instance: Vektori):
+ def __init__(self, vektori_instance: Vektori, default_user_id: str = "default_openclaw_user"):
self.memory = vektori_instance
+ self.default_user_id = default_user_id
@@
- async def search_vektori(query: str, depth: str = "l1") -> str:
+ async def search_vektori(query: str, depth: str = "l1", user_id: str | None = None) -> str:
"""Search the Vektori long-term graph memory."""
- results = await self.memory.search(query=query, user_id="default_openclaw_user", depth=depth)
+ results = await self.memory.search(
+ query=query,
+ user_id=user_id or self.default_user_id,
+ depth=depth,
+ )🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/integrations/openclaw.py` around lines 41 - 44, search_vektori is
hard-coded to "default_openclaw_user" which breaks per-user isolation; modify
search_vektori to accept or derive the caller's user id and pass that to
self.memory.search instead of the fixed string — e.g., add a user_id parameter
(user_id: Optional[str] = None) or read the same context user used when writing
(self.context.user_id or context.user.id) and call
self.memory.search(query=query, user_id=user_id_to_use, depth=depth); update any
callers to pass the current user where needed so retrieval uses the same
user-scoped memory as the writes.
Comment on lines
+47
to
+48
| if results.get("episodes"): | ||
| output.append("Insights:\n" + "\n".join(f"- {ep['text']}" for ep in results["episodes"])) |
There was a problem hiding this comment.
Evidence rendering should include sentence fallback when episodes are empty.
Line 47 only checks episodes; sentence hits can be silently omitted from tool output.
Proposed fix
- if results.get("episodes"):
- output.append("Insights:\n" + "\n".join(f"- {ep['text']}" for ep in results["episodes"]))
+ episodes = results.get("episodes", []) or results.get("sentences", [])
+ if episodes:
+ output.append("Insights:\n" + "\n".join(f"- {ep['text']}" for ep in episodes))🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@vektori/integrations/openclaw.py` around lines 47 - 48, The current rendering
only includes results["episodes"] and drops sentence-level hits; update the
block that builds the "Insights" output (the code using results, episodes and
output.append) to fall back to results.get("sentences") when episodes is empty
(or include both if desired): detect if results.get("episodes") is truthy,
otherwise use results.get("sentences") and join each sentence's text into the
same "- {text}" list before appending the "Insights:\n" string; reference the
same results variable and the output.append call so the change is localized to
that rendering logic.
Content conflicts resolved: - README.md: keep branch's extended examples list - vektori/client.py: keep skip_extraction param, adopt result= style from main - vektori/retrieval/scoring.py: use already-parsed meta dict (main's approach) - tests/unit/test_factory.py: combine both import sets and test suites Restored files dropped by auto-merge: - vektori/context.py (AgentContextLoader used by agent harness) - ChatModelProvider + ChatCompletionResult in vektori/models/base.py - LiteLLMChatModel in vektori/models/litellm_provider.py - OpenAIChatModel in vektori/models/openai.py - CHAT_REGISTRY + create_chat_model in vektori/models/factory.py
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 by CodeRabbit