Monitoring and observability

[Anand-puthiyapurayil]

@dosu ,

import os
import re
from typing import List, TypedDict
from langchain_core.prompts import PromptTemplate
from langchain_core.documents import Document
from langgraph.graph import START, StateGraph
from langchain_core.messages import HumanMessage
from langchain_ollama import ChatOllama
from utils.utils import load_config, initialize_embeddings, load_faiss_store
from logger.logger import get_logger

logger = get_logger(file)

-------------------------------------------------------------------

1) Prompt for LLM Re-ranking (for products)

-------------------------------------------------------------------

template = """You are an intelligent assistant that reorders product IDs based on any constraints in the user query.

Instructions:

• You have a set of {doc_count} products, each with metadata: product_id, product_name, price, rating, etc.
• The user’s query might contain any constraint (e.g., a price limit, color requirement, brand name, rating threshold, etc.).
• First, identify which products match those constraints (the “most relevant” or direct matches) and place them first.
• Then list any remaining products afterward in any sensible order.
• Return all product IDs (you must not omit any), each on its own line, prefixed by "Product ID:".
• Provide no extra commentary, no disclaimers—just the lines "Product ID: XYZ".

Context:
{context}

Question: {question}

Helpful Answer:
"""

custom_rag_prompt = PromptTemplate(
template=template,
input_variables=["context", "question", "doc_count"]
)

-------------------------------------------------------------------

2) Define Pipeline State (new, simplified)

-------------------------------------------------------------------

class ProductState(TypedDict):
question: str
k: int # Dynamic number of documents to retrieve.
context: List[Document] # Holds up to k retrieved product documents.
final_ids_output: str # The final combined output after re-ranking.
loop_step: int

-------------------------------------------------------------------

3) Helper: Regex Parsing for Product IDs

-------------------------------------------------------------------

def parse_product_ids_from_llm_text(llm_text: str) -> List[str]:
"""
Extracts all IDs from lines containing 'Product ID: ...' via regex.
e.g. 'Product ID: 1234' => '1234'
"""
pattern = r"Product ID:\s*(\S+)"
matches = re.findall(pattern, llm_text)
# Remove trailing punctuation (if any)
cleaned = [m.rstrip(",.") for m in matches]
return cleaned

-------------------------------------------------------------------

4) Retrieval: Retrieve k Products in One Call

-------------------------------------------------------------------

def retrieve_products(state: ProductState, faiss_store) -> ProductState:
"""
Uses the FAISS store retriever to fetch up to k product documents.
"""
try:
k = state.get("k", 200) # Use the dynamic k value (default to 200)
retriever = faiss_store.as_retriever(search_kwargs={'k': k})
docs = retriever.invoke(state["question"])
logger.info(f"Retrieved {len(docs)} product documents with k={k}.")
return {"context": docs}
except Exception as e:
logger.error(f"Error retrieving {k} product docs: {e}", exc_info=True)
return {"context": []}

-------------------------------------------------------------------

5) Rerank Top 25 and Append Remainder

-------------------------------------------------------------------

def rerank_and_append(state: ProductState, llm) -> ProductState:
"""
From the retrieved product documents, take the top 25 and send them to the LLM
for re-ranking, then append the remaining product IDs (docs 26 onward) after the LLM result.
"""
docs = state.get("context", [])
if not docs:
logger.warning("No product documents retrieved for re-ranking.")
return {"final_ids_output": "No relevant product IDs found."}

top25 = docs[:25]
remainder = docs[25:]

formatted_context = "\n".join(
    f"Product ID: {doc.metadata.get('product_id', 'Unknown')}, "
    f"Product Name: {doc.metadata.get('product_name', 'Unknown')}, "
    f"Price: {doc.metadata.get('price', 'Unknown')}, "
    f"Rating: {doc.metadata.get('rating', 'Unknown')}"
    for doc in top25
)
logger.info("Formatted context for top 25 products:\n" + formatted_context)

prompt = custom_rag_prompt.format(
    doc_count=len(top25),
    context=formatted_context,
    question=state["question"]
)

