JuliusBrussee/caveman · error · NotImplementedError

Caveman delegates through public chat_with_tools methods

Error message

Caveman delegates through public chat_with_tools methods

What it means

_prepare_chat_with_tools is the abstract extension hook of the LlamaIndex LLM contract; caveman's wrapper intentionally does not implement it because all public chat_with_tools/astream methods delegate to the wrapped LLM. Calling it raises NotImplementedError by design.

Solutions

1. Use the public chat_with_tools / astream_chat_with_tools / stream_chat_with_tools methods instead of the private hook
2. Build your agent loop on top of the public tool-calling API the wrapper delegates
3. If you need custom preparation logic, subclass the native LLM, not the caveman wrapper

Example fix

// before
prepared = caveman_llm._prepare_chat_with_tools(tools, user_msg=msg)
// after
response = caveman_llm.chat_with_tools(tools, user_msg=msg)

if hasattr(fn, "__name__") and fn.__name__ == "_prepare_chat_with_tools":
    raise RuntimeError("use the public chat_with_tools API")

def uses_public_tool_api(obj) -> bool:
    return callable(getattr(obj, "chat_with_tools", None)) or callable(getattr(obj, "astream_chat_with_tools", None))

try:
    return llm._prepare_chat_with_tools(*args, **kwargs)
except NotImplementedError:
    return llm.chat_with_tools(*args, **kwargs)

When it happens

Trigger: Code that calls the wrapper's _prepare_chat_with_tools directly, or a LlamaIndex internal/agent path that relies on the protected hook instead of the public methods.

Common situations: Custom agents or subclasses invoking the private hook, framework upgrades that changed the delegation flow, or tests probing the abstract contract.

Understand the failure class

Background: "NotImplementedError: Subclasses should override this method" / "must be implemented" — abstract method errors explained — this error's family across 40 libraries.

Related errors

• cave_async_jobs_unavailable
• Caveman recovery metadata is immutable
• Caveman recovery tool is immutable
• Expected a native ToolSelection
• Expected an existing native LlamaIndex LLM

AI-assisted analysis of JuliusBrussee/caveman@3ee70a1026 (2026-09-20). Data as JSON: /api/errors/1d3c6cf1e3300f19. Report an issue: GitHub.

Appendix: source

Thrown at packages/middleware/python/caveman_middleware/llama_index.py:628

    async def astream_chat(self, messages, **kwargs):
        return self._astream(self.wrapped.astream_chat, messages, None, kwargs)

    def chat_with_tools(self, tools, user_msg=None, chat_history=None, **kwargs):
        return self._call(self.wrapped.chat_with_tools, chat_history or [], tools, {"user_msg": user_msg, **kwargs})

    async def achat_with_tools(self, tools, user_msg=None, chat_history=None, **kwargs):
        return await self._acall(self.wrapped.achat_with_tools, chat_history or [], tools, {"user_msg": user_msg, **kwargs})

    def stream_chat_with_tools(self, tools, user_msg=None, chat_history=None, **kwargs):
        return self._stream(self.wrapped.stream_chat_with_tools, chat_history or [], tools, {"user_msg": user_msg, **kwargs})

    async def astream_chat_with_tools(self, tools, user_msg=None, chat_history=None, **kwargs):
        return self._astream(self.wrapped.astream_chat_with_tools, chat_history or [], tools, {"user_msg": user_msg, **kwargs})

    def _prepare_chat_with_tools(self, *args, **kwargs):
        # Abstract extension contract only; public tool methods delegate above.
        raise NotImplementedError("Caveman delegates through public chat_with_tools methods")

    def get_tool_calls_from_response(self, response, **kwargs):
        return self.wrapped.get_tool_calls_from_response(response, **kwargs)

    def complete(self, prompt, formatted=False, **kwargs):
        return self._call(self.wrapped.complete, prompt, None, {"formatted": formatted, **kwargs}, passthrough="no_candidate")

    async def acomplete(self, prompt, formatted=False, **kwargs):
        return await self._acall(self.wrapped.acomplete, prompt, None, {"formatted": formatted, **kwargs}, passthrough="no_candidate")

    def stream_complete(self, prompt, formatted=False, **kwargs):
        return self._passive_stream(self.wrapped.stream_complete, (prompt,), {"formatted": formatted, **kwargs}, "no_candidate")

    async def astream_complete(self, prompt, formatted=False, **kwargs):
        return await self._passive_astream(self.wrapped.astream_complete, (prompt,), {"formatted": formatted, **kwargs}, "no_candidate")

    def structured_predict(self, *args, **kwargs):
        return self._call(lambda _: self.wrapped.structured_predict(*args, **kwargs), None, None, {}, passthrough="structured_output")

View on GitHub (pinned to 3ee70a1026)