Retriever callbacks not triggered with Qdrant client

[lukasschoettler]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

import os
from uuid import UUID
from typing import Any, Generator, List, Union, Dict, Sequence
from collections import defaultdict
from langchain.callbacks.base import BaseCallbackHandler
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableLambda, RunnableParallel
from langchain_ollama import ChatOllama
from langchain_qdrant import QdrantVectorStore, FastEmbedSparse, RetrievalMode
from langchain_community.embeddings.fastembed import FastEmbedEmbeddings
from qdrant_client import QdrantClient # type: ignore
from qdrant_client.models import VectorParams, SparseVectorParams, Distance # type: ignore
import streamlit as st # type: ignore
from operator import itemgetter

# Environment
vectordb_host = os.getenv("VECTORDB_CONTAINER_NAME")
vectordb_port = os.getenv("VECTOR_DB_PORT")
ollama_url = os.getenv("OLLAMA_BASE_URL")
ollama_llm = os.getenv("OLLAMA_LLM")
fastembed_embedding_model = os.getenv("FASTEMBED_EMBEDDING_MODEL")
fastembed_sparse_model = os.getenv("FASTEMBED_SPARSE_MODEL")

class CustomCallbackHandler(BaseCallbackHandler):
    def __init__(self):
        self.docs = []
        self.search_string = ""

    def on_chain_end(self, _outputs: Dict[str, Any], _inputs: Dict[str, Any], **_kwargs) -> None:
        print("Event on_chain_end triggered") # This works

    def on_chain_error(self, **_kwargs) -> None:
        print("Event on_chain_error triggered")

    def on_retriever_start(self, _serialized: Dict[str, Any], query: str, *_args: Any, **_kwargs: Any) -> Any:
        print("Event on_retriever_start triggered") # Doesn't work

    def on_retriever_end(self, documents: Sequence[Document], **_kwargs: Any) -> None:
        print("Event on_retriever_end triggered")

    def on_retriever_error(self, _error: BaseException, *, _run_id: UUID, _parent_run_id: UUID | None = None, **_kwargs: Any) -> Any:
        print("Event on_retriever_error triggered")

chat_llm = ChatOllama(base_url=ollama_url,
                      model=ollama_llm,
                      temperature=0.6,
                      repetition_penalty=1.1)

embeddings = FastEmbedEmbeddings(model_name = fastembed_embedding_model)
sparse_embeddings = FastEmbedSparse(model_name = fastembed_sparse_model)

qdrant_client = QdrantClient(host = vectordb_host, port = vectordb_port)

qdrant_client.set_model(fastembed_embedding_model)
qdrant_client.set_sparse_model(fastembed_sparse_model)

vector_store = QdrantVectorStore(
    client = qdrant_client,
    collection_name = "langchain",
    vector_name = qdrant_client.get_vector_field_name(),
    sparse_vector_name = qdrant_client.get_sparse_vector_field_name(),
    embedding = embeddings,
    sparse_embedding = sparse_embeddings,
    distance = Distance.COSINE,
    retrieval_mode = RetrievalMode.HYBRID,
)

retriever = vector_store.as_retriever()
output_parser = StrOutputParser()

chat_system_template = """You are an AI assistant."""

chat_human_template = """Please answer my question, using
1. Chathistory: {chat_history}
2. Provided Documents: {context}
3. My Question: {user_question}"""

chat_system_prompt = SystemMessagePromptTemplate.from_template(chat_system_template)
chat_human_prompt = HumanMessagePromptTemplate.from_template(chat_human_template)
chat_prompt = ChatPromptTemplate.from_messages([chat_system_prompt, chat_human_prompt])

chain = (
    RunnableParallel({
        "context": retriever,
        "user_question": itemgetter("user_question"),
        "chat_history": itemgetter("chat_history")
    }) |
    chat_prompt |
    chat_llm |
    output_parser
)

callback = CustomCallbackHandler()

def get_answer_stream(user_query: str, chat_history: List[Union[HumanMessage, AIMessage]]) -> Generator[str, None, None]:
    for response in chain.stream(
        {
            "chat_history": chat_history,
            "user_question": user_query,
        },
        config={"callbacks": [callback]}
    ):
        yield response

# Generator is used inside a streamlit application.

Description