try:
    response = llm.invoke([HumanMessage(content=prompt)])
    llm_text = response.content
    logger.info("LLM successfully re-ranked the top 25 product documents.")
    
    # Extract product IDs from LLM output.
    llm_ids = parse_product_ids_from_llm_text(llm_text)
    original_ids = [str(doc.metadata.get("product_id", "Unknown")) for doc in top25]
    
    # Append any missing IDs that were in the original top25.
    missing = [pid for pid in original_ids if pid not in set(llm_ids)]
    if missing:
        logger.warning(f"LLM omitted {len(missing)} IDs: {missing}. Appending them.")
    combined_top25 = llm_ids + missing
    
    # Remove duplicates while preserving order.
    seen = set()
    final_top25 = []
    for pid in combined_top25:
        if pid not in seen:
            seen.add(pid)
            final_top25.append(pid)
    
    # Convert the remainder documents to ID strings.
    remainder_ids = [f"Product ID: {doc.metadata.get('product_id', 'Unknown')}" for doc in remainder]
    
    # Combine the re-ranked top 25 with the remainder appended.
    final_output = "\n".join(f"Product ID: {pid}" for pid in final_top25)
    if remainder_ids:
        final_output += "\n" + "\n".join(remainder_ids)
    
    return {"final_ids_output": final_output}
except Exception as e:
    logger.error("LLM re-ranking failed for top 25 product documents.", exc_info=True)
    return {"final_ids_output": "No relevant product IDs found."}

-------------------------------------------------------------------

6) Build LangGraph Workflow

-------------------------------------------------------------------

def build_product_graph(faiss_store, llm):
"""
Build a LangGraph pipeline with two nodes:
1. Retrieve k product documents.
2. Rerank top 25 and append the remaining product IDs.
"""
graph_builder = StateGraph(ProductState)
graph_builder.add_node("retrieve", lambda s: retrieve_products(s, faiss_store))
graph_builder.add_node("rerank_and_append", lambda s: rerank_and_append(s, llm))
graph_builder.add_edge(START, "retrieve")
graph_builder.add_edge("retrieve", "rerank_and_append")
compiled_graph = graph_builder.compile()
logger.info("LangGraph product pipeline compiled successfully.")
return compiled_graph

-------------------------------------------------------------------

7) Main Execution (for local testing)

-------------------------------------------------------------------

def main():
try:
this_dir = os.path.dirname(os.path.abspath(file))
config_path = os.path.join(this_dir, "..", "config.yaml")
config = load_config(config_path)
logger.info(f"Configuration loaded from {config_path}")

    # Use the llama3-70b-8192 model with Groq.
    local_llm = "llama3.1:8b"
    llm = ChatOllama(model=local_llm, temperature=0)
    
    embeddings = initialize_embeddings(config["output_dir"])
    faiss_store = load_faiss_store(config["product_store_path"], embeddings)
    logger.info(f"FAISS vector store loaded from: {config['product_store_path']}")

    # Build the pipeline using the new logic.
    graph = build_product_graph(faiss_store, llm)

    while True:
        query = input("\nEnter your product search query (or type 'exit' to quit): ").strip()
        if query.lower() in ["exit", "quit"]:
            logger.info("Exiting the application.")
            break

        # Ask the user for the k value
        k_value_input = input("Enter number of documents to retrieve (k): ").strip()
        try:
            k_value = int(k_value_input)
        except ValueError:
            logger.warning("Invalid k value provided, defaulting to 200.")
            k_value = 200

        initial_state: ProductState = {
            "question": query,
            "k": k_value,
            "context": [],
            "final_ids_output": "",
            "loop_step": 0,
        }

        final_state = graph.invoke(initial_state)
        final_ids = final_state.get("final_ids_output", "No response provided.")

        print("\n=== Final Output ===")
        print(final_ids)
        print("====================\n")

except KeyboardInterrupt:
    logger.info("Application interrupted by user. Exiting gracefully.")

if name == "main":
main()
i have the retrieval pipeline where i should faiss retrieval , llm to rerank, and all under the framework of langgraph , , now what are the componets and that i can monitor and observe and how can i do it ?

[dosubot[bot]]

I couldn't find specific information on monitoring and observing components within the StateGraph class or the PromptTemplate class in the LangChain framework. However, I can provide some general guidance on how you might approach monitoring and observability in your retrieval pipeline:

1. Logging: You are already using a logger in your code. Ensure that you log key events, errors, and important state changes. This will help you trace the execution flow and identify issues.

2. StateGraph Transitions: While specific methods for monitoring state transitions in StateGraph are not documented, you can add logging at each node transition to track the flow of data and state changes.

