| """Generate node - produces answers using the generation chain.""" |
|
|
| from typing import Any, Dict, List |
|
|
| from langchain_core.documents import Document |
|
|
| from src.chains import generation_chain |
| from src.core import logger |
| from src.core.state import GraphState |
|
|
|
|
| def format_documents_for_context(documents: List[Document]) -> str: |
| """ |
| Format documents into a readable context string with sources. |
| |
| Args: |
| documents: List of Document objects |
| |
| Returns: |
| Formatted context string |
| """ |
| if not documents: |
| return "No relevant documents found." |
| |
| formatted = [] |
| for i, doc in enumerate(documents, 1): |
| source = doc.metadata.get("source", "unknown") |
| title = doc.metadata.get("title", "N/A") |
| content = doc.page_content.strip() |
| |
| formatted.append( |
| f"[Document {i}]\n" |
| f"Source: {source}\n" |
| f"Title: {title}\n" |
| f"Content: {content}\n" |
| f"{'='*60}" |
| ) |
| |
| return "\n\n".join(formatted) |
|
|
|
|
| def generate_node(state: GraphState) -> Dict[str, Any]: |
| """ |
| Generate an answer using the retrieved documents. |
| |
| Args: |
| state: Current graph state with question and documents. |
| |
| Returns: |
| Updated state with generation. |
| """ |
| logger.debug("Generating answer...") |
|
|
| question = state["question"] |
| documents = state["documents"] |
| |
| |
| context = format_documents_for_context(documents) |
| logger.debug(f"Formatted {len(documents)} documents into context") |
|
|
| generation = generation_chain.invoke({"question": question, "context": context}) |
|
|
| logger.info(f"Generated response ({len(generation)} chars)") |
|
|
| return {"generation": generation, "documents": documents} |
|
|