I have a RAG chain and use a CustomCallbackHandler to retrieve intermediate outputs, like the documents retrieved by the retriever or a rewritten query send to the retriever. The chain works, the result and API calls prove that the retriever is working as expected. Yet, the on_retriever_start, or on_retriever_end events seem to not be triggered. I don't know, whether this is an error in my implementation of the methods or a bug.

I know, that propagation of callbacks to child elements is influenced by how you attach the callback handler, but I believe to have done this correctly, and I use python 3.11 anyways, where propagation should be less of a concern, according to the docs.

System Info

"python -m langchain_core.sys_info"

System Information

OS: Linux
OS Version: #1 SMP Fri Mar 29 23:14:13 UTC 2024
Python Version: 3.11.10 (main, Nov 12 2024, 06:03:56) [GCC 12.2.0]

Package Information

langchain_core: 0.3.21
langchain: 0.3.9
langchain_community: 0.3.8
langsmith: 0.1.147
langchain_ollama: 0.2.0
langchain_qdrant: 0.2.0
langchain_text_splitters: 0.3.2
langchain_unstructured: 0.1.6

Optional packages not installed

langserve

Other Dependencies

aiohttp: 3.11.9
async-timeout: Installed. No version info available.
dataclasses-json: 0.6.7
fastembed: 0.3.6
httpx: 0.27.2
httpx-sse: 0.4.0
jsonpatch: 1.33
langsmith-pyo3: Installed. No version info available.
numpy: 1.26.4
ollama: 0.4.2
onnxruntime: 1.19.2
orjson: 3.10.12
packaging: 24.2
pydantic: 2.9.2
pydantic-settings: 2.6.1
PyYAML: 6.0.2
qdrant-client: 1.12.1
requests: 2.32.3
requests-toolbelt: 1.0.0
SQLAlchemy: 2.0.35
tenacity: 9.0.0
typing-extensions: 4.12.2
unstructured-client: 0.28.1
unstructured[all-docs]: Installed. No version info available.