3. Performance Metrics: You can capture performance metrics such as execution time for each node in the pipeline. This can be done by recording timestamps before and after node execution and calculating the duration.

4. Error Handling: Ensure that you log exceptions and errors with sufficient detail to diagnose issues. You are already doing this in your retrieve_products and rerank_and_append functions.

5. Custom Observability: Implement custom observability hooks or callbacks if the framework supports it. This might involve extending classes or using decorators to capture additional metrics or state information.

6. External Monitoring Tools: Consider integrating external monitoring tools or services that can provide more comprehensive observability features, such as distributed tracing or real-time analytics.

For more detailed and specific observability features, you might need to refer to the LangChain documentation or explore the source code further to identify any built-in capabilities that can be leveraged.

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Bug Report | Other}

[dosubot[bot]]

The similarity_search_with_score method will return a list of tuples containing documents and their similarity scores, but it does not include model metrics. The model metrics, such as retriever name, vector store provider, and embedding provider, are obtained separately using the _get_ls_params method. The similarity_search_with_score method focuses on providing the similarity scores for the retrieved documents ^[1][2].

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Bug Report | Other}

[Anand-puthiyapurayil]

`def retrieve_products(state: ProductState, faiss_store, embeddings) -> ProductState:
try:
k = state.get("k", 200)

    retriever = faiss_store.as_retriever(search_kwargs={'k': k})
    # Use similarity_search_with_score to get both documents and their similarity scores
    results = retriever.similarity_search_with_score(state["question"])
    logger.info(f"Retrieved {len(results)} product documents with similarity search (k={k}).")
    
    # Separate the documents and their corresponding scores
    docs, scores = zip(*results) if results else ([], [])
    
    # Get model metrics using the _get_ls_params method
    model_metrics = faiss_store._get_ls_params()
    logger.info(f"Model Metrics: {model_metrics}")
    
    # Log the retrieval record including query details, scores, and model metrics.
    retrieval_record = {
        "query": state["question"],
        "retrieved_count": len(results),
        "model_metrics": model_metrics,
        "scores": scores
    }`  
    
  2025-03-06 12:29:03 - __main__ - ERROR - Error retrieving 10 product docs: 'VectorStoreRetriever' object has no attribute 'similarity_search_with_score'

@dosu , this gives me error

[dosubot[bot]]

The error you're encountering, "'VectorStoreRetriever' object has no attribute 'similarity_search_with_score'," indicates that the VectorStoreRetriever class does not have a method named similarity_search_with_score. To resolve this, ensure you are using the correct class that implements this method.

In the LangChain library, the similarity_search_with_score method is available in classes like InMemoryVectorstoreWithSearch. You should use an instance of a class that supports this method. Here's an example setup:

# Ensure you are using the correct vector store class
vectorstore = InMemoryVectorstoreWithSearch()
retriever = MultiVectorRetriever(
    vectorstore=vectorstore,
    docstore=InMemoryStore(),
    doc_id="doc_id"
)

# Perform similarity search with score
results = vectorstore.similarity_search_with_score(state["question"], k=state.get("k", 200))

Make sure your faiss_store or equivalent is an instance of a class that implements similarity_search_with_score. Adjust your imports and class usage accordingly to match the correct implementation ^[1][2][3].

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Bug Report | Other}

[Anand-puthiyapurayil]

'_get_ls_params' @dosu , where exactly does this function fit ? give me an example

[dosubot[bot]]

The _get_ls_params function is used within the LangChain framework to obtain standard parameters for tracing, specifically in the context of the Ollama chat models. It constructs a LangSmithParams object with parameters such as ls_provider, ls_model_type, and optionally includes ls_stop, ls_model_name, and ls_temperature based on the attributes of the class or provided keyword arguments. Here's an example of its usage:

# Assuming `self` is an instance of a class that inherits from a superclass
# with a `_get_ls_params` method, and `self._identifying_params` is defined.

# Example usage of _get_ls_params
params = self._get_ls_params(stop=["\n"], max_tokens=100, temperature=0.7)

This function is typically used internally within the class to prepare parameters for making API calls or logging, ensuring that the correct values for max_tokens, temperature, and other relevant parameters are included. The function also determines the default provider as "ollama" and sets the model type to "chat", adjusting it if necessary based on the class attributes ^[1][2][3][4].

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Bug Report | Other}