"pip freeze"
aiofiles==24.1.0
aiohappyeyeballs==2.4.4
aiohttp==3.11.9
aiosignal==1.3.1
altair==5.5.0
annotated-types==0.7.0
antlr4-python3-runtime==4.9.3
anyio==4.6.2.post1
attrs==24.2.0
backoff==2.2.1
beautifulsoup4==4.12.3
blinker==1.9.0
cachetools==5.5.0
certifi==2024.8.30
cffi==1.17.1
chardet==5.2.0
charset-normalizer==3.4.0
click==8.1.7
coloredlogs==15.0.1
contourpy==1.3.1
cryptography==44.0.0
cycler==0.12.1
dataclasses-json==0.6.7
Deprecated==1.2.15
effdet==0.4.1
emoji==2.14.0
et_xmlfile==2.0.0
eval_type_backport==0.2.0
fastembed==0.3.6
filelock==3.16.1
filetype==1.2.0
flatbuffers==24.3.25
fonttools==4.55.0
frozenlist==1.5.0
fsspec==2024.10.0
gitdb==4.0.11
GitPython==3.1.43
google-api-core==2.23.0
google-auth==2.36.0
google-cloud-vision==3.8.1
googleapis-common-protos==1.66.0
greenlet==3.1.1
grpcio==1.68.1
grpcio-status==1.68.1
grpcio-tools==1.68.1
h11==0.14.0
h2==4.1.0
hpack==4.0.0
html5lib==1.1
httpcore==1.0.7
httpx==0.27.2
httpx-sse==0.4.0
huggingface-hub==0.26.3
humanfriendly==10.0
hyperframe==6.0.1
idna==3.10
iopath==0.1.10
Jinja2==3.1.4
joblib==1.4.2
jsonpatch==1.33
jsonpath-python==1.0.6
jsonpointer==3.0.0
jsonschema==4.23.0
jsonschema-specifications==2024.10.1
kiwisolver==1.4.7
langchain==0.3.9
langchain-community==0.3.8
langchain-core==0.3.21
langchain-ollama==0.2.0
langchain-qdrant==0.2.0
langchain-text-splitters==0.3.2
langchain-unstructured==0.1.6
langdetect==1.0.9
langsmith==0.1.147
layoutparser==0.3.4
loguru==0.7.2
lxml==5.3.0
Markdown==3.7
markdown-it-py==3.0.0
MarkupSafe==3.0.2
marshmallow==3.23.1
matplotlib==3.9.3
mdurl==0.1.2
mmh3==4.1.0
mpmath==1.3.0
multidict==6.1.0
mypy-extensions==1.0.0
narwhals==1.15.1
nest-asyncio==1.6.0
networkx==3.4.2
nltk==3.9.1
numpy==1.26.4
nvidia-cublas-cu12==12.4.5.8
nvidia-cuda-cupti-cu12==12.4.127
nvidia-cuda-nvrtc-cu12==12.4.127
nvidia-cuda-runtime-cu12==12.4.127
nvidia-cudnn-cu12==9.1.0.70
nvidia-cufft-cu12==11.2.1.3
nvidia-curand-cu12==10.3.5.147
nvidia-cusolver-cu12==11.6.1.9
nvidia-cusparse-cu12==12.3.1.170
nvidia-nccl-cu12==2.21.5
nvidia-nvjitlink-cu12==12.4.127
nvidia-nvtx-cu12==12.4.127
olefile==0.47
ollama==0.4.2
omegaconf==2.3.0
onnx==1.17.0
onnxruntime==1.19.2
opencv-python==4.10.0.84
openpyxl==3.1.5
orjson==3.10.12
packaging==24.2
pandas==2.2.3
pdf2image==1.17.0
pdfminer.six==20231228
pdfplumber==0.11.4
pi_heif==0.21.0
pikepdf==9.4.2
pillow==10.4.0
portalocker==2.10.1
propcache==0.2.1
proto-plus==1.25.0
protobuf==5.29.0
psutil==6.1.0
pyarrow==18.1.0
pyasn1==0.6.1
pyasn1_modules==0.4.1
pycocotools==2.0.8
pycparser==2.22
pydantic==2.9.2
pydantic-settings==2.6.1
pydantic_core==2.23.4
pydeck==0.9.1
Pygments==2.18.0
pypandoc==1.14
pyparsing==3.2.0
pypdf==5.1.0
pypdfium2==4.30.0
PyStemmer==2.2.0.3
python-dateutil==2.9.0.post0
python-docx==1.1.2
python-dotenv==1.0.1
python-iso639==2024.10.22
python-magic==0.4.27
python-multipart==0.0.19
python-oxmsg==0.0.1
python-pptx==1.0.2
pytz==2024.2
PyYAML==6.0.2
qdrant-client==1.12.1
RapidFuzz==3.10.1
referencing==0.35.1
regex==2024.11.6
requests==2.32.3
requests-toolbelt==1.0.0
rich==13.9.4
rpds-py==0.21.0
rsa==4.9
safetensors==0.4.5
scipy==1.14.1
six==1.16.0
smmap==5.0.1
sniffio==1.3.1
snowballstemmer==2.2.0
soupsieve==2.6
SQLAlchemy==2.0.35
streamlit==1.40.2
sympy==1.13.1
tenacity==9.0.0
timm==1.0.11
tokenizers==0.20.3
toml==0.10.2
torch==2.5.1
torchvision==0.20.1
tornado==6.4.2
tqdm==4.67.1
transformers==4.46.3
triton==3.1.0
typing-inspect==0.9.0
typing_extensions==4.12.2
tzdata==2024.2
unstructured==0.16.8
unstructured-client==0.28.1
unstructured-inference==0.8.1
unstructured.pytesseract==0.3.13
urllib3==2.2.3
watchdog==6.0.0
webencodings==0.5.1
wrapt==1.17.0
xlrd==2.0.1
XlsxWriter==3.2.0
yarl==1.18.3

[feijoes]

@lukasschoettler
Instead of passing the callback on the chain.stream, perhaps try passing it during the creation of the ChatOllama:

chat_llm = ChatOllama(base_url=ollama_url,
                      model=ollama_llm,
                      temperature=0.6,
                      repetition_penalty=1.1,
                      callbacks=[callback]
                      )

[lukasschoettler]

I tried doing that, as well as passing in the callback at the creation of the retriever, which I assume you meant anyway.

retriever = vector_store.as_retriever(callbacks=[callback])

It didn't work. To me, this doesn't look like a problem of attaching the the callback handler. on_chain_end gets triggered 5 or 6 times. Maybe it has to do with the use of a client accessing an external vectorstore as a retriever?