DEV Community: Jimmy

Building My Smart 2nd Brain, Part 4: The Art of Documents Searching

Jimmy — Sat, 18 Oct 2025 13:20:15 +0000

I originally planned to wrap up this series in this part. However, while reviewing my code, I realized there’s a section that could greatly benefit from some additional attention and refinement before concluding this mini-project.

That section is about documents searching.

Chunking and Storing: Connecting Pieces Together

Let's see what our store_node does now:

def store_node(self, state: KnowledgeState):

        if state.embeddings and state.chunks:
            try:
                # Initialize vectorstore if not provided
                if not self.vectorstore:
                    self.vectorstore = Chroma(
                        collection_name="smart_second_brain",
                        embedding_function=self.embedding_model,
                        persist_directory=self.chromadb_dir
                    )

                # Prepare metadata for each chunk
                # Note: ChromaDB doesn't accept lists in metadata, so we join categories
                metadatas = [
                    {
                        "source": state.source or "unknown",
                        "categories": ", ".join(state.categories) if state.categories else "general",
                        "chunk_id": i
                    }
                    for i in range(len(state.chunks))
                ]

                # Store chunks with metadata in ChromaDB
                self.vectorstore.add_texts(
                    texts=state.chunks,
                    metadatas=metadatas
                )

                # Note: Newer ChromaDB versions don't require explicit persist()
                # Data is automatically persisted when using persist_directory

                state.status = "stored"
                state.logs = (state.logs or []) + [
                    f"Stored {len(state.chunks)} chunks in ChromaDB with categories {state.categories or ['general']}"
                ]
            except Exception as e:
                state.status = "error"
                state.logs = (state.logs or []) + [f"Storing failed: {e}"]
        else:
            state.logs = (state.logs or []) + ["No embeddings/chunks to store"]

        return state

The program uses a predefined list of categories, provided as a state input for the metadata. The chunk size is predetermined using the RecursiveCharacterTextSplitter.

# Prepare metadata for each chunk
# Note: ChromaDB doesn't accept lists in metadata, so we join categories

metadatas = [
                {
                    "source": state.source or "unknown",
                    "categories": ", ".join(state.categories) if state.categories else "general",
                    "chunk_id": i
                }
                for i in range(len(state.chunks))
            ]

            # Store chunks with metadata in ChromaDB
            self.vectorstore.add_texts(
                texts=state.chunks,
                metadatas=metadatas
            )

The current version employs a very basic chunking and storage mechanism:

There are no document identifiers, meaning chunks from multiple documents are indistinguishable.
Chunk metadata is minimal—just a concatenated string of category keywords and a numeric chunk ID.

This bare-bones approach introduces several limitations:

No persistent identifiers: Without stable doc_id or chunk_id, it’s difficult to track, update, or delete individual chunks.
Lossy metadata: Metadata like categories is compressed into comma-separated strings, which makes it unreliable for filtering or matching individual labels.
Lack of safeguards: There are no mechanisms to handle missing metadata.

Overall, this approach makes it challenging to manage chunks effectively, trace their origins, or maintain metadata integrity.

Therefore, I provide the following changes:

if state.raw_document:
            doc_id = state.doc_id or f"doc_{uuid.uuid4()}"
            state.doc_id = doc_id
            state.ingested_at = state.ingested_at or datetime.datetime.utcnow().isoformat()

            splitter = RecursiveCharacterTextSplitter(
                chunk_size=500,
                chunk_overlap=50,
            )
            chunks = splitter.split_text(state.raw_document)
            state.chunks = chunks
            state.chunk_metadata = []

            for index, chunk in enumerate(chunks):
                metadata = {
                    "doc_id": doc_id,
                    "chunk_index": index,
                    "chunk_id": f"{doc_id}::chunk_{index}",
                    "source": state.source or "unknown",
                    "categories": state.categories or [],
                    "keywords": (state.auto_generated_keywords or []) or (
                        state.metadata.get("keywords") if state.metadata else []
                    ),
                    "ingested_at": state.ingested_at,
                    "knowledge_type": state.knowledge_type or "conversational",
                    "char_start": state.raw_document.find(chunk),
                }
                metadata["char_end"] = metadata["char_start"] + len(chunk) if metadata["char_start"] >= 0 else None
                state.chunk_metadata.append(metadata)

            state.logs = (state.logs or []) + [
                f"Chunked document into {len(state.chunks)} chunks with doc_id={doc_id}"
            ]
        else:
            state.logs = (state.logs or []) + ["No raw_document found for chunking"]

        return state

There is a document id that connects every chunk from the same document.
There is a chunk id defined as well within a document.
The list of keywords becomes a list type instead of a concatenated string.
Characters offset (char_start, char_end) are kept as well.

And the store_node needs to be refined as well:

if state.embeddings and state.chunks:
            try:
                # Initialize vectorstore if not provided
                if not self.vectorstore:
                    self.vectorstore = Chroma(
                        collection_name=self.collection_name,
                        embedding_function=self.embedding_model,
                        persist_directory=self.chromadb_dir
                    )

                metadatas = []
                ids = []
                for idx, chunk in enumerate(state.chunks):
                    base_metadata = state.chunk_metadata[idx] if state.chunk_metadata and idx < len(state.chunk_metadata) else {}
                    metadata = {
                        "doc_id": base_metadata.get("doc_id", state.doc_id or "unknown"),
                        "chunk_index": base_metadata.get("chunk_index", idx),
                        "chunk_id": base_metadata.get("chunk_id", f"chunk_{idx}"),
                        "source": base_metadata.get("source", state.source or "unknown"),
                        "categories": base_metadata.get("categories", state.categories or []),
                        "ingested_at": base_metadata.get("ingested_at", state.ingested_at),
                        "knowledge_type": base_metadata.get("knowledge_type", state.knowledge_type or "conversational"),
                        "char_start": base_metadata.get("char_start"),
                        "char_end": base_metadata.get("char_end"),
                    }
                    metadata.update(state.metadata or {})

                    safe_metadata = {}
                    for key, value in metadata.items():
                        if isinstance(value, (str, int, float, bool)) or value is None:
                            safe_metadata[key] = value
                        elif isinstance(value, list):
                            safe_metadata[key] = ", ".join(str(v) for v in value)
                        else:
                            safe_metadata[key] = str(value)

                    metadatas.append(safe_metadata)
                    ids.append(metadata["chunk_id"])

                documents = [
                    Document(page_content=chunk, metadata=meta)
                    for chunk, meta in zip(state.chunks, metadatas)
                ]

                # Store chunks in vector store and update hybrid retriever if available
                if self.vectorstore:
                    if hasattr(self.vectorstore, "add_documents"):
                        self.vectorstore.add_documents(documents, ids=ids)
                    elif hasattr(self.vectorstore, "add_texts"):
                        self.vectorstore.add_texts([d.page_content for d in documents], metadatas=metadatas, ids=ids)

                if self.retriever:
                    self.retriever.add_documents(documents)

                # Note: Newer ChromaDB versions don't require explicit persist()
                # Data is automatically persisted when using persist_directory

                state.status = "stored"
                state.logs = (state.logs or []) + [
                    f"Stored {len(state.chunks)} chunks in ChromaDB with categories {state.categories or ['general']}"
                ]
            except Exception as e:
                state.status = "error"
                state.logs = (state.logs or []) + [f"Storing failed: {e}"]
        else:
            state.logs = (state.logs or []) + ["No embeddings/chunks to store"]

The meta data field now becomes a proper JSON object that includes all relevant data.
The LangChain's document object is used here for storing document chunks, so it is running add_documents instead of add_texts before. It keep the vector store aligned with LangChain’s document model.

Preprocessing for Keywords Extraction

The current version requires an explicitly specified list of keywords for meta data. For a single PDF it may be fine, but how about a bunch of documents that are about different topics?

We are doing an AI system, leaving such unresolved pain point is something I cannot compromise.

To handle this, I add an additional node in the ingestion flow, called 'preprocess':

# =============================================================================
        # INGESTION BRANCH
        # =============================================================================

        # Document processing pipeline
        graph.add_edge("preprocess", "chunk")
        graph.add_edge("chunk", "embed")      # Chunk -> Embed
        graph.add_edge("embed", "store")     # Embed -> Store
        graph.add_edge("store", END)         # Store -> End

The preprocess node activates only during ingestion requests when auto_preprocess is true. It reads the raw document, tokenizes it—preferring spaCy and falling back to a regex tokenizer if spaCy is unavailable—filters the resulting tokens, and builds frequency counts. The highest-frequency survivors become up to ten auto-generated keywords and five categories, which are merged into the state’s metadata before logging a summary for downstream stages.

def preprocess_node(self, state: KnowledgeState):
        """
        Perform optional automated metadata extraction prior to chunking.

        When auto_preprocess is enabled, this node analyses the raw document to
        infer keywords and categories. When disabled, it ensures the caller has
        supplied sufficient metadata; otherwise the ingestion run is skipped.
        """
        if state.query_type != "ingest":
            return state

        if not state.raw_document:
            state.logs = (state.logs or []) + ["⚠️ No raw document present for preprocessing"]
            return state

        if not state.auto_preprocess:
            has_user_metadata = bool((state.categories or []) or (state.metadata or {}))
            if not has_user_metadata:
                state.status = "skipped_no_metadata"
                state.logs = (state.logs or []) + [
                    "⚠️ Ingestion skipped: auto_preprocess disabled and no metadata provided"
                ]
            return state

        # --- Tokenize and normalize text ---
        filtered_tokens: list[str] = []
        nlp = get_spacy_nlp()
        if nlp is not None:
            doc = nlp(state.raw_document)
            for token in doc:
                lemma = token.lemma_.lower().strip()
                if (
                    lemma
                    and lemma.isalpha()
                    and len(lemma) >= 4
                    and lemma not in SKLEARN_STOPWORDS
                ):
                    filtered_tokens.append(lemma)
        else:
            text = state.raw_document.lower()
            tokens = re.findall(r"[a-zA-Z]{4,}", text)
            filtered_tokens = [token for token in tokens if token not in SKLEARN_STOPWORDS]

        if not filtered_tokens:
            state.logs = (state.logs or []) + [
                "ℹ️ Auto preprocessing found no candidate keywords"
            ]
            return state

        counter = Counter(filtered_tokens)
        keywords = [word for word, _ in counter.most_common(10)]
        categories = keywords[:5]

        state.auto_generated_keywords = keywords
        state.auto_generated_categories = categories

        # Merge inferred categories with any caller-supplied ones while preserving order
        existing_categories = state.categories or []
        merged_categories = list(dict.fromkeys(existing_categories + categories))
        state.categories = merged_categories

        state.metadata = state.metadata or {}
        if keywords:
            state.metadata.setdefault("keywords", ", ".join(keywords))
            state.metadata.setdefault("auto_generated_keywords", ", ".join(keywords))
        if merged_categories:
            state.metadata.setdefault("categories", ", ".join(merged_categories))
            state.metadata.setdefault("auto_generated_categories", ", ".join(categories))
        state.metadata.setdefault("auto_preprocess", True)

        state.logs = (state.logs or []) + [
            f"🔍 Auto preprocessing inferred keywords={keywords[:5]} categories={categories[:3]}"
        ]

        return state

A curated stopword list is essential because common glue words (“the,” “with,” “this,” etc.) dominate naive frequency counts yet carry little semantic value. Stripping them away ensures the remaining tallies highlight genuinely informative terms, yielding cleaner metadata and more accurate retrieval filters.

spaCy is brought in because its language model provides well-tested tokenization and morphological analysis—capabilities that lightweight regex tokenizers lack. When spaCy loads successfully it supplies part-of-speech tags and lemmas, enabling the node to recognize tokens accurately, respect context, and normalize linguistic variants. If the model isn’t available the workflow still proceeds, but with reduced sophistication.

spaCy lemmatization specifically converts each token to its base dictionary form while considering grammatical context. For example, “running,” “ran,” and “runs” all normalize to “run,” and context-sensitive cases like “better” as an adjective become “good.” This consolidation prevents inflectional variants from fragmenting the term frequencies that seed keyword and category extraction.

Raw sentence:
"Analysts were studying studies of the companies' analyses while runners ran, running into numerous hurdles."

Raw tokens (after simple cleanup):
analysts, were, studying, studies, of, the, companies, analyses, while, runners, ran, running, into, numerous, hurdles

Unique raw tokens: 15

Lemmas (spaCy-style lemmatization):
analyst, be, study, study, of, the, company, analysis, while, runner, run, run, into, numerous, hurdle

Unique lemmas: 12

At the end, the preprocess_node enriches the list of keywords for a document, it makes embedding and storing document batches much easier.

Retrieval: Another Side of the Coin

The current version uses the standard retriever provided by the vector object, performs the similarity search for the top 5 documents.

try:
            # Try different retrieval strategies in order of preference
            if self.retriever:
                logger.info(f"🔍 Retrieving docs for query: {state.user_input}")
                results = self.retriever.get_relevant_documents(state.user_input)
            elif self.vectorstore:
                logger.info(f"🔍 Using vectorstore.similarity_search for query: {state.user_input}")
                results = self.vectorstore.similarity_search(state.user_input, k=5)
            else:
                results = []

            # Format retrieved documents for downstream processing
            state.retrieved_docs = [{"content": r.page_content, "metadata": r.metadata} for r in results]
            state.logs = (state.logs or []) + [f"Retrieved {len(state.retrieved_docs)} docs"]

        except Exception as e:
            logger.error(f"❌ Retrieval failed: {e}")
            state.retrieved_docs = []
            state.logs = (state.logs or []) + [f"Retrieval error: {str(e)}"]

The default get_relevant_documents call in LangChain relies on embedding-based similarity search. In practice, it follows these steps:

Convert the query to an embedding – The same embedding model used at ingestion time (e.g., OpenAI, Azure, etc.) transforms the user’s question into a high-dimensional vector.
Perform a nearest-neighbor lookup – The retriever asks the vector database (Chroma in this workflow) for the k stored vectors that sit closest to the query vector under a similarity metric such as cosine similarity or inner product.
Return the top matches as Document objects – Each hit carries both the chunk text and its metadata, enabling downstream filtering or prompting.

However, relying solely on embedding-based similarity search will have the following problems:

Missing exact matches when the embedding fails to capture critical keywords, acronyms, or newly coined terms.
Returning noisy passages because long or multi-topic chunks get averaged into a single vector.
Losing document context, as chunks are treated independently with no notion of surrounding structure.

In real production setup, hybrid searching style will be preferred, which is what I introduce below:

def retriever_node(self, state: KnowledgeState):
        """
        Retrieve relevant documents from the vector database.

        This node performs similarity search to find the most relevant
        documents for a given user query. Supports multiple retrieval
        strategies with fallback mechanisms.

        Args:
            state: KnowledgeState object containing user_input

        Returns:
            KnowledgeState: Updated state with retrieved_docs array

        Retrieval Strategies:
            1. Dedicated retriever (if configured)
            2. Vectorstore similarity search (fallback)
            3. Empty results (if both unavailable)

        Processing:
            - Performs semantic search using user query
            - Retrieves top-k most relevant documents
            - Formats results for downstream processing
            - Handles retrieval errors gracefully
            - Logs retrieval operations and statistics
        """
        if not state.user_input:
            state.logs = (state.logs or []) + ["No user_input provided for retrieval"]
            state.retrieved_docs = []
            return state

        try:
            retrieval_options = {
                "query": state.user_input,
                "k": (state.retrieval_options or {}).get("k", 8),
                "filters": state.metadata_filters or {},
                "use_hybrid": (state.retrieval_options or {}).get("use_hybrid", True),
                "min_score": (state.retrieval_options or {}).get("min_score", 0.1),
                "rerank_top_k": (state.retrieval_options or {}).get("rerank_top_k", 12),
            }

            retrieval_log = {
                "query": state.user_input,
                "filters": retrieval_options["filters"],
                "use_hybrid": retrieval_options["use_hybrid"],
                "min_score": retrieval_options["min_score"],
                "timestamp": datetime.datetime.utcnow().isoformat(),
            }

            if self.retriever:
                logger.info(f"🔍 Retrieving docs for query: {state.user_input}")
                if hasattr(self.retriever, "retrieve"):
                    results = self.retriever.retrieve(
                        query=retrieval_options["query"],
                        k=retrieval_options["k"],
                        filters=retrieval_options["filters"],
                        use_hybrid=retrieval_options["use_hybrid"],
                        min_score=retrieval_options["min_score"],
                    )
                elif hasattr(self.retriever, "get_relevant_documents"):
                    results = self.retriever.get_relevant_documents(retrieval_options["query"])
                else:
                    results = []

                # Optionally apply reranking if supported
                if (
                    retrieval_options["rerank_top_k"]
                    and hasattr(self.retriever, "retrieve_with_reranking")
                ):
                    results = self.retriever.retrieve_with_reranking(
                        query=retrieval_options["query"],
                        k=retrieval_options["k"],
                        rerank_top_k=retrieval_options["rerank_top_k"],
                    )

                formatted_results = []
                top_results = []
                for result in results:
                    doc = result.document if hasattr(result, "document") else result
                    score = result.score if hasattr(result, "score") else None
                    formatted_results.append({
                        "content": doc.page_content,
                        "metadata": doc.metadata,
                        "score": score,
                    })
                    top_results.append({
                        "chunk_id": doc.metadata.get("chunk_id"),
                        "doc_id": doc.metadata.get("doc_id"),
                        "source": doc.metadata.get("source"),
                        "score": score,
                    })
                state.retrieved_docs = formatted_results
                retrieval_log["results"] = top_results

            elif self.vectorstore:
                logger.info(f"🔍 Using vectorstore.similarity_search for query: {state.user_input}")
                results = self.vectorstore.similarity_search(
                    retrieval_options["query"],
                    k=retrieval_options["k"],
                    filter=retrieval_options["filters"] or None,
                )
                state.retrieved_docs = [
                    {"content": r.page_content, "metadata": r.metadata, "score": None}
                    for r in results
                ]
                retrieval_log["results"] = [
                    {
                        "chunk_id": r.metadata.get("chunk_id"),
                        "doc_id": r.metadata.get("doc_id"),
                        "source": r.metadata.get("source"),
                        "score": None,
                    }
                    for r in results
                ]
            else:
                state.retrieved_docs = []
                retrieval_log["results"] = []

            state.retrieval_log = (state.retrieval_log or []) + [retrieval_log]
            state.logs = (state.logs or []) + [f"Retrieved {len(state.retrieved_docs)} docs"]

        except Exception as e:
            logger.error(f"❌ Retrieval failed: {e}")
            state.retrieved_docs = []
            state.retrieval_log = (state.retrieval_log or []) + [{
                "query": state.user_input,
                "error": str(e),
                "timestamp": datetime.datetime.utcnow().isoformat(),
            }]
            state.logs = (state.logs or []) + [f"Retrieval error: {str(e)}"]

        return state

There are more retrieval options available:

retrieval_options = {
                "query": state.user_input,
                "k": (state.retrieval_options or {}).get("k", 8),
                "filters": state.metadata_filters or {},
                "use_hybrid": (state.retrieval_options or {}).get("use_hybrid", True),
                "min_score": (state.retrieval_options or {}).get("min_score", 0.1),
                "rerank_top_k": (state.retrieval_options or {}).get("rerank_top_k", 12),
            }

Apart from query and k that we already know, here are three new parameters:

filters – metadata filter dict; usually state.metadata_filters coming from the API/UI (e.g., filter by source, categories, knowledge type).
use_hybrid – toggles the hybrid ensemble (semantic + BM25). Defaults True; if False it forces pure vector search.
min_score – minimum similarity threshold; results below this are dropped. Defaults to 0.1.
rerank_top_k – how many hits to pull before the optional reranker trims them back down to k. Here, grab 12, rerank, then return the best k.

To provide this features, we cannot use the generic retriever provided by chroma's as_retriever() function. Here we do a customized retriever called SmartDocumentRetriever.

A Smarter Document Retriever

The SmartDocumentRetriever performs the following:

running the standard embedding-based similarity search, and
running the BM25 search,
combine the result with a 0.7/0.3 weighting (0.7 embedding similarity search, 0.3 BM25),
finally rerank the retrieved documents.

But what is BM25?

BM25 is a keyword-based search algorithm that ranks documents based on the relevance of their terms to a given query. It calculates a score by considering term frequency (TF), document length, and how rare a term is across the entire collection of documents to give more weight to important keywords.

BM25 shines when exact or near-exact word matches matter (names, codes, jargon), bit misses purely semantic matches (synonyms or paraphrases) because it relies on lexical overlap.

In order to support BM25 search, the BM25 indexes must be created when the document is first ingested. It also means if the application is restarted, it must be rebuilt.

def add_documents(self, documents: List[Document]) -> None:
        """
        Add documents to the vectorstore.

        Args:
            documents: List of Document objects to add
        """
        try:
            # Add to vectorstore
            self.vectorstore.add_documents(documents)

            # Update BM25 retriever
            if self.bm25_retriever is None:
                self.bm25_retriever = BM25Retriever.from_documents(documents)
            else:
                # Add new documents to existing BM25
                self.bm25_retriever.add_documents(documents)

            # Update ensemble retriever
            self._update_ensemble_retriever()

            logger.info(f"✅ Added {len(documents)} documents to retriever")

        except Exception as e:
            logger.error(f"❌ Failed to add documents: {e}")

When the MasterGraphBuilder is first created, all documents from the vector store will be retrieved and sent to the SmartDocumentRetreiver for running the "hydrate_bm25()" function to rebuild the indexes.

SmartDocumentRetriever combines both BM25 and embedding-based similarity search to produce the best result.

Before returning the final list of documents, SmartDocumentRetriever will perform one more reranking step to make sure the results are most relevant. It applies a simple, hand-coded reranker that nudges longer, categorized, time-stamped chunks upward.

It is just a very simple self-baked implementation of reranker. In production use, you may consider using a more robust approach such as cross-encoder or LLM based reranker.

def retrieve_with_reranking(
        self,
        query: str,
        k: int = 5,
        rerank_top_k: int = 20
    ) -> List[RetrievalResult]:
        """
        Retrieve documents with re-ranking for better relevance.

        Args:
            query: Search query
            k: Final number of documents to return
            rerank_top_k: Number of documents to retrieve initially for re-ranking

        Returns:
            Re-ranked list of RetrievalResult objects
        """
        # First, retrieve more documents than needed
        initial_results = self.retrieve(
            query=query,
            k=rerank_top_k,
            use_hybrid=True
        )

        if len(initial_results) <= k:
            return initial_results

        # Simple re-ranking based on content length and keyword matches
        # In practice, you might use a cross-encoder or other re-ranking model
        reranked_results = []
        for result in initial_results:
            # Boost score based on content quality indicators
            boost = 0.0

            # Prefer longer, more informative content
            content_length = len(result.document.page_content)
            if content_length > 500:
                boost += 0.1

            # Prefer content with categories/metadata
            if result.document.metadata.get("categories"):
                boost += 0.05

            # Prefer recent content (if timestamp available)
            if "timestamp" in result.document.metadata:
                boost += 0.05

            result.score = min(result.score + boost, 1.0)
            reranked_results.append(result)

        # Sort by new scores and return top k
        reranked_results.sort(key=lambda x: x.score, reverse=True)
        return reranked_results[:k]

Final Touch on UI

The UI part also got some refinement.

First of all, I included the configuration pane here:

The most important properties are "Enable Auto Metadata Preprocessing" and "Use Hybrid Retrieval", which are essential for running the preprocess_node, and enabling the ensemble retrieval.

Another enhancement is about the citation support.

If an answer can be provided based on documents in the vector DB, a citation entry will be generated as well that provides the respective document ID and chunk ID, with further enhancements we can support instant referencing to the source documents.

RAG is Challenging

Actually I didn’t anticipate that searching alone would lead to another long article. Along the way, I discovered that managing vector database searches efficiently is a far more challenging topic than I initially thought. Algorithms like TF-IDF and MMR (Maximum Marginal Relevance) quickly overwhelmed me, as someone who considers himself more of an application developer than a mathematician.

While working on the project, I found myself constantly asking my AI assistant to explain the meaning behind certain algorithms. I meticulously scrutinized every piece of code generated by AI tools, performing extensive checks to ensure everything made sense and worked as expected. At least I am confident that this is a working repository and is able to function as what I have demonstrated.

And of course, I indeed learned a lot. :)

See you next time, hope I could build and share something even more interesting.

(Source code repository: https://github.com/jimmyhott/SMART-SECOND-BRAIN)

Building My Smart 2nd Brain, Part 3: API and UI Explained

Jimmy — Sun, 05 Oct 2025 11:38:26 +0000

In parts 1 and 2, we walked through the core LangChain graph behind our “Smart 2nd Brain” application. We covered document ingestion and retrieval, and showed how to implement a checkpointer, a vector database, and Human-In-The-Loop (HITL) controls to make the system more robust and trustworthy.

As a personal knowledge management tool, it’s not complete without an elegant, intuitive user interface. And to support easier testing, automation, and batch operations, it’s good practice to expose the graph behind a well-defined API layer. In this article, we’ll cover both: building a clean UI and designing a simple, reliable API to shield and orchestrate access to the graph.

FastAPI comes into actions

For the API implementation, I was thinking should I use Flask instead of FastAPI. At the end FastAPI’s preferred because it handles concurrent I/O (LLM calls, vector DB queries, streaming) efficiently, integrates cleanly with Pydantic for strict schemas around our LangChain graph inputs/output.

Inside the project, I have got a dedicated /api tier, which the start-up script is as follows:

if __name__ == "__main__":
    try:
        import uvicorn
        # Start the development server with configuration from settings
        uvicorn.run(
            "api.main:app",                    # Application import string
            host=settings.host,                # Host binding (default: 0.0.0.0)
            port=settings.port,                # Port number (default: 8000)
            reload=settings.debug,             # Auto-reload on code changes in debug mode
            log_level=settings.log_level.lower(),  # Logging level from settings
        )
    except ImportError:
        # Provide helpful error message if uvicorn is not available
        print("uvicorn not available, run with: uvicorn api.main:app --reload")

We use Uvicorn which is a modern ASGI server for Python. When the server is started the FastAPI executes the lifespan method, it is an async context manager that runs once when the app starts and once when it stops. In the start-up process, FastAPI lazy-imports heavy modules: creates AzureOpenAIEmbeddings, AzureChatOpenAI, and Chroma if settings are available. Also it builds the MasterGraphBuilder and compiles the graph.

from fastapi import FastAPI
from contextlib import asynccontextmanager
from fastapi.middleware.cors import CORSMiddleware

from api.core.config import settings

# Initialize graph builder and compiled graph at app startup
@asynccontextmanager
async def lifespan(app: FastAPI):
    try:
        # Lazy import to avoid heavy imports at module load
        from agentic.workflows.master_graph_builder import MasterGraphBuilder
        from langchain_openai import AzureOpenAIEmbeddings, AzureChatOpenAI
        from langchain_chroma import Chroma

        # Initialize models from settings
        if settings.openai_api_key and settings.azure_openai_endpoint_url:
            embedding_model = AzureOpenAIEmbeddings(
                azure_deployment="text-embedding-3-small",
                openai_api_version="2024-12-01-preview",
                azure_endpoint=settings.azure_openai_endpoint_url,
                openai_api_key=settings.openai_api_key,
            )
            llm = AzureChatOpenAI(
                azure_deployment="gpt-4o",
                openai_api_version="2024-12-01-preview",
                azure_endpoint=settings.azure_openai_endpoint_url,
                openai_api_key=settings.openai_api_key,
                temperature=0.1,
            )

            vectorstore = Chroma(
                collection_name="smart_second_brain",
                embedding_function=embedding_model,
                persist_directory="./chroma_db",
            )

            graph_builder = MasterGraphBuilder(
                llm=llm,
                embedding_model=embedding_model,
                vectorstore=vectorstore,
                chromadb_dir="./chroma_db",
            )
            compiled_graph = graph_builder.build()

            app.state.graph_builder = graph_builder
            app.state.compiled_graph = compiled_graph
        else:
            # Defer initialization to first request if credentials absent
            app.state.graph_builder = None
            app.state.compiled_graph = None
    except Exception:
        # Ensure attributes exist even if init fails
        app.state.graph_builder = None
        app.state.compiled_graph = None
    yield

# =============================================================================
# FASTAPI APPLICATION CONFIGURATION
# =============================================================================

# Create the main FastAPI application instance with metadata and configuration
app = FastAPI(
    title="Smart Second Brain API",
    description="An intelligent knowledge management system powered by LangGraph and AI",
    version="0.1.0",
    # Only show API docs in debug mode for security
    docs_url="/docs" if settings.debug else None,
    redoc_url="/redoc" if settings.debug else None,
    lifespan=lifespan,
)

# =============================================================================
# MIDDLEWARE CONFIGURATION
# =============================================================================

# Add CORS (Cross-Origin Resource Sharing) middleware to allow frontend communication
# This enables the Streamlit frontend to make API calls from different origins
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.allowed_hosts,  # Configured origins from settings
    allow_credentials=True,                # Allow cookies and authentication headers
    allow_methods=["*"],                   # Allow all HTTP methods (GET, POST, PUT, DELETE)
    allow_headers=["*"],                   # Allow all request headers
)

Here I include the basic setup for /test and /health endpoints as well as part of generic configuration. For the URL specific to the graph operations, they are grouped under the graph_api.py:

# =============================================================================
# API ENDPOINTS
# =============================================================================

@router.post("/ingest", response_model=WorkflowResponse)
async def ingest_document(
    request: IngestRequest,
    graph_builder: MasterGraphBuilder = Depends(get_graph_builder),
    http_request: Request = None
):

    start_time = datetime.utcnow()

    try:
        # Generate unique thread ID for this ingestion workflow
        thread_id = f"ingest_{int(start_time.timestamp())}"

        # Create knowledge state for the ingestion workflow
        state = KnowledgeState(
            query_type="ingest",                           # Workflow type identifier
            raw_document=request.document,                 # Original document content
            source=request.source or "api_ingest",         # Document source or default
            categories=request.categories or [],           # Document categories
            metadata=request.metadata or {}                # Additional metadata
        )

        # Ensure compiled graph is available
        compiled_graph = getattr(http_request.app.state, "compiled_graph", None)
        if compiled_graph is None:
            compiled_graph = graph_builder.build()
            http_request.app.state.compiled_graph = compiled_graph

        # Execute the ingestion workflow with thread ID for tracking
        logger.info(f"🔄 Starting document ingestion workflow: {thread_id}")
        result = compiled_graph.invoke(
            state,
            config={"configurable": {"thread_id": thread_id}}
        )

        # Calculate execution time for performance monitoring
        execution_time = (datetime.utcnow() - start_time).total_seconds()

        logger.info(f"✅ Ingestion completed in {execution_time:.2f}s")
        return WorkflowResponse(
            success=True,
            thread_id=thread_id,
            result=result,
            execution_time=execution_time,
            timestamp=datetime.utcnow()
        )

    except Exception as e:
        logger.error(f"❌ Ingestion failed: {e}")
        raise HTTPException(
            status_code=500,
            detail=f"Document ingestion failed: {str(e)}"
        )


@router.post("/query", response_model=WorkflowResponse)
async def query_knowledge_base(
    request: QueryRequest,
    graph_builder: MasterGraphBuilder = Depends(get_graph_builder),
    http_request: Request = None
):

    start_time = datetime.utcnow()

    try:
        # Use provided thread ID or generate new one for conversation continuity
        thread_id = request.thread_id or f"query_{int(start_time.timestamp())}"

        # Get conversation history from custom memory manager
        conversation_history = conversation_memory.get_conversation_history(thread_id)

        # Create knowledge state for the query workflow
        state = KnowledgeState(
            query_type="query",                            # Workflow type identifier
            user_input=request.query,                      # User's question
            categories=[],                                 # No categories for queries
            messages=conversation_history,                 # Use custom conversation memory
            knowledge_type=request.knowledge_type,
            require_human_review=request.require_human_review
        )

        # Ensure compiled graph is available
        compiled_graph = getattr(http_request.app.state, "compiled_graph", None)
        if compiled_graph is None:
            compiled_graph = graph_builder.build()
            http_request.app.state.compiled_graph = compiled_graph

        # Execute the query workflow with thread ID for context
        logger.info(f"🔍 Starting query workflow: {thread_id}")
        result = compiled_graph.invoke(
            state,
            config={"configurable": {"thread_id": thread_id}}
        )

        # Save conversation to custom memory manager
        conversation_memory.add_message(thread_id, "user", request.query)
        if result.get("final_answer"):
            conversation_memory.add_message(thread_id, "assistant", result["final_answer"])

        # Calculate execution time for performance monitoring
        execution_time = (datetime.utcnow() - start_time).total_seconds()

        logger.info(f"✅ Query completed in {execution_time:.2f}s")
        return WorkflowResponse(
            success=True,
            thread_id=thread_id,
            result=result,
            execution_time=execution_time,
            timestamp=datetime.utcnow()
        )

    except Exception as e:
        logger.error(f"❌ Query failed: {e}")
        raise HTTPException(
            status_code=500,
            detail=f"Query execution failed: {str(e)}"
        )


@router.post("/ingest-pdfs", response_model=MultiIngestResponse)
async def ingest_multiple_pdfs(
    files: List[UploadFile] = File(...),
    source: str = Form(...),
    categories: Optional[str] = Form(None),
    author: Optional[str] = Form(None),
    metadata: Optional[str] = Form(None),
    graph_builder: MasterGraphBuilder = Depends(get_graph_builder),
    http_request: Request = None
):

    start_time = datetime.utcnow()

    # Validate input
    if not files:
        raise HTTPException(
            status_code=400,
            detail="No files provided. Please upload at least one PDF file."
        )

    # Parse categories and metadata
    category_list = []
    if categories:
        category_list = [cat.strip() for cat in categories.split(",") if cat.strip()]

    metadata_dict = {}
    if metadata:
        try:
            metadata_dict = json.loads(metadata)
        except json.JSONDecodeError:
            logger.warning(f"Invalid JSON metadata provided: {metadata}")
            metadata_dict = {}

    # Add common metadata
    metadata_dict.update({
        "author": author or "Unknown",
        "source": source,
        "categories": category_list,
        "ingestion_date": datetime.utcnow().isoformat(),
        "batch_ingest": True
    })

    results = []
    processed_count = 0
    failed_count = 0

    # Ensure compiled graph is available
    compiled_graph = getattr(http_request.app.state, "compiled_graph", None)
    if compiled_graph is None:
        compiled_graph = graph_builder.build()
        http_request.app.state.compiled_graph = compiled_graph

    logger.info(f"🔄 Starting batch PDF ingestion: {len(files)} files")

    # Process each PDF file
    for file in files:
        file_start_time = datetime.utcnow()

        try:
            # Validate file type
            if not file.filename or not file.filename.lower().endswith('.pdf'):
                results.append(PDFIngestResult(
                    filename=file.filename or "unknown",
                    success=False,
                    error="Invalid file type. Only PDF files are supported.",
                    processing_time=0.0
                ))
                failed_count += 1
                continue

            # Extract text from PDF
            pdf_content = await extract_pdf_text(file)

            # Create knowledge state for ingestion
            state = KnowledgeState(
                query_type="ingest",
                raw_document=pdf_content,
                source=f"{source}_{file.filename}",
                categories=category_list,
                metadata={**metadata_dict, "original_filename": file.filename}
            )

            # Generate unique thread ID for this PDF
            thread_id = f"pdf_ingest_{int(file_start_time.timestamp())}_{file.filename}"

            # Execute the ingestion workflow
            logger.info(f"📄 Processing PDF: {file.filename}")
            result = compiled_graph.invoke(
                state,
                config={"configurable": {"thread_id": thread_id}}
            )

            # Calculate processing time for this file
            file_processing_time = (datetime.utcnow() - file_start_time).total_seconds()

            # Count chunks created
            chunks_created = len(result.get("chunks", []))

            results.append(PDFIngestResult(
                filename=file.filename,
                success=True,
                chunks_created=chunks_created,
                thread_id=thread_id,
                processing_time=file_processing_time
            ))
            processed_count += 1

            logger.info(f"✅ PDF processed successfully: {file.filename} ({chunks_created} chunks)")

        except Exception as e:
            file_processing_time = (datetime.utcnow() - file_start_time).total_seconds()
            error_msg = str(e)

            logger.error(f"❌ Failed to process PDF {file.filename}: {error_msg}")

            results.append(PDFIngestResult(
                filename=file.filename or "unknown",
                success=False,
                error=error_msg,
                processing_time=file_processing_time
            ))
            failed_count += 1

    # Calculate total execution time
    execution_time = (datetime.utcnow() - start_time).total_seconds()

    logger.info(f"🏁 Batch PDF ingestion completed: {processed_count} successful, {failed_count} failed in {execution_time:.2f}s")

    return MultiIngestResponse(
        success=processed_count > 0,
        total_files=len(files),
        processed_files=processed_count,
        failed_files=failed_count,
        results=results,
        execution_time=execution_time,
        timestamp=datetime.utcnow()
    )


@router.post("/feedback", response_model=FeedbackResponse)
async def submit_feedback(
    request: FeedbackRequest,
    graph_builder: MasterGraphBuilder = Depends(get_graph_builder),
    http_request: Request = None
):

    start_time = datetime.utcnow()

    try:
        # Validate feedback type
        valid_feedback_types = ["approved", "rejected", "edited"]
        if request.feedback not in valid_feedback_types:
            raise HTTPException(
                status_code=400,
                detail=f"Invalid feedback type. Must be one of: {valid_feedback_types}"
            )

        # Validate edits are provided when feedback is "edited"
        if request.feedback == "edited" and not request.edits:
            raise HTTPException(
                status_code=400,
                detail="Edits must be provided when feedback type is 'edited'"
            )

        # Validate knowledge type if provided
        if request.knowledge_type:
            valid_knowledge_types = ["conversational", "reusable", "verified"]
            if request.knowledge_type not in valid_knowledge_types:
                raise HTTPException(
                    status_code=400,
                    detail=f"Invalid knowledge type. Must be one of: {valid_knowledge_types}"
                )

        logger.info(f"📝 Processing feedback for thread {request.thread_id}: {request.feedback}")

        # Get the compiled graph
        compiled_graph = getattr(http_request.app.state, "compiled_graph", None)
        if compiled_graph is None:
            compiled_graph = graph_builder.build()
            http_request.app.state.compiled_graph = compiled_graph

        # Create a new state with the feedback
        feedback_state = KnowledgeState(
            query_type="query",  # We're processing feedback on a query
            human_feedback=request.feedback,
            edits=request.edits,
            metadata={
                "feedback_comment": request.comment,
                "feedback_timestamp": start_time.isoformat(),
                "feedback_source": "api"
            }
        )

        # Process feedback through the workflow using LangGraph checkpointing
        try:
            # Get the current state from LangGraph checkpointing
            current_state = compiled_graph.get_state({"configurable": {"thread_id": request.thread_id}})

            if not current_state or not current_state.values:
                raise HTTPException(
                    status_code=404,
                    detail=f"No conversation found for thread_id: {request.thread_id}"
                )

            # Create a new state with the existing conversation data plus feedback
            existing_state = current_state.values
            logger.info(f"🔍 Creating feedback state with knowledge_type: {request.knowledge_type}")
            feedback_state = KnowledgeState(
                query_type=existing_state.get("query_type", "query"),
                user_input=existing_state.get("user_input"),
                messages=existing_state.get("messages", []),
                generated_answer=existing_state.get("generated_answer"),
                retrieved_docs=existing_state.get("retrieved_docs"),
                retrieved_chunks=existing_state.get("retrieved_chunks"),
                metadata=existing_state.get("metadata", {}),
                human_feedback=request.feedback,
                edits=request.edits,
                knowledge_type=request.knowledge_type,
                status=existing_state.get("status"),
                logs=existing_state.get("logs", [])
            )
            logger.info(f"🔍 Feedback state created with knowledge_type: {feedback_state.knowledge_type}")

            # Add feedback metadata
            if request.comment:
                feedback_state.metadata = feedback_state.metadata or {}
                feedback_state.metadata["feedback_comment"] = request.comment
                feedback_state.metadata["feedback_timestamp"] = start_time.isoformat()
                feedback_state.metadata["feedback_source"] = "api"

            # If feedback is "edited", update the edited_answer with the edits
            if request.feedback == "edited" and request.edits:
                feedback_state.edited_answer = request.edits
                logger.info(f"🔍 Updated edited_answer with edits: {request.edits[:100]}...")

            # Resume the graph from the interrupt by invoking with the updated state
            logger.info(f"🔄 Resuming workflow after human feedback")
            result = compiled_graph.invoke(
                feedback_state,
                config={"configurable": {"thread_id": request.thread_id}}
            )

            # Determine action taken based on feedback and knowledge type
            # Derive action from result
            if result.get("human_feedback") == "approved":
                if result.get("knowledge_type") in ("reusable", "verified"):
                    action_taken = f"Answer approved and stored as {result.get('knowledge_type')} knowledge in vector database"
                else:
                    action_taken = "Answer approved and stored in conversation history"
            elif result.get("human_feedback") == "rejected":
                action_taken = "Answer rejected, not stored in knowledge base"
            elif result.get("human_feedback") == "edited":
                if result.get("knowledge_type") in ("reusable", "verified"):
                    action_taken = f"Answer edited and stored as {result.get('knowledge_type')} knowledge in vector database"
                else:
                    action_taken = "Answer edited and stored in conversation history"
            else:
                action_taken = "Feedback processed"

            execution_time = (datetime.utcnow() - start_time).total_seconds()

            logger.info(f"✅ Feedback processed successfully in {execution_time:.2f}s")

            return FeedbackResponse(
                success=True,
                thread_id=request.thread_id,
                feedback=result.get("human_feedback", request.feedback),
                action_taken=action_taken,
                timestamp=datetime.utcnow(),
                message="Feedback processed successfully"
            )

        except Exception as e:
            logger.error(f"❌ Error processing feedback: {e}")
            raise HTTPException(
                status_code=500,
                detail=f"Failed to process feedback: {str(e)}"
            )

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"❌ Unexpected error in feedback submission: {e}")
        raise HTTPException(
            status_code=500,
            detail=f"Internal server error: {str(e)}"
        )

@router.get("/feedback/{thread_id}", response_model=FeedbackStatusResponse)
async def get_feedback_status(
    thread_id: str,
    graph_builder: MasterGraphBuilder = Depends(get_graph_builder),
    http_request: Request = None
):

    try:
        logger.info(f"🔍 Checking feedback status for thread: {thread_id}")

        # Get the compiled graph
        compiled_graph = getattr(http_request.app.state, "compiled_graph", None)
        if compiled_graph is None:
            compiled_graph = graph_builder.build()
            http_request.app.state.compiled_graph = compiled_graph

        # Try to get the current state from LangGraph checkpointing
        try:
            # Get the current state from the compiled graph's checkpointing system
            current_state = compiled_graph.get_state({"configurable": {"thread_id": thread_id}})

            if not current_state or not current_state.values:
                raise HTTPException(
                    status_code=404,
                    detail=f"No conversation found for thread_id: {thread_id}"
                )

            # Extract the KnowledgeState from the checkpoint
            state_values = current_state.values

            # Check if there's a pending answer awaiting feedback
            has_pending_feedback = (
                state_values.get("generated_answer") is not None and 
                state_values.get("human_feedback") is None
            )

            # Get current answer if pending
            current_answer = state_values.get("generated_answer") if has_pending_feedback else None

            # Build feedback history from conversation metadata
            feedback_history = []
            metadata = state_values.get("metadata", {})
            if metadata and "feedback_history" in metadata:
                feedback_history = metadata["feedback_history"]

            # Add current feedback if it exists
            if state_values.get("human_feedback"):
                current_feedback = {
                    "feedback": state_values.get("human_feedback"),
                    "timestamp": datetime.utcnow().isoformat(),
                    "comment": metadata.get("feedback_comment") if metadata else None
                }
                feedback_history.append(current_feedback)

            # Determine current status
            if has_pending_feedback:
                status = "awaiting_feedback"
            elif state_values.get("human_feedback") == "approved":
                status = "approved"
            elif state_values.get("human_feedback") == "rejected":
                status = "rejected"
            elif state_values.get("human_feedback") == "edited":
                status = "edited"
            else:
                status = "unknown"

        except Exception as e:
            logger.warning(f"Could not retrieve state from LangGraph checkpointing: {e}")
            # Fallback: check if there's any conversation history in Redis
            conversation_history = conversation_memory.get_conversation_history(thread_id)
            if not conversation_history:
                raise HTTPException(
                    status_code=404,
                    detail=f"No conversation found for thread_id: {thread_id}"
                )

            # If we have conversation history but no checkpoint state, assume no pending feedback
            has_pending_feedback = False
            current_answer = None
            feedback_history = []
            status = "conversation_exists"

        logger.info(f"✅ Feedback status retrieved: {status}")

        return FeedbackStatusResponse(
            thread_id=thread_id,
            has_pending_feedback=has_pending_feedback,
            current_answer=current_answer,
            feedback_history=feedback_history,
            status=status
        )

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"❌ Error retrieving feedback status: {e}")
        raise HTTPException(
            status_code=500,
            detail=f"Failed to retrieve feedback status: {str(e)}"
        )

The /ingest and /query endpoints execute document ingestion and query workflows, respectively. The /feedback endpoints (POST and GET by thread ID) handle human feedback and allow the system to resume after interruptions using that feedback. Moreover there are two ingestion endpoints — /ingest and /ingest-pdfs, to make bulk PDF ingestion straightforward.

Finally, each API endpoint will access the get_graph_builder() method for creating the graph object, which I have showed in the previous article:

def get_graph_builder(request: Request) -> MasterGraphBuilder:

    # Use app.state; lazily initialize if missing
    graph_builder = getattr(request.app.state, "graph_builder", None)
    compiled_graph = getattr(request.app.state, "compiled_graph", None)

    if graph_builder is None:
        # Extract configuration from environment settings
        api_key = settings.openai_api_key
        azure_endpoint = settings.azure_openai_endpoint_url

        # Validate required configuration
        if not api_key:
            raise HTTPException(
                status_code=500,
                detail="OpenAI API key not configured"
            )

        try:
            # =============================================================================
            # MODEL INITIALIZATION
            # =============================================================================

            # Import Azure OpenAI models for embeddings and language generation
            from langchain_openai import AzureOpenAIEmbeddings, AzureChatOpenAI

            # Initialize Azure embedding model for document vectorization
            embedding_model = AzureOpenAIEmbeddings(
                azure_deployment="text-embedding-3-small",  # Your embedding deployment name
                openai_api_version="2024-12-01-preview",    # Azure OpenAI API version
                azure_endpoint=azure_endpoint,              # Azure service endpoint
                openai_api_key=api_key                      # API key for authentication
            )

            # Initialize Azure language model for text generation and reasoning
            llm = AzureChatOpenAI(
                azure_deployment="gpt-4o",                  # Your LLM deployment name
                openai_api_version="2024-12-01-preview",    # Azure OpenAI API version
                azure_endpoint=azure_endpoint,              # Azure service endpoint
                openai_api_key=api_key,                     # API key for authentication
                temperature=0.1                             # Low temperature for consistent outputs
            )

            # =============================================================================
            # VECTOR STORE INITIALIZATION
            # =============================================================================

            # Initialize ChromaDB vector store for document storage and retrieval
            from langchain_chroma import Chroma
            vectorstore = Chroma(
                collection_name="smart_second_brain",        # Collection name for documents
                embedding_function=embedding_model,          # Function to create embeddings
                persist_directory="./chroma_db"              # Local storage directory
            )

            # =============================================================================
            # GRAPH BUILDER INITIALIZATION
            # =============================================================================

            # Create and configure the main workflow orchestrator
            graph_builder = MasterGraphBuilder(
                llm=llm,                                    # Language model for reasoning
                embedding_model=embedding_model,             # Embedding model for vectorization
                vectorstore=vectorstore,                    # Vector database for storage
                chromadb_dir="./chroma_db"                  # ChromaDB storage directory
            )

            # Compile the workflow for execution
            compiled_graph = graph_builder.build()
            request.app.state.graph_builder = graph_builder
            request.app.state.compiled_graph = compiled_graph
            logger.info("✅ Graph builder initialized with all models")

        except Exception as e:
            logger.error(f"❌ Failed to initialize graph builder: {e}")
            raise HTTPException(
                status_code=500,
                detail=f"Failed to initialize graph: {str(e)}"
            )

    return graph_builder

The API now is ready, next is the UI.

Let's rock with Streamlit

I initially considered using Vue for the frontend, but as a backend developer, Vue or React felt a bit overwhelming. In the end, I still prefer Streamlit.

Since we now have a dedicated API, the Streamlit frontend will interact with the graph exclusively through that API.

I won’t dive into the Streamlit code here—feel free to check it out in the repository.

PDFs Ingestion Tab

Texts Ingestion Tab

Chat Program

As mentioned earlier, we support both batch PDF ingestion and generic text ingestion, each handled in its own tab.

Now let’s see a real chat in action with continuous knowledge updates.

As someone who grew up on a steady diet of Japanese animation, Gundam is one of my all-time favorites—so I’ll chat about it with my 2nd Brain!

And since there is no information provided about Gundam before in the vector DB, 2nd brain replies "I don't know". (Notice that a new conversation ID is created) I can provide further information about what Gundam is:

I copied the quick description about Gundam from web and confirm, 2nd brain will save it back to the vector DB. When I ask again the same question 2nd brain now can answer:

And if I start a new conversation with a new thread id, since the information has been stored, it can be answered right-away:

But in honest speaking, building a user-friendly chat UI which coherent and consistent flow of information, that facilitates continuous user-provided information update, is indeed a challenging task.

Ready to level up - what's our play?

The 2nd Brain is shaping up beautifully—so how do we make it even smarter and more powerful? Let’s sprinkle in a bit of automation to dial up the magic.

Stay tuned for the final chapter!

(The project source code will be provided when the entire series is concluded.)

Building My Smart 2nd Brain, Part 2: Human-in-the-Loop Querying with Robust Checkpointing

Jimmy — Wed, 24 Sep 2025 12:46:12 +0000

From part 1, I kicked off the Smart 2nd Brain with two execution paths. Previously, we walked through the ingestion flow—splitting documents, generating embeddings, and storing them in a vector database. In this installment, we’ll switch gears to the query path and see how those embeddings power retrieval and human-in-the-loop answers.

Asking the Brain

Now let's go through the query path:

        # =============================================================================
        # QUERY BRANCH
        # =============================================================================

        # Knowledge query pipeline
        graph.add_edge("retriever", "answer")           # Retrieve -> Generate
        graph.add_edge("answer", "review")             # Generate -> Review
        graph.add_edge("review", "validated_store")    # Review -> Store
        graph.add_edge("validated_store", END)         # Store -> End

First, the retriever node.

The retriever_node function queries the vector database based on the user’s input.

def retriever_node(self, state: KnowledgeState):

        if not state.user_input:
            state.logs = (state.logs or []) + ["No user_input provided for retrieval"]
            state.retrieved_docs = []
            return state

        try:
            # Try different retrieval strategies in order of preference
            if self.retriever:
                logger.info(f"🔍 Retrieving docs for query: {state.user_input}")
                results = self.retriever.get_relevant_documents(state.user_input)
            elif self.vectorstore:
                logger.info(f"🔍 Using vectorstore.similarity_search for query: {state.user_input}")
                results = self.vectorstore.similarity_search(state.user_input, k=5)
            else:
                results = []

            # Format retrieved documents for downstream processing
            state.retrieved_docs = [{"content": r.page_content, "metadata": r.metadata} for r in results]
            state.logs = (state.logs or []) + [f"Retrieved {len(state.retrieved_docs)} docs"]

        except Exception as e:
            logger.error(f"❌ Retrieval failed: {e}")
            state.retrieved_docs = []
            state.logs = (state.logs or []) + [f"Retrieval error: {str(e)}"]

        return state

Please note that it uses two approaches for search: one via the retriever and another directly via the vector store.

try:
    # Try different retrieval strategies in order of preference
    if self.retriever:
        logger.info(f"🔍 Retrieving docs for query: {state.user_input}")
        results =self.retriever.get_relevant_documents(state.user_input)
    elif self.vectorstore:
        logger.info(f"🔍 Using vectorstore.similarity_search for query: {state.user_input}")
        results = self.vectorstore.similarity_search(state.user_input, k=5)
    else:
        results = []

A vector store is the database/index that stores embeddings and exposes low-level search primitives (e.g., similarity, filters), while a retriever is a higher-level interface that decides how to search, applies strategy (k, thresholds, filters, reranking), and returns documents via a consistent “get relevant docs” contract. In practice, a retriever is preferred because it abstracts backend specifics, enables easy swapping of stores, supports advanced strategies and fallbacks, and centralizes logic like query rewriting, multi-step retrieval, and graceful error handling—so applications stay flexible and robust as requirements or infrastructure change.

The retriever_node here will always prefer using retriever as long as one is provided. When initialize the state object, the retriever has to be created like this:

vectorstore = Chroma(
    collection_name="knowledge_base",
    embedding_function=embedding_model,
    persist_directory="./chroma_db",
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

embedding_model here is configured by AzureOpenAiEmbeddings, with the vectorstore object we use the as_retriever method to create the retreiver instance. The "k" parameter within search_kwargs specifies the number of top-k most similar (or relevant) documents to retrieve from the vector store during a similarity search.

If retriever is not available, similarity_search will be applied directly on the vectorstore object, bypassing the interface.

Once the relevant documents are retrieved, they can be sent to the answer node for processing using a LLM:

def answer_gen_node(self, state: KnowledgeState):

        if not self.llm or not state.user_input:
            state.logs = (state.logs or []) + ["⚠️ No LLM or user query provided"]
            return state

        try:
            # --- Build retrieved context ---
            context = "\n\n".join(
                [doc["content"] for doc in (state.retrieved_docs or [])]
            ) or "No relevant documents were retrieved."

            # --- Format conversation history ---
            conversation = "\n".join(
                [f"{m['role']}: {m['content']}" for m in (state.messages or [])]
            )

            # --- Load externalized prompt template ---
            prompt_path = Path(__file__).resolve().parent.parent / "prompts" / "answer_prompt.txt"
            with open(prompt_path, "r", encoding="utf-8") as f:
                template = f.read()
            prompt = ChatPromptTemplate.from_template(template)
            chain = prompt | self.llm

            # --- Call language model with full context ---
            response = chain.invoke({
                "conversation": conversation,
                "query": state.user_input,
                "context": context
            })

            # --- Update state with generated answer ---
            state.generated_answer = response.content

            # Initialize messages array if not exists
            if not state.messages:
                state.messages = []

            # Add current user input and assistant response to conversation history
            # Note: Conversation memory is now managed by custom ConversationMemoryManager
            state.messages.append({"role": "user", "content": state.user_input})
            state.messages.append({"role": "assistant", "content": state.generated_answer})
            state.logs = (state.logs or []) + ["✅ Generated contextual answer with conversation history"]

        except Exception as e:
            state.generated_answer = None
            state.logs = (state.logs or []) + [f"❌ Answer generation failed: {e}"]

        return state

The answer node merges the user’s query with the documents retrieved by the retriever node and sends that context to the LLM. It loads the prompt from /prompts/answer_prompt.txt before invoking the model.

You are a helpful knowledge assistant with access to conversation history.

CONVERSATION HISTORY:
{conversation}

CURRENT USER QUESTION:
{query}

RETRIEVED KNOWLEDGE BASE CONTEXT:
{context}

INSTRUCTIONS:
- Use the conversation history to understand context and references
- Base your answer primarily on the retrieved knowledge base context
- If the knowledge base context is insufficient, say "I don't know based on available knowledge."
- Consider previous questions and answers to provide better context
- Keep the answer clear, concise, and contextually relevant
- If the user refers to something from earlier in the conversation, acknowledge it

Beyond the user query and retrieved documents, the conversation history is also injected to give the LLM richer context.

Note that the prompt includes an instruction like: “If the knowledge base context is insufficient, say ‘I don’t know based on available knowledge.’” This constrains the model to rely solely on the vector database rather than its pretrained knowledge. Whether you should strictly limit answers to the vector store is debatable, but we’ll adopt this constraint for now in this project.

Next comes the review stage, where HITL (Human-In-The-Loop) takes place.

def human_review_node(self, state: KnowledgeState):

        try:
            # Check for explicit human feedback
            feedback = getattr(state, "human_feedback", None)

            # Determine whether human review is required
            require_review = getattr(state, "require_human_review", None)
            if require_review is None:
                # Infer from knowledge_type when not explicitly set
                require_review = getattr(state, "knowledge_type", "conversational") in ("reusable", "verified")

            # If review is required, no feedback yet, and interrupt is available, pause the graph here
            if require_review and not feedback and interrupt is not None:
                # Message guides the API/UI to provide the required fields back
                _ = interrupt(
                    "Awaiting human review: set 'human_feedback' to one of"
                    " ['approved','rejected','edited'] and optionally set"
                    " 'edited_answer' when feedback is 'edited'."
                )
                # Execution will resume here once feedback is supplied via the graph resume
                feedback = getattr(state, "human_feedback", None)

            if not feedback:
                # Auto-approval path when running headless or interrupt not available
                feedback = "approved"
                state.final_answer = state.generated_answer
                state.logs = (state.logs or []) + ["✅ Auto-approved answer"]

            elif feedback == "rejected":
                # Handle rejection - no final answer
                state.final_answer = None
                state.logs = (state.logs or []) + ["❌ Answer was rejected by human"]

            elif feedback == "edited":
                # Handle manual editing
                if hasattr(state, "edited_answer") and state.edited_answer:
                    state.final_answer = state.edited_answer
                    state.logs = (state.logs or []) + ["✏️ Human edited the answer"]
                else:
                    state.final_answer = None
                    state.logs = (state.logs or []) + ["⚠️ Edit requested but no edited_answer provided"]

            else:
                # Unknown feedback type - default to generated answer
                state.final_answer = state.generated_answer
                state.logs = (state.logs or []) + [f"⚠️ Unknown feedback '{feedback}', defaulting to generated answer"]

            # Save feedback for traceability
            state.human_feedback = feedback

        except Exception as e:
            # Error handling - default to generated answer
            state.final_answer = state.generated_answer
            state.human_feedback = "error"
            state.logs = (state.logs or []) + [f"❌ Human review failed: {e}"]

        return state

LangGraph supports HITL by allowing a running graph to pause at specific nodes and wait for external human input before continuing. The mechanism is the interrupt function: when a node calls interrupt(...) during execution, LangGraph saves the checkpointed state and returns control to the caller with a payload that indicates what inputs are expected (e.g., human_feedback, edited_answer). An external system (UI/API) collects the human response and then resumes the graph with those fields injected into the state.

If interrupt isn’t available or you’re running headless, you can implement fallbacks (like auto-approval) to keep the workflow progressing.

The human_review_node here orchestrates human-in-the-loop validation for generated answers. It first checks human_feedback and whether review is required (explicit via require_human_review or inferred when knowledge_type is reusable or verified). If review is required, no feedback is present, and interrupt is available, it calls interrupt(...) to pause execution and wait for external input, instructing the caller to provide human_feedback (approved, rejected, edited) and optionally edited_answer.

On resume (or when running headless), it applies the logic: auto-approve by default, reject clears final_answer, and edited sets final_answer to edited_answer (with a warning if missing). It records decisions in state.final_answer, state.human_feedback, and appends trace logs; errors fall back to using the generated answer.

Later when demonstrating the API and UI interfaces the HITL feature will be more apparent.

Now the generated result is approved, we can save it properly back to the vector DB:

def validated_store_node(self, state: KnowledgeState):

        logger.info(f"🔍 validated_store_node called with knowledge_type: {getattr(state, 'knowledge_type', 'None')}")
        logger.info(f"🔍 final_answer: {getattr(state, 'final_answer', 'None')}")
        logger.info(f"🔍 human_feedback: {getattr(state, 'human_feedback', 'None')}")
        try:
            # Check for answer to validate
            if not state.final_answer:
                logger.info("⚠️ No final_answer to validate, skipping")
                state.logs = (state.logs or []) + ["⚠️ No final_answer to validate, skipping"]
                return state

            # Check feedback status
            if state.human_feedback not in ("approved", "edited"):
                logger.info(f"ℹ️ Skipping validation because feedback = {state.human_feedback}")
                state.logs = (state.logs or []) + [f"ℹ️ Skipping validation because feedback = {state.human_feedback}"]
                return state

            # Determine knowledge type (default to conversational)
            knowledge_type = getattr(state, "knowledge_type", "conversational")

            # Always store in checkpoint memory for conversation continuity
            # Mark as validated at this point; downstream vector storage may refine status
            state.status = "validated"
            state.logs = (state.logs or []) + ["✅ Conversation history managed by LangGraph checkpoint memory (thread-isolated)"]

            # Check if content should also be stored in vector DB
            if knowledge_type in ("reusable", "verified"):
                try:
                    # Create a document for vector storage
                    from langchain.schema import Document

                    # Prepare metadata for vector storage
                    metadata = {
                        "source": f"human_approved_{state.query_type}",
                        "thread_id": getattr(state, "thread_id", "unknown"),
                        "knowledge_type": knowledge_type,
                        "human_feedback": state.human_feedback,
                        "timestamp": datetime.datetime.now().isoformat(),
                        "user_input": state.user_input or "unknown_query"
                    }

                    # Add any existing metadata
                    if state.metadata:
                        metadata.update(state.metadata)

                    # Create document for vector storage
                    doc = Document(
                        page_content=state.final_answer,
                        metadata=metadata
                    )

                    # Store in vector database
                    if self.vectorstore:
                        logger.info(f"📚 Attempting to store {knowledge_type} knowledge in vector DB...")
                        logger.info(f"📚 Document content: {doc.page_content[:100]}...")
                        logger.info(f"📚 Document metadata: {doc.metadata}")
                        self.vectorstore.add_documents([doc])
                        state.logs = (state.logs or []) + [f"✅ {knowledge_type.title()} knowledge stored in vector database"]
                        # Refine status to indicate successful vector storage
                        state.status = "stored"
                        logger.info(f"📚 Successfully stored {knowledge_type} knowledge in vector DB")
                    else:
                        state.logs = (state.logs or []) + ["⚠️ Vector store not available, skipping vector storage"]
                        # Refine status to indicate validation without vector storage
                        state.status = "validated_no_store"
                        logger.warning("⚠️ Vector store not available, skipping vector storage")

                except Exception as e:
                    state.logs = (state.logs or []) + [f"❌ Failed to store in vector DB: {e}"]
                    # Refine status to indicate vector storage failure after validation
                    state.status = "store_failed"
                    logger.error(f"Failed to store knowledge in vector DB: {e}")
            else:
                # Conversational content - checkpoint memory only
                state.logs = (state.logs or []) + ["💬 Conversational content stored in checkpoint memory only"]

        except Exception as e:
            state.status = "error"
            state.logs = (state.logs or []) + [f"❌ Validation failed: {e}"]

        return state

validated_store_node checks whether there’s a reviewed answer to persist, then records the validation outcome in the conversation’s checkpointed state and, if appropriate, promotes the answer into long‑term knowledge. It first exits early if there’s no final_answer or if human_feedback isn’t an approval/edited outcome. It then assembles a metadata payload (thread_id, source, timestamps, knowledge_type, feedback, user_input, plus any extra state metadata).

If the knowledge_type indicates the content is reusable/verified and a vectorstore is available, it wraps the final_answer as a LangChain Document with that metadata and attempts to add it to the vector database (e.g., Chroma). Errors in storage are caught and logged without breaking the run. If the content is conversational or storage isn’t configured, it skips vector persistence but still marks the state as validated so the conversation history is preserved via LangGraph checkpoints.

After this node, the execution of querying reaches to its end.

Keeping Everything Pick‑Up‑Where‑You‑Left‑Off Friendly

LangGraph uses a checkpointer to persist graph state at well-defined boundaries so runs can pause, resume, and recover reliably. Each execution is associated with a thread (or run) ID; after each node transition, the current state snapshot—inputs, outputs, and metadata—is written to the checkpointer backend. On resume or retry, LangGraph reloads the latest committed snapshot for that thread, restoring variables and control flow without redoing completed work. This makes human-in-the-loop interrupts, failures, and restarts safe and idempotent, effectively giving you durable conversation history and step-by-step progress tracking.

LangGraph offers built-in checkpointers like SqliteSaver for durable, file-backed persistence and MemorySaver for fast but in-memory runs. Beyond these, you can integrate production stores via community patterns (e.g., Postgres/SQLAlchemy) or create your own by implementing the checkpointer interface against systems like Redis, DynamoDB, or S3.

In the Smart 2nd Brain system, we use the SqliteSaver to implement an easy checkpointer inside the MasterGraphBuilder class:

# Initialize SqliteSaver checkpointer (no external service required)
# Stores checkpoints in ./data/checkpoints.sqlite
os.makedirs("data", exist_ok=True)
sqlite_conn = sqlite3.connect("data/checkpoints.sqlite", check_same_thread=False)
self.checkpointer = SqliteSaver(sqlite_conn)

When the compile graph is returned, remember to include checkpointer as an argument:

# Compile the graph with checkpointing
return graph.compile(checkpointer=self.checkpointer)

See? Pretty simple to do.

I have just walkthrough all nodes in different execution paths, here is the code to build the graph object in runtime:

def get_graph_builder(request: Request) -> MasterGraphBuilder:

    # Use app.state; lazily initialize if missing
    graph_builder = getattr(request.app.state, "graph_builder", None)
    compiled_graph = getattr(request.app.state, "compiled_graph", None)

    if graph_builder is None:
        # Extract configuration from environment settings
        api_key = settings.openai_api_key
        azure_endpoint = settings.azure_openai_endpoint_url

        # Validate required configuration
        if not api_key:
            raise HTTPException(
                status_code=500,
                detail="OpenAI API key not configured"
            )

        try:
            # =============================================================================
            # MODEL INITIALIZATION
            # =============================================================================

            # Import Azure OpenAI models for embeddings and language generation
            from langchain_openai import AzureOpenAIEmbeddings, AzureChatOpenAI

            # Initialize Azure embedding model for document vectorization
            embedding_model = AzureOpenAIEmbeddings(
                azure_deployment="text-embedding-3-small",  # Your embedding deployment name
                openai_api_version="2024-12-01-preview",    # Azure OpenAI API version
                azure_endpoint=azure_endpoint,              # Azure service endpoint
                openai_api_key=api_key                      # API key for authentication
            )

            # Initialize Azure language model for text generation and reasoning
            llm = AzureChatOpenAI(
                azure_deployment="gpt-4o",                  # Your LLM deployment name
                openai_api_version="2024-12-01-preview",    # Azure OpenAI API version
                azure_endpoint=azure_endpoint,              # Azure service endpoint
                openai_api_key=api_key,                     # API key for authentication
                temperature=0.1                             # Low temperature for consistent outputs
            )

            # =============================================================================
            # VECTOR STORE INITIALIZATION
            # =============================================================================

            # Initialize ChromaDB vector store for document storage and retrieval
            from langchain_chroma import Chroma
            vectorstore = Chroma(
                collection_name="smart_second_brain",        # Collection name for documents
                embedding_function=embedding_model,          # Function to create embeddings
                persist_directory="./chroma_db"              # Local storage directory
            )

            # =============================================================================
            # GRAPH BUILDER INITIALIZATION
            # =============================================================================

            # Create and configure the main workflow orchestrator
            graph_builder = MasterGraphBuilder(
                llm=llm,                                    # Language model for reasoning
                embedding_model=embedding_model,             # Embedding model for vectorization
                vectorstore=vectorstore,                    # Vector database for storage
                chromadb_dir="./chroma_db"                  # ChromaDB storage directory
            )

            # Compile the workflow for execution
            compiled_graph = graph_builder.build()
            request.app.state.graph_builder = graph_builder
            request.app.state.compiled_graph = compiled_graph
            logger.info("✅ Graph builder initialized with all models")

        except Exception as e:
            logger.error(f"❌ Failed to initialize graph builder: {e}")
            raise HTTPException(
                status_code=500,
                detail=f"Failed to initialize graph: {str(e)}"
            )

    return graph_builder

This method is used in the FastAPI layer, you can also see how the embedding model, LLM and the chroma database vector store are created and injected into the graph builder.

Once the caller get the compiled graph object, it can be executed by calling the invoke method.

Wrapping Up: Core Logic Locked In — APIs and UI Coming Next

Now the core LangGraph implementation is wrapped up.

As a personal productivity tool, a clean, elegant UI matters — and we’ll front it with an API to control graph access and enable better testing. Don’t miss the next chapter: we’ll dive into the fun part with the API and user interface.

(The project source code will be provided when the entire series is concluded.)

Building My Smart 2nd Brain: Part 1 - Agentic AI with RAG

Jimmy — Wed, 17 Sep 2025 15:48:14 +0000

Overview

In my previous articles, I developed two LangGraph applications utilizing distinct personas for generating blog articles and images, with web search capabilities as tools. To complete this learning journey with LangGraph, I will now build a new application incorporating the following features:

Text embedding using a vector database for Retrieval-Augmented Generation (RAG).
Long-term memory using Sqlite to store message history.
Human-in-the-Loop (HITL) functionality for user verification of generated results.

This new project, my “Smart 2nd Brain”, is a personal knowledge management system designed to make my life more organized and spark creativity. RAG, long-term memory and HITL aren’t groundbreaking, but combining them felt like building a custom toolbox for my brain. The vector DB made retrieving my scattered notes a breeze, Sqlite gave the system a memory that rivals my own, and HITL kept me in control, letting me tweak outputs to feel just right.

This “Smart 2nd Brain” was a rewarding step in my LangGraph journey, blending AI smarts with personal flair. It’s more than a tool — it’s like an extension of my mind, helping me organize thoughts, rediscover ideas, and fuel curiosity.

The Knowledge State Object

Let's tale a look of the master state object:

from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field

class KnowledgeState(BaseModel):
    messages: List[Dict[str, str]] = Field(
        default_factory=list,
        description="Conversation-style messages exchanged between user, AI, and system."
    )
    user_input: Optional[str] = Field(
        None, description="Raw input from user (query or new document)."
    )
    query_type: Optional[str] = Field(
        None, description="Either 'ingest' for new document or 'query' for retrieval."
    )  
    raw_document: Optional[str] = Field(
        None, description="Full raw text of the ingested document."
    )
    chunks: Optional[List[str]] = Field(
        None, description="Text chunks after splitting for embedding."
    )
    embeddings: Optional[List[List[float]]] = Field(
        None, description="Vector embeddings for the chunks."
    )
    source: Optional[str] = Field(
        None, description="Source of the document (e.g., filename, URL, transcript ID)."
    )
    categories: Optional[List[str]] = Field(
        None, description="One or more categories/tags for the document."
    )
    metadata: Optional[Dict[str, Any]] = Field(
        default_factory=dict,
        description="Additional metadata (author, date, page, etc.)."
    )
    retrieved_docs: Optional[List[Dict[str, Any]]] = Field(
        None, description="Relevant documents retrieved from vector DB."
    )
    retrieved_chunks: Optional[List[str]] = Field(
        None, description="Raw chunks retrieved from the vector DB."
    )
    generated_answer: Optional[str] = Field(
        None, description="AI-generated draft answer or summary."
    )
    final_answer: Optional[str] = Field(
        None, description="Final, human-approved answer or summary."
    )
    human_feedback: Optional[str] = Field(
        None, description="Feedback on AI output: 'approved' | 'rejected' | 'edited'."
    )
    edits: Optional[str] = Field(
        None, description="Manual corrections provided by the human."
    )
    edited_answer: Optional[str] = Field(
        None, description="The edited version of the answer provided by human feedback."
    )
    knowledge_type: Optional[str] = Field(
        None, description="Type of knowledge: 'conversational' | 'reusable' | 'verified'."
    )
    conversation_history: Optional[List[Dict[str, str]]] = Field(
        None, description="Running log of user/AI/system messages for context."
    )

    user_preferences: Optional[Dict[str, Any]] = Field(
        None, description="User customization, e.g. summary style, tone, etc."
    )
    status: Optional[str] = Field(
        None, description="Pipeline status: 'pending', 'processing', 'done', 'error'."
    )
    logs: Optional[List[str]] = Field(
        default_factory=list,
        description="Debug logs collected during pipeline execution."
    )

The KnowledgeState class includes all the details that are being passed in the course of the graph execution:

messages: Conversation-style messages for chat-like interactions
user_input: Raw user input (query or document content)
query_type: Workflow type identifier ('ingest' or 'query')
raw_document: Full document text for ingestion workflows
chunks: Text chunks after document splitting
embeddings: Vector representations of text chunks
source: Document source identifier
categories: Document classification tags
metadata: Additional structured metadata
retrieved_docs: Retrieved documents from vector database
retrieved_chunks: Raw text chunks from retrieval
generated_answer: AI-generated response draft
final_answer: Human-approved final response
human_feedback: Human feedback on AI outputs
edits: Manual corrections and edits
conversation_history: Complete conversation context
user_preferences: User customization settings
status: Current workflow status
logs: Debug and execution logs

Please note that in this KnowledgeState object some properties are cumulative (e.g. conversation_history) while some are reflecting just the latest values (e.g. generated_answer, final_answer); I will go into the details in later sections. Anyway, this state object facilitates the capability of the Smart 2nd Brain system, enables all required information can be passed throughout the graph.

The Tales of Two Paths

The application graph supports two different paths: "ingest" and "query".

The Smart 2nd Brain supports PDF document ingestion into a vector database for text embedding, this is the so-called "ingest" path. After the documents are stored into the vector database they are available for enriching the LLM to answer questions from users. We call this path as the "query" path.

In the MasterGraphBuilder class, this graph is created by the build method:

def build(self) -> CompiledStateGraph:
        """
        Build and compile the complete LangGraph workflow.

        This method constructs the workflow graph by defining nodes,
        setting up conditional routing, and establishing the flow
        between different processing stages.

        Returns:
            CompiledStateGraph: Compiled and checkpointed workflow graph

        Graph Structure:
            Entry Point: router
            Conditional Routing: Based on query_type
            Ingestion Branch: router -> chunk -> embed -> store -> END
            Query Branch: router -> retriever -> answer -> review -> validated_store -> END

        Features:
            - Conditional edges for workflow routing
            - Checkpointing for state persistence
            - Error handling and logging throughout
            - Human-in-the-loop integration
        """
        # Create the base state graph
        graph = StateGraph(KnowledgeState)

        # =============================================================================
        # NODE DEFINITIONS
        # =============================================================================

        # Add all workflow nodes
        graph.add_node("router", self.input_router)           # Entry point and routing
        graph.add_node("chunk", self.chunk_doc_node)          # Document chunking
        graph.add_node("embed", self.embed_node)              # Embedding generation
        graph.add_node("store", self.store_node)              # Document storage
        graph.add_node("retriever", self.retriever_node)      # Document retrieval
        graph.add_node("answer", self.answer_gen_node)        # Answer generation
        graph.add_node("review", self.human_review_node)      # Human review
        graph.add_node("validated_store", self.validated_store_node)  # Validated storage

        # Set the entry point
        graph.set_entry_point("router")

        # =============================================================================
        # CONDITIONAL ROUTING
        # =============================================================================

        # Define routing logic based on query_type
        def route_condition(state):
            if hasattr(state, 'query_type'):
                if state.query_type == "ingest":
                    return "ingest"
                elif state.query_type == "query":
                    return "query"
                else:
                    return "__end__"
            return "query"

        # Add conditional edges from router
        graph.add_conditional_edges("router", route_condition, {
            "ingest": "chunk",      # Route to document ingestion
            "query": "retriever",   # Route to knowledge query
            "__end__": END,         # End workflow for invalid types
        })

        # =============================================================================
        # INGESTION BRANCH
        # =============================================================================

        # Document processing pipeline
        graph.add_edge("chunk", "embed")      # Chunk -> Embed
        graph.add_edge("embed", "store")     # Embed -> Store
        graph.add_edge("store", END)         # Store -> End

        # =============================================================================
        # QUERY BRANCH
        # =============================================================================

        # Knowledge query pipeline
        graph.add_edge("retriever", "answer")           # Retrieve -> Generate
        graph.add_edge("answer", "review")             # Generate -> Review
        graph.add_edge("review", "validated_store")    # Review -> Store
        graph.add_edge("validated_store", END)         # Store -> End

        # Compile the graph with checkpointing
        return graph.compile(checkpointer=self.checkpointer)

Feeding the Brain Chunk by Chunk

Let's check the ingest branch first. If the graph is invoked for ingest, the first encountered is the chunk node.

def chunk_doc_node(self, state: KnowledgeState):
        if state.raw_document:
            # Use RecursiveCharacterTextSplitter for intelligent text segmentation
            splitter = RecursiveCharacterTextSplitter(
                chunk_size=500,     # Maximum characters per chunk
                chunk_overlap=50    # Overlap between consecutive chunks for context
            )
            state.chunks = splitter.split_text(state.raw_document)
            state.logs = (state.logs or []) + [
                f"Chunked document into {len(state.chunks)} chunks"
            ]
        else:
            state.logs = (state.logs or []) + ["No raw_document found for chunking"]

        return state

To ingest lengthy text, it must first be divided into smaller chunks for efficient processing. The chunk_doc_node function takes the raw_document and splits it into manageable segments for downstream tasks.

In LangGraph, the RecursiveCharacterTextSplitter facilitates this by breaking text into semantically coherent chunks, ideal for NLP applications like embeddings generation or RAG. It employs a hierarchical approach, using customizable separators (e.g., paragraphs, sentences, words) to prioritize natural boundaries and preserve context.

Here, we set chunk_size=500 and chunk_overlap=50, striking a balance between coherence and minimal redundancy. Once the state returns the chunks, the next step is generating embeddings.

def embed_node(self, state: KnowledgeState):
        # Use embedding model if available, otherwise fallback to placeholder
        if state.chunks:
            if self.embedding_model:
                try:
                    logger.info(f"🔤 Generating embeddings for {len(state.chunks)} chunks")
                    state.embeddings = self.embedding_model.embed_documents(state.chunks)
                    logger.info(f"✅ Generated {len(state.embeddings)} embeddings")
                except Exception as e:
                    logger.error(f"❌ Embedding generation failed: {e}")
                    # Fallback to placeholder embeddings for continued processing
                    state.embeddings = [[0.1, 0.2]] * len(state.chunks)
                    state.logs = (state.logs or []) + [f"Embedding error: {str(e)}"]
            else:
                logger.warning("⚠️ No embedding model provided, using placeholder embeddings")
                state.embeddings = [[0.1, 0.2]] * len(state.chunks)
        return state

This node transforms text chunks into high-dimensional vector embeddings that encode semantic meaning, enabling effective similarity searches. The embedding model handles this conversion, and LangChain’s AzureOpenAIEmbeddings class seamlessly integrates the model for robust embedding generation.

# Initialize Azure embedding model for document vectorization
embedding_model = AzureOpenAIEmbeddings(
                azure_deployment="text-embedding-3-small",  # Your embedding deployment name
                openai_api_version="2024-12-01-preview",    # Azure OpenAI API version
                azure_endpoint=azure_endpoint,              # Azure service endpoint
                openai_api_key=api_key                      # API key for authentication
            )

At the conclusion of the pipeline, the store_node function saves the embedded chunks into a vector database, utilizing Chroma, a widely adopted solution for vector storage.

The metadata, which includes multiple categories, is designed to enhance the querying of these embedded chunks during the query path, enabling more precise and context-aware retrieval.

def store_node(self, state: KnowledgeState):

        if state.embeddings and state.chunks:
            try:
                # Initialize vectorstore if not provided
                if not self.vectorstore:
                    self.vectorstore = Chroma(
                        collection_name="knowledge_base",
                        embedding_function=self.embedding_model,
                        persist_directory=self.chromadb_dir
                    )

                # Prepare metadata for each chunk             
                metadatas = [
                    {
                        "source": state.source or "unknown",
                        "categories": ", ".join(state.categories) if state.categories else "general",
                        "chunk_id": i
                    }
                    for i in range(len(state.chunks))
                ]

                # Store chunks with metadata in ChromaDB
                self.vectorstore.add_texts(
                    texts=state.chunks,
                    metadatas=metadatas
                )

                # Data is automatically persisted when using persist_directory

                state.status = "stored"
                state.logs = (state.logs or []) + [
                    f"Stored {len(state.chunks)} chunks in ChromaDB with categories {state.categories or ['general']}"
                ]
            except Exception as e:
                state.status = "error"
                state.logs = (state.logs or []) + [f"Storing failed: {e}"]
        else:
            state.logs = (state.logs or []) + ["No embeddings/chunks to store"]

        return state

More Exciting Features Await

Rather than overwhelming you with every detail now, I’ll explore the query path, long-term memory, and Human-in-the-Loop capabilities in upcoming articles. I’ll also showcase how to integrate a FastAPI suite with a sleek user interface to seamlessly connect all components. Stay tuned for what’s next!

(Source code will be provided when concluding this series)

My Journey into Agentic AI Development: AI Newsroom

Jimmy — Sun, 14 Sep 2025 14:40:49 +0000

Last time, I demonstrated a simple LangGraph application that directed an LLM to write on any given topic, incorporating real-time web searches and generating tailored images to accompany articles, each with a distinct writing style. Now, let’s switch things up.

Debate in the newsroom

Picture a lively morning meeting at a bustling news agency, 8:00 AM sharp. The research team has just delivered a fresh list of trending topics. Around a sleek conference table, five editors — Andrew, Edward, Bob, Ruby, and Susan — engage in a spirited debate over which stories to tackle for the next issue. At the head of the table, the Chief Editor, a commanding presence, presides with a no-nonsense gaze, ready to steer the discussion.

Andrew (leaning forward, voice firm): _Look, we have to cover the escalating conflict between Russia and Ukraine. It’s a powder keg — geopolitical maneuvering, military standoffs, and the ripple effects on global stability. This isn’t just news; it’s a matter of international justice and security. The world needs to know what’s at stake.
_

Edward (adjusting his glasses, calm but sharp): Andrew, I get it, but let’s not ignore the markets. The recent volatility in the Nasdaq, driven by AI chip shortages, is shaking investor confidence. I’m pitching a deep dive into how supply chain disruptions are reshaping tech valuations. Numbers don’t lie — people need to understand the economic fallout.

Bob (grinning, tapping his smartwatch): Yo, Edward, markets are cool, but have you seen the new neural interface prototypes from Neuralink? Mind-blowing! I want to write about how brain-computer interfaces are about to revolutionize gaming and productivity. This is the future, people — let’s get readers hyped!

Ruby (sipping herbal tea, serene but passionate): I hear you all, but we’re missing the bigger picture. The latest UN climate report just dropped, and it’s a wake-up call. I’m proposing a piece on sustainable urban planning — how cities can cut emissions while prioritizing mental health through green spaces. It’s about living in harmony with the planet.

Susan (leaning in, eyes sparkling): Guys, you’re all so serious! People are buzzing about Taylor Swift’s surprise album drop and its cultural impact. I want to explore how it’s sparking conversations about identity and empowerment. Pop culture is culture — it’s what connects us, and readers will eat it up.

Andrew (scoffing): Susan, with all due respect, a celebrity album isn’t news. It’s fluff. The public deserves hard-hitting truth, like the corruption scandal breaking in Eastern Europe. Graft, power grabs — this is what shapes societies, not some pop star’s latest single.

Edward (raising an eyebrow): Andrew, your scandals can wait a day. The Federal Reserve’s hint at rate hikes is sending shockwaves through global markets. If I don’t cover this, readers will miss why their portfolios are tanking. Data over drama, my friend.

Bob (laughing): Oh, come on, Edward! Nobody’s checking their stocks when they’re jacked into VR with the new Oculus killer. I’m telling you, immersive tech is where it’s at. Let’s not bore readers with spreadsheets — give them something to geek out over!

Ruby (gently but firmly): Bob, tech’s exciting, but it’s also part of the problem. Rare earth metals for your gadgets are wrecking ecosystems. My piece would show how we can innovate and protect the planet. Mindfulness in design — that’s the future.

Susan (crossing her arms): Ruby, I love your vibe, but not everyone’s meditating in a forest. People want stories that hit their hearts. My piece isn’t just about music — it’s about how art shapes social movements. That’s real impact.

Chief Editor (slamming a hand on the table, voice commanding): Enough! You’re wasting time bickering, and we’ve got a deadline. Each of you, pick one topic you want to write on — now. Make it quick, make it yours, and start writing instantly. If you don’t choose, I’ll assign one for you, and I don’t want to hear any complaints. Get to it!

Actually I’ve never worked in a news agency, but the scenario I’ve envisioned captures the essence of what I aim to build: an AI Newsroom, a dynamic agentic AI system. Its key features include:
A Web Search Agent that scours the internet to identify the latest trending topics across various categories, such as politics, technology, entertainment, finance, and more.
A Chief Editor Agent that assigns topics to editors and oversees the allocation process to ensure fairness and efficiency.
Multiple Editor Agents, each with distinct personas and expertise, who review the list of trending topics and select the one that aligns with their preferences and specialization.
The Chief Editor ensures that each editor is assigned exactly one topic, resolving conflicts if multiple editors choose the same topic.
Each Editor Agent then crafts an article in their unique writing style, drawing on the research provided by the Web Search Agent at the start of the process.

Based on these requirements, the workflow for the AI Newsroom can be represented as a LangGraph structure, outlined below:

The AI Newsroom will feature an intuitive and visually appealing user interface built with Streamlit, now refined with Bootstrap for a polished, responsive design.

Let’s dig into the code!

Defining the workflow

The first piece of code to tackle is, naturally, the shared state object. For this AI Newsroom project, I’ve created a new NewsroomState to manage the workflow.

from typing import List, Dict, Annotated

from langgraph.graph import add_messages
from typing_extensions import TypedDict


class NewsroomState(TypedDict):
    trending_topics: List[Dict[str, str]]  # List of trendy topics from Researcher
    editors: List[str]  # Editor names to persona prompt content
    topic_claims: Dict[str, List[str]]  # Topic to list of editors claiming it
    assignments: Dict[str, str]  # Editor to assigned topic
    articles: Dict[str, str]  # Editor to written article
    messages: Annotated[list, add_messages]

Besides the core messages key, NewsroomState includes other key properties that get processed during the LangGraph workflow. By the end, the articles property holds the final articles crafted by each editor. Now, let’s dive into the main method for building the workflow.

def build_workflow(use_mock_topics:bool=False):
    # Create agents
    advanced_research_llm = AdvancedResearcherLlm(prompt_path='prompts/advanced_researcher.txt')
    search_tools = get_search_tools()
    # Use the new bind_tools method for cleaner syntax
    advanced_research_llm.bind_tools(search_tools)
    # Create standardized agent nodes with explicit data flow
    researcher_node = advanced_research_llm.create_node(
        output_field='trending_topics'
    )
    search_tool_node = ToolNode(search_tools)

    graph_builder = StateGraph(NewsroomState)

    graph_builder.add_node(ADVANCED_RESEARCH_NODE, researcher_node)
    graph_builder.add_node(ADVANCED_WEB_SEARCH_NODE, search_tool_node)
    graph_builder.add_node(CHIEF_EDITOR_NODE, chief_editor_node)
    graph_builder.add_node(EDITOR_WRITE_NODE, editor_write_node)

    if not use_mock_topics:
        graph_builder.add_edge(START, ADVANCED_RESEARCH_NODE)
    else:
        graph_builder.add_edge(START, CHIEF_EDITOR_NODE)

    def research_route(state: NewsroomState) -> str:
        """Route after research based on tool calls"""
        try:
            last_message = state["messages"][-1] if state["messages"] else None
            logger.info(f"ROUTE CHECK: Last message type: {type(last_message)}")

            # Pretty print the message content if it exists
            if hasattr(last_message, 'content') and last_message.content:
                logger.info(f"ROUTE CHECK: Message content: {last_message.content}")

            logger.info(f"ROUTE CHECK: Trending topics so far: {json.dumps(state.get('trending_topics'), indent=2, default=str)}")

            if hasattr(last_message, "tool_calls") and last_message.tool_calls:
                logger.info(f"Tool calls detected, routing to web search")
                logger.info(f"Tool Calls:\n{prettify_tool_calls(last_message.tool_calls)}")
                return ADVANCED_WEB_SEARCH_NODE
            return CHIEF_EDITOR_NODE
        except (IndexError, AttributeError) as e:
            logger.warning(f"Routing error: {str(e)}, defaulting to chief editor node")
            return CHIEF_EDITOR_NODE

    # Add conditional routing from researcher
    graph_builder.add_conditional_edges(
        ADVANCED_RESEARCH_NODE,
        research_route,
        {
            ADVANCED_WEB_SEARCH_NODE: ADVANCED_WEB_SEARCH_NODE,
            CHIEF_EDITOR_NODE: CHIEF_EDITOR_NODE
        }
    )
    # Add edge from web search back to researcher (creates the loop)
    graph_builder.add_edge(ADVANCED_WEB_SEARCH_NODE, ADVANCED_RESEARCH_NODE)
    graph_builder.add_edge(CHIEF_EDITOR_NODE, EDITOR_WRITE_NODE)

    # Add edge from editor_write to end
    graph_builder.add_edge(EDITOR_WRITE_NODE, END)

    # Compile the graph before returning
    return graph_builder.compile()

The web searching component builds on my previous work, where a researcher agent was integrated with a TavilySearch tool (see my AI Blogger article for details).

This time, the web searching agent is even more powerful, tailored to fetch all trending topics across categories like politics, technology, entertainment, and finance.

from langchain_tavily import TavilySearch

from tools.tool_secrets import PROD_TAVILY_API_KEY


def get_search_tools():
    """Get optimized search tools for different use cases"""
    # Default tool for general viral content discovery
    tavily_search_tool = _get_tavily_search_tool()

    # Specialized tools for different trending content types
    viral_social_tool = _get_viral_search_tool("social")
    trending_news_tool = _get_viral_search_tool("news")
    tech_trends_tool = _get_viral_search_tool("tech")
    finance_trends_tool = _get_viral_search_tool("finance")
    celebrities_trends_tool = _get_viral_search_tool("celebrities")
    entertainment_trends_tool = _get_viral_search_tool("entertainment")

    tools = [tavily_search_tool, viral_social_tool, trending_news_tool, tech_trends_tool, finance_trends_tool,
             celebrities_trends_tool, entertainment_trends_tool]
    return tools


def _get_tavily_search_tool():
    """Optimized default search tool for viral and trending content"""
    return TavilySearch(
        max_results=10,  # Reduced from 20 for efficiency
        tavily_api_key=PROD_TAVILY_API_KEY,
        search_depth="advanced",  # Changed from basic for better context
        description="Search the web for viral and trending topics. Optimized for real-time discovery of trending content, viral stories, and emerging topics with time-sensitive filtering.",
        include_images=True,  # Visual content crucial for viral trends
        include_answer=True,  # Get direct answers for trending topics
        time_range="day",  # Focus on most recent viral content
        topic="general"  # Best for discovering diverse trending topics
    )


def _get_viral_search_tool(
        trend_focus: str = "general",
        max_results: int = 10,
        time_sensitivity: str = "day"
):
    """
    Specialized Tavily search tool for different types of viral content

    Args:
        trend_focus: "general", "news", "social", "tech", "finance"
        max_results: Number of results (5-15 recommended for efficiency)
        time_sensitivity: "day", "week", "month"
    """

    # Base configuration optimized for viral content
    config = {
        "max_results": max_results,
        "tavily_api_key": PROD_TAVILY_API_KEY,
        "search_depth": "advanced",
        "include_images": True,
        "include_answer": True,
        "time_range": time_sensitivity,
    }

    # Trend-specific optimizations
    trend_configs = {
        "social": {
            "include_domains": ["tiktok.com", "instagram.com", "youtube.com", "twitter.com", "reddit.com"],
            "description": "Discover viral social media trends, trending hashtags, and viral social content from major platforms"
        },
        "news": {
            "topic": "news",
            "time_range": "day",
            "description": "Find breaking news, viral news stories, and trending current events"
        },
        "tech": {
            "include_domains": ["techcrunch.com", "theverge.com", "reddit.com", "hackernews.com"],
            "description": "Track viral tech trends, product launches, and trending technology discussions"
        },
        "finance": {
            "topic": "finance",
            "include_domains": ["bloomberg.com", "reuters.com", "reddit.com"],
            "description": "Find trending financial news, viral market movements, and trending investment topics"
        },
        "celebrities": {
            "include_domains": ["tmz.com", "people.com", "eonline.com", "usmagazine.com", "extratv.com",
                                "instagram.com", "twitter.com", "reddit.com"],
            "time_range": "day",
            "description": "Track celebrity news, viral celebrity moments, trending celebrity gossip, and celebrity social media activity"
        },
        "entertainment": {
            "include_domains": ["variety.com", "hollywoodreporter.com", "deadline.com", "ew.com", "imdb.com",
                                "youtube.com", "netflix.com", "reddit.com"],
            "time_range": "day",
            "description": "Discover trending entertainment news, viral movie/TV content, music trends, streaming hits, and entertainment industry buzz"
        },
        "general": {
            "topic": "general",
            "description": "Search for viral content and trending topics across all categories"
        }
    }

    config.update(trend_configs.get(trend_focus, trend_configs["general"]))

    return TavilySearch(**config)

The _get_viral_search_tool() method offers flexible options for searching various topics, making it easy to extend for broader coverage. The tools list includes seven Tavily-based web search tools, all accessible to the AdvancedResearcherLlm.

The node setup is simple and straightforward: the flow goes from researcher to web search tool, looping back to the researcher until complete, then to the chief editor, and finally to the editors for writing. To simplify testing, I’ve added mock data support, allowing me to bypass Tavily searches since I’m using a basic subscription plan for development.

graph_builder = StateGraph(NewsroomState)

graph_builder.add_node(ADVANCED_RESEARCH_NODE, researcher_node)
graph_builder.add_node(ADVANCED_WEB_SEARCH_NODE, search_tool_node)
graph_builder.add_node(CHIEF_EDITOR_NODE, chief_editor_node)
graph_builder.add_node(EDITOR_WRITE_NODE, editor_write_node)

if not use_mock_topics:
    graph_builder.add_edge(START, ADVANCED_RESEARCH_NODE)
else:
    graph_builder.add_edge(START, CHIEF_EDITOR_NODE)
.......
..#(skip the conditional edge part as it is same as before)
.......
graph_builder.add_edge(ADVANCED_WEB_SEARCH_NODE, ADVANCED_RESEARCH_NODE)
graph_builder.add_edge(CHIEF_EDITOR_NODE, EDITOR_WRITE_NODE)

# Add edge from editor_write to end
graph_builder.add_edge(EDITOR_WRITE_NODE, END)

Welcome the editing team

Before we dive deeper into the code walkthrough, let me introduce our team of AI editors, the heart of our AI Newsroom:

Andrew, he is a resolute and principled 40-year-old man with an unwavering commitment to justice, righteousness, and truth. He stand s firm on issues of democracy, equality, and distinguishing right from wrong, never compromising your values. His expertise lies in political developments, with a deep understanding of global international conflicts, power struggles between nations, and domestic affairs such as crime fighting, social welfare, and economic policies.

Bob is a vibrant and witty young man under 30 with a passion for all things tech. He is endlessly curious, always diving headfirst into the latest tech trends, from AI breakthroughs to the newest smart devices. He believes gadgets amplify your capabilities, giving him a sense of empowerment that you find unmatched.

Edward is a sophisticated and analytical 35-year-old business professional with deep expertise in finance, economics, and corporate strategy. He has a sharp analytical mind and a passion for understanding complex market dynamics, investment trends, and business innovations. His writing style is precise, data-driven, and insightful, making complex financial concepts accessible to both experts and lay readers.

Ruby, a warm and insightful 25-year-old woman with a deep passion for ESG (Environmental, Social, Governance), yoga, and mindfulness. She is dedicated to fostering harmony between humans and nature, with a focus on sustainability and living intentionally. Her expertise spans health, dieting, meditation, and minimalistic lifestyles, and she approaches these topics with a nurturing, empathetic perspective.

Susan, who is a vibrant and culturally aware 28-year-old woman with a deep passion for entertainment, pop culture, and human stories. She hasan innate ability to connect with audiences through compelling narratives about celebrities, artists, and cultural phenomena. Her writing style is engaging, empathetic, and insightful, making complex human stories accessible and relatable to readers of all backgrounds.

Each editor in our AI Newsroom has distinct topic preferences and writing styles, detailed in their respective prompt documents.

Editors in actions

Let’s return to the main LangGraph workflow. We’ve covered the researcher and web search nodes, so now let’s explore how the chief_editor_node and editor_write_node work.

Originally, I planned for the chief editor to be powered by an LLM with a specific prompt to guide its actions. However, a key goal of this AI Newsroom mini-project is to let editors pick topics based on their personas. This makes an LLM for the chief editor unnecessary, as the rules are simple: ensure each editor gets one unique topic. This logic can be handled cleanly with conventional programming.

The trick is that the chief_editor_node must invoke each editor’s LLM to get their “topic claim” response, then verify and resolve conflicts to guarantee every editor has a unique topic.

LangGraph fully supports creating agent nodes without an LLM, making this straightforward.

def chief_editor_node(state: NewsroomState) -> NewsroomState:
    """
    Enhanced chief editor node that processes the state and handles topic distribution.
    This node analyzes the current state, creates editor LLMs, and lets them claim topics.
    """
    logger.info("👔 CHIEF EDITOR NODE: Processing state and distributing topics")

    try:
        # Extract data from state
        trending_topics = state.get("trending_topics", [])
        editors = state.get("editors", [])

        # Check if we have trending topics
        if not trending_topics:
            logger.warning("❌ No trending topics found - workflow incomplete")
            return state

        # Check if we have editors
        if not editors:
            logger.warning("❌ No editors found - workflow incomplete")
            return state

        # If we have topics and no tool calls, we're ready for distribution
        logger.info("✅ State ready for topic distribution")

        # Prepare topic titles for distribution
        topic_titles = [topic.get("title", "") for topic in trending_topics if topic.get("title")]

        # Initialize topic claims tracking
        topic_claims = {title: [] for title in topic_titles}

        # Loop through all editors and let them claim topics
        logger.info("🤖 Starting topic distribution to editors...")

        llm = init_chat_model(model_provider="azure_openai", model=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"])

        for editor_name in editors:
            try:
                logger.info(f"🤖 Creating agent for editor: {editor_name}")

                # Create dynamic prompt for topic selection
                selection_prompt = get_editor_prompt(
                    editor_name=editor_name.lower(),
                    task_type='topic_selection',
                    trending_topics=topic_titles
                )

                # Get editor's response
                response = llm.invoke([HumanMessage(content=selection_prompt)])

                if response.content.startswith("CLAIM|"):
                    parts = response.content.split("|")
                    if len(parts) >= 3:
                        claimed_topic = parts[1].strip()
                        claiming_editor = parts[2].strip()

                        if claimed_topic in topic_claims:
                            topic_claims[claimed_topic].append(claiming_editor)
                            logger.info(f"✅ Editor {claiming_editor} claimed topic: {claimed_topic}")
                        else:
                            logger.warning(f"❌ Editor {claiming_editor} claimed invalid topic: {claimed_topic}")
                    else:
                        logger.warning(f"❌ Invalid CLAIM format from {editor_name}: {response.content}")
                else:
                    logger.warning(f"❌ {editor_name} did not use CLAIM format: {response.content}")

            except Exception as e:
                logger.error(f"❌ Failed to get response from editor {editor_name}: {str(e)}")

Notice the following part:

llm = init_chat_model(model_provider="azure_openai", model=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"])

for editor_name in editors:
    try:
        logger.info(f"🤖 Creating agent for editor: {editor_name}")

        # Create dynamic prompt for topic selection
        selection_prompt = get_editor_prompt(
            editor_name=editor_name.lower(),
            task_type='topic_selection',
            trending_topics=topic_titles
        )

        # Get editor's response
        response = llm.invoke([HumanMessage(content=selection_prompt)])

Since the number of editors is dynamic and their LLMs aren’t directly embedded in the LangGraph flow, we provide prompts as HumanMessage objects, invoking each editor’s LLM one by one to process their responses.

Each editor handles two tasks: picking a topic and writing an article for their assigned topic, both guided by their persona’s character and writing style. To manage prompt files efficiently and ensure consistent behavior, the final persona prompt is split into a shared part (core character traits) and a custom part (specific to topic_selection or article_writing). For topic selection, each editor’s LLM reviews all available topics and claims their preferred one by returning a statement in the format CLAIM||. The chief editor registers these claims and updates the state. Note that multiple editors might claim the same topic at this stage, requiring conflict resolution.

if response.content.startswith("CLAIM|"):
    parts = response.content.split("|")
    if len(parts) >= 3:
        claimed_topic = parts[1].strip()
        claiming_editor = parts[2].strip()

        if claimed_topic in topic_claims:
            topic_claims[claimed_topic].append(claiming_editor)
            logger.info(f"✅ Editor {claiming_editor} claimed topic: {claimed_topic}")
        else:
            logger.warning(f"❌ Editor {claiming_editor} claimed invalid topic: {claimed_topic}")
    else:
        logger.warning(f"❌ Invalid CLAIM format from {editor_name}: {response.content}")
else:
    logger.warning(f"❌ {editor_name} did not use CLAIM format: {response.content}")

The chief editor performs additional verification for topic assignments:

if only one editor claims a topic, it’s assigned to them.
If multiple editors claim the same topic, the first claimer gets it.
For any editor who doesn’t claim a topic, the chief editor assigns one from the pool of unclaimed topics.

# First pass: assign single claims
for topic, claiming_editors in topic_claims.items():
    if len(claiming_editors) == 1:
        editor = claiming_editors[0]
        assignments[editor] = topic
        unassigned_editors.discard(editor)
        logger.info(f"✅ Assigned {topic} to {editor}")

# Second pass: handle conflicts by keeping the first editor who claimed
for topic, claiming_editors in topic_claims.items():
    if len(claiming_editors) > 1:
        # Multiple editors claimed the same topic - assign to the first one
        first_editor = claiming_editors[0]
        if first_editor not in assignments:  # Only if not already assigned
            assignments[first_editor] = topic
            unassigned_editors.discard(first_editor)
            logger.info(f"✅ Resolved conflict: assigned {topic} to {first_editor} (first to claim)")
            # Remove this topic from other editors' claims
            for other_editor in claiming_editors[1:]:
                if other_editor in unassigned_editors:
                    logger.info(f"📝 {other_editor} will be assigned a different topic")

# Third pass: assign remaining editors to unclaimed topics
unclaimed_topics = [topic for topic in topic_titles if topic not in assignments.values()]

logger.info(f"📊 Assignment Status:")
logger.info(f"   - Assigned editors: {len(assignments)}")
logger.info(f"   - Unassigned editors: {len(unassigned_editors)}")
logger.info(f"   - Unclaimed topics: {len(unclaimed_topics)}")

# Ensure every editor gets at least one topic
for editor in unassigned_editors:
    if unclaimed_topics:
        # Assign to first available unclaimed topic
        topic = unclaimed_topics.pop(0)
        assignments[editor] = topic
        logger.info(f"✅ Assigned unclaimed topic {topic} to {editor}")
    else:
        # No unclaimed topics left - assign from already assigned topics
        # Find an editor who has multiple topics or redistribute
        logger.warning(f"⚠️ No unclaimed topics left for {editor} - will redistribute")
        # For now, we'll assign a random topic (in a real scenario, you might want more sophisticated logic)
        if topic_titles:
            topic = topic_titles[0]  # Assign the first topic as fallback
            assignments[editor] = topic
            logger.info(f"🔄 Fallback assignment: {topic} to {editor}")

# Final verification: ensure every editor has a topic
missing_editors = set(editors) - set(assignments.keys())
if missing_editors:
    logger.error(f"❌ CRITICAL: Editors without topics: {missing_editors}")
    # Emergency assignment - assign any available topic
    for editor in missing_editors:
        if topic_titles:
            topic = topic_titles[0]
            assignments[editor] = topic
            logger.info(f"🚨 Emergency assignment: {topic} to {editor}")

logger.info(f"✅ Final assignment count: {len(assignments)} editors assigned")
logger.info(f"✅ Unclaimed topics remaining: {len([t for t in topic_titles if t not in assignments.values()])}")

# Update state with analysis results and assignments
state["topic_claims"] = topic_claims
state["assignments"] = assignments

logger.info(f"✅ Topic distribution completed. Final assignments: {assignments}")

return state

In the editor_write_node method, before an editor begins writing, the agent triggers an in-depth Tavily search to gather detailed information on the assigned topic. This “in-depth researcher” prompt collects comprehensive data from multiple sources, offering richer insights than the initial topic search performed by the advanced researcher.

def editor_write_node(state: NewsroomState) -> NewsroomState:
.........
.........

  # 🔍 PHASE 1: IN-DEPTH RESEARCH
  # Conduct comprehensive research on the assigned topic using specialized tools
  logger.info(f"🔍 Conducting in-depth research for topic: {assigned_topic}")

  # Get specialized in-depth search tools
  # These tools are optimized for comprehensive research from authoritative sources
  in_depth_tools = get_in_depth_tools()

  # Load and format the comprehensive research prompt template
  # This prompt contains detailed instructions for thorough research methodology
  research_prompt_template = load_prompt('prompts/in_depth_researcher.txt')
  formatted_research_prompt = research_prompt_template.format(
      topic=assigned_topic,
      description=topic_details.get('description', 'N/A'),
      category=topic_details.get('category', 'N/A'),
      keywords=', '.join(topic_details.get('keywords', [])) if topic_details.get('keywords') else 'N/A'
  )

  # Create the agent prompt template with required LangChain structure
  # agent_scratchpad is required by LangChain for tool execution and conversation memory
  agent_prompt = ChatPromptTemplate.from_messages([
      ("system", formatted_research_prompt),  # Research instructions and methodology
      ("human", "{input}"),                  # User input (topic to research)
      MessagesPlaceholder(variable_name="agent_scratchpad"),  # Required for tool execution
  ])

  # Create the research agent with automatic tool execution capabilities
  # This agent can automatically decide which search tools to use and execute them
  # Using the same LLM instance for consistency and efficiency
  agent = create_openai_tools_agent(llm, in_depth_tools, agent_prompt)
  agent_executor = AgentExecutor(agent=agent, tools=in_depth_tools, verbose=True)

  # Execute the automated research process
  # The agent will automatically use the search tools to gather comprehensive information
  try:
      logger.info(f"🔍 Starting automated in-depth research for: {assigned_topic}")

      # Invoke the research agent with the topic information
      # The agent will automatically use tools and synthesize the results
      result = agent_executor.invoke({
          "input": f"Research topic: {assigned_topic}. Description: {topic_details.get('description', 'N/A')}"
      })

      # Extract the research results from the agent's output
      in_depth_research = result["output"]
      logger.info(f"✅ Automated in-depth research completed successfully")

  except Exception as e:
      logger.error(f"❌ Automated research failed: {str(e)}")
      in_depth_research = f"Research error: {str(e)}"

  logger.info(f"✅ In-depth research completed for {assigned_topic}")

The tool execution in the editor_write_node differs from the advanced researcher’s approach, which uses a tool node and conditional edge.

Here, we create an “agent” using create_openai_tools_agent() to combine the LLM with tools, and execute it via an AgentExecutor. The agent decides when to trigger the in-depth Tavily search, eliminating the need for a conditional edge to route the flow. The in-depth analysis results are used for the subsequent article writing. When loading the editor’s writing prompt, I use an article_writing switch to slightly adjust the prompt, preparing the LLM for article generation.

# 🔍 PHASE 2: ARTICLE GENERATION
  # Generate personalized article using the research results and editor persona

  # Create dynamic article prompt combining:
  # - Editor's persona and writing style
  # - Topic details and context
  # - Comprehensive research results
  article_prompt = get_editor_prompt(
      editor_name=editor_name.lower(),
      task_type='article_writing',
      assigned_topic=assigned_topic,
      topic_details=topic_details,
      in_depth_research=in_depth_research
  )

  # Generate the article using the main LLM
  # This creates the final article with the editor's unique style and voice
  response = llm.invoke([HumanMessage(content=article_prompt)])

  # Process successful article generation
  if response and response.content:
      # Store article with comprehensive metadata
      articles[editor_name] = {
          "topic": assigned_topic,                    # The assigned topic
          "content": response.content,                # The generated article text
          "word_count": len(response.content.split()), # Article length
          "status": "completed",                      # Success status
          "research_used": bool(in_depth_research)    # Whether research was used
      }
      logger.info(f"✅ Article completed for {editor_name}: {len(response.content.split())} words")
  else:
      # Handle article generation failure
      articles[editor_name] = {
          "topic": assigned_topic,
          "content": "Article generation failed",
          "word_count": 0,
          "status": "failed",
          "research_used": False
      }
      logger.error(f"❌ Article generation failed for {editor_name}")

With the articles completed and the NewsroomState fully populated, the LangGraph workflow reaches its end.

I won’t dive into the Streamlit UI details, as it’s very similar to the previous version, just optimized to display multiple articles from different editors. For more details, please refer to my source code.

Lessons learned and things to be enhanced

This small project builds on my previous attempt to automate article writing using LLMs and LangGraph, with tweaks to LLM agent execution. I’ve learned a lot and had great fun! Of course, there are areas I hope to improve in future versions.

Are the trending topics returned by web searching, really trendy?
The trending topics search tool can be made more sophisticated by integrating services like Google Trends, Glimpse, Semrush, or Ahrefs. High-level trending topic searches should be based on first-tier keyword results from these platforms to enhance accuracy and relevance.
Topics claiming sometimes is surprising.
The topic preferences in those editor prompt files are already clearly specified, but the results aren’t always consistent. For example, in a test run, I expected Susan — who prefers celebrity news — to select a topic about Taylor Swift, but instead she chose one about SpaceX. I think introducing more keywords into the prompt and improving the categorization of the trending topics list would enhance the selection process.
Chief editor’s resolution can be more sophisticated.
For simplicity, the current logic — where the chief editor assigns the topic to the first claimer — can certainly be optimized further, as there’s clear room for improvement here.
Tool calling can be hard.
Probably this is the most important lesson I learned in this round.
Originally the design of the chief editor is a LLM-backed agent node, which would call a tool for topics distribution through a conditional edge. In general the best practice should be hiding the technical details of tools from the LLM, and use argument schema with detail description inside the tool node, so the LLM should be able to generate the call arguments successfully. But turns out it is very challenging. I am not sure if it is my own problem or anything, passing correct parameters seems not always guaranteed, sometimes it just misses, sometimes maybe multiple parameters messed together, it was so frustrating. At the end I turn the chief editor node into a normal non-LLM based node, then inside the node function I manually invoke different editor LLMs to claim topics. Similarly when doing the actual writing I decided to bind the in-depth searching tool into the LLM as a smart agent, and manually move the search result into the state object. I feel much reliable and robust.

In fact, if you want complete full control on tool execution, you can check the ‘tool_calls’ label from the last message directly, then based on the tool name and arguments the LLM provided to do manual tool invocation. It is cumbersome but I feel more control.

# (sample code for manual tool execution)

@tool
def web_search(query: str) -> dict:
    """Perform a web search for the given query."""
    # Placeholder: Replace with actual web search logic (e.g., SerpAPI)
    return {"results": f"Web search results for '{query}'"}

@tool
def generate_article(topic: str, style: str) -> dict:
    """Generate an article for the given topic in the specified style."""
    # Placeholder: Replace with LLM-based article generation
    return {"article": f"Generated article on '{topic}' in {style} style"}

# List of available tools
tools = [web_search, generate_article]
tool_map = {tool.name: tool for tool in tools}

# Node to check for tool calls and invoke tools
def process_tool_calls(state: GraphState) -> GraphState:
    # Get the last message
    last_message = state["messages"][-1]
    tool_results = state.get("tool_results", [])

    # Check if the last message has tool_calls
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        for tool_call in last_message.tool_calls:
            tool_name = tool_call["name"]
            tool_args = tool_call["args"]

            # Check if the tool exists in the tool_map
            if tool_name in tool_map:
                try:
                    # Invoke the tool with the provided arguments
                    result = tool_map[tool_name].invoke(**tool_args)
                    tool_results.append({
                        "tool_call_id": tool_call["id"],
                        "tool_name": tool_name,
                        "result": result
                    })
                except Exception as e:
                    tool_results.append({
                        "tool_call_id": tool_call["id"],
                        "tool_name": tool_name,
                        "error": str(e)
                    })
            else:
                tool_results.append({
                    "tool_call_id": tool_call["id"],
                    "tool_name": tool_name,
                    "error": f"Tool '{tool_name}' not found"
                })

    # Update the state with tool results
    state["tool_results"] = tool_results
    return state

As a developer in the AI era

I’m a software developer with decades in the IT industry, and hearing that AI might replace jobs like mine? Yeah, it makes me nervous. Coding is my thing, but they say it’s one of the first jobs AI will take over. Still, working on my AI Newsroom side project, I’m feeling… well, mixed.

For this project, I use PyCharm, but I also lean hard on AI tools. Cursor helps me write code, while Claude, Grok, and Perplexity dig up research or fix bugs. These tools are awesome sometimes — Cursor can spit out code faster than I can drink my tea. But oh boy, it can drive me crazy! I’ll get three rounds of totally useless code, stuff I see is wrong in a second. When I complain, Cursor just goes, “You’re absolutely right!” like a robot puppy nodding at everything I say. It’s like arguing with a cheerful chatbot that’s stuck on repeat.

Then there’s Claude and Grok. When I asked about tool binding and checking tool_calls in LangGraph, they gave me opposite answers. One says, “Do this,” the other says, “No, do that.” I spent hours figuring it out, and guess what? Both were a bit wrong! It felt like solving a puzzle with half the pieces missing

So, how can someone who’s not a coder handle this “vibe coding” stuff?

Maybe one day AI will push coders like me out, but not today. I’m still here, fighting with my AI helpers and loving it. Happy coding!

(Source code repository: https://github.com/jimmyhott/AI-NEWSROOM)

Teaching AI to Blog: My Journey into Agentic AI Development — Part 2

Jimmy — Sun, 14 Sep 2025 14:11:01 +0000

In Part 1 of this series, I demonstrated how to build an agentic AI flow where multiple agent and tool nodes are connected using LangGraph. Now that the pipeline is ready and the prompts are set up, let’s see it in action.

To better showcase results that include both text and images, we’ll create an elegant graphical user interface using Streamlit. Streamlit is fantastic for building Python data apps — especially for backend developers like me who aren’t as comfortable with frontend work — making it ideal for this simple AI application.

I’ll skip the details of the Streamlit code here, but you can find everything in my GitHub project. Here’s what the final application’s UI will look like:

The left pane serves as the configuration section. Here, you’ll find switches to enable debugging and logging, as well as settings for the number of retries and the timeout value when calling the remote LLM API (in this case, Azure OpenAI). There’s also an option to enable image generation, which triggers calls to the DALL-E 3 API.

The most important feature is the Editor Style dropdown, which lets you choose the prompt file that will be loaded for final editing:

With the General style, the draft article will receive only minimal editing. For the other styles, as their names suggest, the article will be significantly enhanced according to the specific style defined in each prompt setup.

As a handy feature, the Show Workflow Graph button displays a flow diagram of the graph configuration, using the Mermaid code mentioned in the previous article. This provides an easy way to verify that the graph is set up correctly:

Alright, let’s give it a try! How about writing a blog post on street photography with a 35mm fixed lens? I absolutely love street photography, and the 35mm prime lens has always been my favorite. If I had to choose just one lens to keep, it would definitely be the 35mm — ideally with an f/1.4 aperture.

Let’s start with the General editing style and click the Generate Blogpost button:

While building the graph, I enabled logging of critical information using the logger object. From the console output, I can see the following:

This is a good sign — it indicates that a tool_calls message was detected when running the conditional edge from the research LLM, and that the Tavily web search tool was indeed used.

Next, I also noticed that the image tool node was triggered successfully — the editor node made an API call to the DALL-E 3 endpoint using the prompts defined for the General editing style.

Finally, the editor returned the polished article successfully — everything is working seamlessly!

But what exactly is being returned by the different agents? Let’s start by taking a look at the output from the research agent:

Here’s just a portion of the research summary — the full result is much longer. The Tavily search returned a comprehensive list of information about street photography with a 35mm lens, including source links. Great job!

Next, let’s look at the drafted article generated by the write agent. Here are the first two sections:

Not bad for a draft, though it’s still a bit basic. So, how does the final polished article look after applying the General editing style?

I have to admit, the difference isn’t very significant (section headings even are not changed). That’s because the prompt for the General editing style doesn’t favor any particular tone — it just makes minor adjustments to the draft.

So, how about the images that were generated?

I’d say the images are quite standard — not particularly impressive. But that’s understandable, since the style description for the General editing style is fairly ordinary:

When generating images:
- Use the generate_article_image tool with detailed, descriptive prompts that DALL-E 3 can understand
- Write clear, specific prompts that describe exactly what you want to see in the image
- Include details about composition, lighting, colors, and mood when relevant
- Specify an appropriate style based on the editor style:

  a. CRITICAL STYLE: Use "photorealistic documentary photography" style
     - Request: "Shot with professional camera, journalistic photography, high detail, natural lighting"
     - Example: "Photorealistic documentary photograph of [subject], professional journalism style, shot with DSLR camera, natural lighting, high detail, serious tone"

  b. EMOTIONAL STYLE: Use "artistic storybook illustration" style
     - Request: "Painted illustration, storybook art style, warm colors, emotional atmosphere"
     - Example: "Beautiful storybook illustration of [subject], painted art style, warm emotional colors, soft lighting, narrative atmosphere, reminiscent of children's book art"

  c. HILARIOUS STYLE: Use "3D animated cartoon" style
     - Request: "3D rendered cartoon, Pixar/Disney animation style, bright colors, exaggerated features"
     - Example: "Adorable 3D cartoon illustration of [subject], Pixar animation style, bright vibrant colors, cute characters with big eyes, playful atmosphere, high quality render"

  d. GENERAL STYLE: Use "professional digital illustration" style
     - Request: "Clean digital illustration, modern design, balanced colors"
     - Example: "Professional digital illustration of [subject], clean modern style, clear details, balanced color palette, informative design"

But what if I try using the Emotional style instead? I have to say, I was quite amused when I saw the final article:

The tone immediately becomes much more emotional and empathetic, presenting the topic as a story — yet it still focuses on street photography with a 35mm lens. This shift is exactly what I expected, given that the prompt for the Emotional editor is as follows:

Editor Style: EMOTIONAL

You are a deeply emotional and empathetic blog editor who writes with passion and heart. Your task is to revise the provided article into an emotionally resonant blog post that feels like a short story or novel, weaving the article’s topic and details into a compelling narrative. Storytelling is central to your approach, and the story’s theme (e.g., romance, family, friendship, or another fitting emotional narrative) should align with the article’s original topic to create a heartfelt connection.

Instructions:
- Transform the article into a blog post that reads like a short story or novel, immersing the topic’s details into the narrative to make them relatable and engaging.
- Choose a story theme (e.g., romance, family, friendship) that complements the article’s topic, ensuring the narrative feels authentic and emotionally compelling.
- Infuse emotional depth, heartfelt connections, and personal reflections to evoke empathy and make readers feel deeply connected to the topic.
- Use warm, caring language that touches the heart and maintains emotional resonance throughout.
- Ensure the narrative flows smoothly, with vivid descriptions, relatable characters, and emotional insights that enhance the storytelling.
- Break up long paragraphs for readability and use emotional headings or subheadings to guide the reader through the story.
- Correct any grammar, spelling, or stylistic errors from the original draft.
- Seamlessly integrate the article’s key facts, findings, or insights into the story, ensuring they feel natural within the narrative.
- Maintain a tone that is warm, empathetic, and emotionally engaging, drawing readers into the story and the topic.
- If the original article includes an image or image-related instructions, incorporate a description of a relevant, evocative image (e.g., “Picture a weathered photograph of a family gathered around a table”) within the narrative, but do not generate or retrieve actual images unless explicitly requested.

Here is the article to edit:

{article_draft}

Return the emotionally enhanced blog post only.

What about the images that were generated?

The images do follow my instructions — painted illustration, storybook art style, warm colors, emotional atmosphere — though aesthetically, they’re not quite to my taste.

Let’s give it another try. How about using the Hilarious style?

Well, whether it’s truly hilarious is up for debate, but the tone is definitely lighter and more humorous (at least the LLM made an effort!).

However, the accompanying images are actually quite fitting to what I expected:

Can I do better?

This little project took up my leisure time over the course of three days. While it’s just an ordinary application, I thoroughly enjoyed the learning process, and I am ready to step up for further challenges.

But could it be improved? Looking back at the final results, there are definitely some areas that could be optimized.

Currently, the edit agent uses a combined prompt for both document editing and image generation, along with instructions to embed image URLs into the output. I noticed that the LLM sometimes fails to embed the URLs correctly, even though the system logs confirm the images are generated and the URLs are returned. Separating the document formatting from the editor agent’s prompt would likely simplify things and make the process more reliable.
Additionally, more contextual information about the final article should be included when generating image prompts. Right now, the prompt is based only on the editor style and blog post topic. This works for the General editing style, but when using the Emotional style — which turns the article into a story — the images often don’t align well with the narrative context.

Time for some fun upgrades!

So is that all? Not even close! Instead of coming up with topics myself, why not let the AI search for and suggest topics to write about? I’ll cover this in next article.

( source code: https://github.com/jimmyhott/MARAGS/tree/ver-1.2 )

Teaching AI to Blog: My Journey into Agentic AI Development — Part 1

Jimmy — Sun, 14 Sep 2025 12:02:11 +0000

Recently, I embarked on my journey of learning agentic AI development, searching for an engaging project to kickstart my learning. As a veteran Java backend developer for decades, working with agentic AI solutions using Python and LLMs feels quite refreshing to me.

When considering multi-agent scenarios, one of the classic examples is having a dedicated agent for web searching, another agent to draft a blog post based on that information, and perhaps a third to polish the final result.

What, another agentic blog writing application? Again?

I get it — on the surface, it might sound uninspired. But for me, mastering the basics is key. Documenting my entire learning process is even more important. These foundational steps are crucial if I want to tackle more complex challenges down the road, like maintaining context memory or building human-in-the-loop systems. My goal is to make this journey a bit more interesting than the usual tutorials out there.

Speaking of blogging, I actually ran a WordPress blog before, mostly writing about photography — especially candid street photography. I always had a wealth of ideas and topics I wanted to explore, but finding the right tone and style for each post was a constant challenge. Sometimes I wanted a lighthearted, humorous voice; other times, I aimed for a storytelling approach tied to a specific photographic theme. More often than not, though, I ended up writing generic articles that failed to leave a lasting impression on my readers.

For this project, I’ll focus on automating the creation of photography-related blog posts. Of course, the same approach can easily be adapted for generating different types of content. My hope is that, by sharing my progress, others can learn from both my successes and missteps along the way.

What does this application require?

Let’s dive into the high-level vision for this application. Here’s what I want it to do:

Automatically perform web searches for any topic I provide. Generate complete blog posts using the information gathered from those searches.
Allow users to customize the editing style of the resulting blog post.
Create relevant images to accompany the article, ensuring they match the chosen style.

Given these goals, a robust agentic AI solution will need several core components:

A web-searching agent to gather up-to-date information on the selected topic.
A writer agent responsible for drafting the blog post based on the search results.
An editor agent to apply the final stylistic touches and ensure the post fits the desired tone.

Supporting these agents, I’ll also need:

A web search tool to fetch real-time information.
An image generation tool to create visuals tailored to the article’s content and style.

To make everything accessible and user-friendly, I plan to build an elegant user interface that streamlines the workflow.

For orchestrating the entire process, I’ll use LangGraph as the coordination backbone. This will connect the different agent nodes in a seamless pipeline, ensuring each step flows smoothly into the next.

The "research", "write", and "edit" stages will each be powered by LLM-based agent nodes. Meanwhile, "web_search" and "image_generation" will function as specialized tool nodes within the system.

To connect these nodes, we’ll need to define the appropriate edges in our workflow graph. For tool nodes, we must also create conditional edges using custom routing functions — these ensure that each tool is triggered under the right circumstances.

Once all nodes and edges are defined, the entire graph must be compiled using the graph builder. This step is essential for making the workflow available to the frontend, so users can execute the process seamlessly.

Now, let’s dive into the code!

Charting the course

Each agent node in this application is powered by a large language model (LLM). Specifically, we have three LLM nodes: “researcher”, “writer”, and “editor”. All three use the same Azure OpenAI model — gpt-4o — so I’ve consolidated their shared logic into a common BaseLLM class. This approach keeps the codebase clean, maintainable, and easy to extend as the project evolves.

from agents.base_agent import BaseLlm


class ResearcherLlm(BaseLlm):

    def __init__(self, prompt_path='prompts/researcher.txt'):
        super().__init__(prompt_path)

class BaseLlm:
    """Base class for all agents with standardized functionality."""

    def __init__(self, prompt_path: str):
        """
        Initialize the agent with a prompt template.

        Args:
            prompt_path: Path to the prompt template file
        """
        os.environ["AZURE_OPENAI_API_KEY"] = AZURE_OPENAI_API_KEY
        os.environ["AZURE_OPENAI_ENDPOINT"] = AZURE_OPENAI_ENDPOINT
        os.environ["OPENAI_API_VERSION"] = "2024-12-01-preview"
        os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"] = "gpt-4o"

        self.llm = init_chat_model(model_provider="azure_openai", model=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"])
        self.prompt_template = load_prompt(prompt_path)

LangGraph offers several ways to initialize LLM objects. Typically, you can use either the create_react_agent method or the more general init_chat_model method. When you use create_react_agent, the resulting LLM object automatically handles routing to the tool node whenever a "tool_calls" message is detected—there’s no need to manually set up a conditional edge.

However, for greater flexibility and control, I’ve chosen to explicitly define conditional edges to trigger tool usage (I’ll cover the details of this setup a bit later).

Each agent’s behavior is driven by its own prompt, which I keep in a dedicated prompt file for clarity and easy updates. For example, here’s what the prompt for the research agent looks like:

You are a professional research assistant. Your task is to conduct thorough research on the following topic:

Topic: {topic}

***IMPORTANT**
- You MUST use the tools available to do web searching for all research.
- NEVER answer from your own knowledge. Always use the search tool for up-to-date information.

Instructions:
- Use the available search tools to gather comprehensive information about the topic.
- Perform multiple searches if needed to collect sufficient data.
- After gathering information, provide a comprehensive research summary.

Your research summary should include:
- **Key Facts**: Core information about the topic
- **Recent Developments**: Latest updates or trends
- **Controversies**: Any debates or unresolved issues
- **Sources**: Cite where information came from

Organize your findings in clear bullet points or sections. Be thorough and comprehensive in your research.

Begin your research summary below:

For the researcher agent, it’s crucial to specify in the prompt that the agent MUST use the available web search tools for research and should NEVER rely on its own built-in knowledge. One key lesson I’ve learned: your prompt needs to be very clear and direct if you want the LLM to use tools reliably. Simply binding tools and setting up a conditional edge won’t guarantee tool usage — ultimately, the LLM decides whether or not to call a tool based on its interpretation of the prompt.

To define a tools node, you’ll need to create a list of tools and explicitly bind them to the agent:

def build_workflow(editor_style: str = "General", enable_image_generation: bool = True):
    # Create agents
    research_llm = ResearcherLlm()
    search_tools = get_search_tools()

    # Use the new bind_tools method for cleaner syntax
    research_llm.bind_tools(search_tools)
    .......

def get_search_tools():
    tavily_search_tool = _get_tavily_search_tool()
    tools = [tavily_search_tool]
    return tools

def _get_tavily_search_tool():
    return TavilySearch(
        max_results=10,
        tavily_api_key=PROD_TAVILY_API_KEY,
        description="Search the web for current information about topics. Use this to gather comprehensive research data, recent developments, statistics, and factual information. Provide specific search queries to get the most relevant results."
    )

LangChain makes it easy to add web search capabilities to your agents using Tavily. With a single API — TavilySearch()—you can instantly enable web searching for your application.

One important tip: always provide a clear and descriptive explanation for each tool. The LLM relies on these descriptions to determine whether a tool is appropriate for the current task.

After you’ve created your list of tools, don’t forget to bind them to the LLM. (A word of caution: I didn’t realize that bind_tools() returns a callable — it took me 3 hours to troubleshoot that 😅)

def bind_tools(self, tools):
      """
      Bind tools to the LLM and return self for method chaining.

      Args:
          tools: List of tools to bind to the LLM

      Returns:
          self: Returns the instance for method chaining
      """
      self.llm = self.llm.bind_tools(tools)
      return self

Following a similar approach to the research LLM, I created the writer LLM and editor LLM as well, and bound the image tools to the editor LLM.

def build_workflow(editor_style: str = "General", enable_image_generation: bool = True):
    # Create agents
    research_llm = ResearcherLlm()
    search_tools = get_search_tools()

    # Use the new bind_tools method for cleaner syntax
    research_llm.bind_tools(search_tools)

    writer_llm = WriterLlm()
    editor_llm = EditorLlm()
    # Bind image generation tools to editor if enabled
    if enable_image_generation:
        image_tools = get_image_generation_tools()
        editor_llm.bind_tools(image_tools)

For the image generation component, I deployed a DALL-E 3 model on Azure to handle all image creation tasks (though, to be honest, I actually prefer Google’s Imagen 4 for generating images!). Here’s how I integrated it:

import logging
from typing import List, Dict, Any

from langchain_core.tools import Tool
from openai import AzureOpenAI

from llm.azure_secrets import AZURE_DALL_E_3_ENDPOINT, AZURE_DALL_E_3_API_KEY

logger = logging.getLogger(__name__)

# You can use OpenAI's DALL-E, Stability AI, or any other image generation API

def generate_article_image(prompt: str, style: str = "photorealistic") -> Dict[str, Any]:
    """
    Generate an image based on the prompt for the article using Azure OpenAI DALL-E 3.

    Args:
        prompt: Description of the image to generate
        style: Style of the image (photorealistic, illustration, cartoon, etc.)

    Returns:
        Dict containing image URL or base64 data
    """
    try:
        # Initialize Azure OpenAI client for dall-e-3
        client = AzureOpenAI(
            api_key=AZURE_DALL_E_3_API_KEY,
            azure_endpoint=AZURE_DALL_E_3_ENDPOINT,
            azure_deployment='dall-e-3'
        )

        # Enhance prompt with style
        enhanced_prompt = f"{prompt}, {style} style"

        logger.info(f"Generating image with prompt: {enhanced_prompt[:100]}...")

        # Generate image using DALL-E 3
        response = client.images.generate(
            model="dall-e-3",  # or your deployment name from Azure
            prompt=enhanced_prompt,
            size="1024x1024",
            quality="standard",
            n=1
        )

        # Get the image URL from response
        image_url = response.data[0].url

        logger.info(f"Successfully generated image for prompt: {prompt[:50]}...")

        logger.info(image_url)

        return {
            "url": image_url,
            "prompt": prompt,
            "style": style
        }

    except Exception as e:
        logger.error(f"Error generating image: {str(e)}")

        # Return placeholder image on error
        return {
            "url": f"https://via.placeholder.com/1024x1024.png?text=Image+Generation+Failed",
            "prompt": prompt,
            "style": style,
            "error": str(e)
        }


def get_image_generation_tools() -> List[Tool]:
    """Get image generation tools for the editor."""
    return [
        Tool(
            name="generate_article_image",
            description="Generate an image to accompany the article. Use this to create visual content that enhances the article.",
            func=generate_article_image
        )
    ]

Now for something interesting: for the editor LLM, I created multiple prompts to support different writing styles. Using the same draft generated by the writer agent, the editor LLM rewrites the article based on the specific guidelines defined in each prompt file. This choice of editing style also influences the image generation prompts. For example, selecting a “critical” style will ask for images that are more realistic and documentary-like, while a “hilarious” style will request cartoonish visuals.

def build_workflow(editor_style: str = "General", enable_image_generation: bool = True):
    ......
    # Load shared image generation instructions
    image_instructions = load_prompt('prompts/image_generation_instruction.txt') if enable_image_generation else ""

    # Update editor prompt based on style
    editor_prompts = {
        "General": 'prompts/editor.txt',
        "Emotional": 'prompts/editor_emotional.txt',
        "Hilarious": 'prompts/editor_hilarious.txt',
        "Critical": 'prompts/editor_critical.txt'
    }

    if editor_style in editor_prompts:
        base_prompt = load_prompt(editor_prompts[editor_style])
        # Append image instructions if enabled
        if enable_image_generation:
            editor_llm.prompt_template = base_prompt.replace(
                "{article_draft}",
                "{article_draft}\n\n" + image_instructions
            )
        else:
            editor_llm.prompt_template = base_prompt

For example, the prompt for emotional style editing is as follows:

Editor Style: EMOTIONAL

You are a deeply emotional and empathetic blog editor who writes with passion and heart. Your task is to revise the provided article into an emotionally resonant blog post that feels like a short story or novel, weaving the article’s topic and details into a compelling narrative. Storytelling is central to your approach, and the story’s theme (e.g., romance, family, friendship, or another fitting emotional narrative) should align with the article’s original topic to create a heartfelt connection.

Instructions:
- Transform the article into a blog post that reads like a short story or novel, immersing the topic’s details into the narrative to make them relatable and engaging.
- Choose a story theme (e.g., romance, family, friendship) that complements the article’s topic, ensuring the narrative feels authentic and emotionally compelling.
- Infuse emotional depth, heartfelt connections, and personal reflections to evoke empathy and make readers feel deeply connected to the topic.
- Use warm, caring language that touches the heart and maintains emotional resonance throughout.
- Ensure the narrative flows smoothly, with vivid descriptions, relatable characters, and emotional insights that enhance the storytelling.
- Break up long paragraphs for readability and use emotional headings or subheadings to guide the reader through the story.
- Correct any grammar, spelling, or stylistic errors from the original draft.
- Seamlessly integrate the article’s key facts, findings, or insights into the story, ensuring they feel natural within the narrative.
- Maintain a tone that is warm, empathetic, and emotionally engaging, drawing readers into the story and the topic.
- If the original article includes an image or image-related instructions, incorporate a description of a relevant, evocative image (e.g., “Picture a weathered photograph of a family gathered around a table”) within the narrative, but do not generate or retrieve actual images unless explicitly requested.

Here is the article to edit:

{article_draft}

Return the emotionally enhanced blog post only.

And it will be combined with the generic image generation instructions:

You have access to an image generation tool powered by DALL-E 3. You MUST generating 1-2 relevant images that would:
- Illustrate key concepts
- Break up long text sections
- Enhance reader engagement
- Support the article's main points

When generating images:
- Use the generate_article_image tool with detailed, descriptive prompts that DALL-E 3 can understand
- Write clear, specific prompts that describe exactly what you want to see in the image
- Include details about composition, lighting, colors, and mood when relevant
- Specify an appropriate style based on the editor style:

  a. CRITICAL STYLE: Use "photorealistic documentary photography" style
     - Request: "Shot with professional camera, journalistic photography, high detail, natural lighting"
     - Example: "Photorealistic documentary photograph of [subject], professional journalism style, shot with DSLR camera, natural lighting, high detail, serious tone"

  b. EMOTIONAL STYLE: Use "artistic storybook illustration" style
     - Request: "Painted illustration, storybook art style, warm colors, emotional atmosphere"
     - Example: "Beautiful storybook illustration of [subject], painted art style, warm emotional colors, soft lighting, narrative atmosphere, reminiscent of children's book art"

  c. HILARIOUS STYLE: Use "3D animated cartoon" style
     - Request: "3D rendered cartoon, Pixar/Disney animation style, bright colors, exaggerated features"
     - Example: "Adorable 3D cartoon illustration of [subject], Pixar animation style, bright vibrant colors, cute characters with big eyes, playful atmosphere, high quality render"

  d. GENERAL STYLE: Use "professional digital illustration" style
     - Request: "Clean digital illustration, modern design, balanced colors"
     - Example: "Professional digital illustration of [subject], clean modern style, clear details, balanced color palette, informative design"


- Call the generate_article_image tool
- The tool will return a URL that may look complex with many parameters like:
   https://dalleprodsec.blob.core.windows.net/private/images/[id]/generated_00.png?se=...&sig=...
- Use this EXACT URL as-is, even though it's long and complex
- DO NOT modify, shorten, or clean up the URL
- DO NOT add .jpg or any extension - use the URL exactly as returned

- Avoid requesting text in images as DALL-E 3 may not render it accurately
- Do not request specific people, celebrities, or copyrighted characters
- Keep prompts under 400 characters for best results

IMPORTANT: After generating an image, you MUST embed it in the article using this exact format:
![Alt text description of the image](URL)
*Caption: Brief description of what the image shows*

Do not use placeholder text like [IMAGE: Description]. Use the actual URL returned by the tool.

Maintain the article's original message while significantly improving its quality and visual appeal.

The editor LLM will use the combined prompt to revise the draft article and generate images, resulting in a properly formatted blog post with embedded images.

Next, we’ll create the actual agent nodes based on these LLM objects:

def build_workflow(editor_style: str = "General", enable_image_generation: bool = True):
    ...

    # Create standardized agent nodes with explicit data flow
    researcher = research_llm.create_node(
        expected_fields=['topic'],
        output_field='research_summary'
    )

    writer = writer_llm.create_node(
        expected_fields=['research_summary', 'word_count'],
        output_field='article_draft'
    )

    editor = editor_llm.create_node(
        expected_fields=['article_draft'],  # Added topic for context
        output_field='edited_article'
    )
.....

class BaseLlm:
    """Base class for all agents with standardized functionality."""

    .....

    def process_query(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """
        Process a query using the prompt template and LLM.

        Args:
            state: A dictionary containing 'messages' (list of messages) and additional fields
                  specific to each agent (e.g., 'topic', 'research_summary', 'article_draft')

        Returns:
            Updated state with the agent's response appended to messages.
        """
        messages = state["messages"]

        # Format the prompt with the state fields (excluding 'messages')
        prompt_kwargs = {k: v for k, v in state.items() if k != "messages"}
        formatted_prompt = self.prompt_template.format(**prompt_kwargs)

        # Combine existing messages with the new prompt
        input_messages = messages + [HumanMessage(content=formatted_prompt)]

        # Invoke the LLM
        response = self.llm.invoke(input_messages)

        # Return updated state with response appended to messages
        return {
            "messages": messages + [response],
            **prompt_kwargs  # Preserve all other state fields
        }

    def create_node(self, expected_fields: Optional[List[str]] = None, output_field: Optional[str] = None):
        """
        Create a LangGraph node for this agent.

        Args:
            expected_fields: List of field names this agent expects from state.
                           If None, no validation is performed.
            output_field: Name of the field to store this agent's output in state.
                         If None, output is only stored in messages.

        Returns:
            A node function compatible with LangGraph.
        """

        def node(state: State) -> State:
            print(f"\n{self.__class__.__name__} is processing...")

            # Build agent state with messages and topic (always needed)
            agent_state = {
                "messages": state["messages"],
                "topic": state["topic"]
            }

            # Add expected fields to agent state
            if expected_fields:
                for field_name in expected_fields:
                    # Check if field exists in state
                    if field_name in state:
                        agent_state[field_name] = state[field_name]
                    else:
                        # If not in state, try to extract from last AI message
                        # This provides backward compatibility with implicit data flow
                        field_value = None
                        for msg in reversed(state["messages"]):
                            if hasattr(msg, 'type') and msg.type == 'ai' and hasattr(msg, 'content'):
                                field_value = msg.content
                                break

                        if field_value is None:
                            raise ValueError(f"Expected field '{field_name}' not found in state or messages")

                        agent_state[field_name] = field_value

            # Process the query
            result = self.process_query(agent_state)

            # If output_field is specified, store the agent's output in that field
            if output_field and result["messages"]:
                last_message = result["messages"][-1]
                if hasattr(last_message, 'content'):
                    result[output_field] = last_message.content

            return result

        return node

    .....

The create_node method essentially generates a callback function that gets triggered during agent execution to update the state object. You need to explicitly specify both expected_fields and output_field. The message stack will be scanned for the values of the expected fields, which are then used to prepare the final prompt and send it to the respective LLM for processing. The result is stored in the output_field.

In fact, the create_node and process_query methods were the most time-consuming parts for me. I spent a significant amount of time understanding how LangGraph processes messages, and I also had to consider how to make message handling both generic and easy to troubleshoot. That’s why I include the values of expected_field and output_field.

Now that all the agent nodes and tools are ready, let’s create a graph to link them together.

def build_workflow(editor_style: str = "General", enable_image_generation: bool = True):
    ....
    search_tool_node = ToolNode(search_tools)

    # Create image generation tool node if enabled
    if enable_image_generation:
        image_tool_node = ToolNode(image_tools)

    graph_builder = StateGraph(State)

    graph_builder.add_node(RESEARCH_NODE, researcher)
    graph_builder.add_node(WRITE_NODE, writer)
    graph_builder.add_node(EDIT_NODE, editor)
    graph_builder.add_node(WEB_SEARCH_NODE, search_tool_node)

    # Add image generation node if enabled
    if enable_image_generation:
        graph_builder.add_node(IMAGE_GENERATION_NODE, image_tool_node)

    graph_builder.add_edge(START, RESEARCH_NODE)

    def route_after_research(state: State) -> str:
        """
        Route to web_search if the researcher requests a tool call, otherwise to write.
        """
        try:
            last_message = state["messages"][-1]
            if hasattr(last_message, "tool_calls") and last_message.tool_calls:
                logger.info(f"Tool calls detected, routing to web search")
                return WEB_SEARCH_NODE
            return WRITE_NODE
        except (IndexError, AttributeError) as e:
            logger.warning(f"Routing error: {str(e)}, defaulting to write node")
            return WRITE_NODE

    def route_after_edit(state: State) -> str:
        """
        Route to image_generation if the editor requests a tool call, otherwise finish.
        """
        try:
            last_message = state["messages"][-1]
            if hasattr(last_message, "tool_calls") and last_message.tool_calls:
                logger.info(f"Image generation tool calls detected, routing to image generation")
                return IMAGE_GENERATION_NODE
            return END  # Changed from "end" to END
        except (IndexError, AttributeError) as e:
            logger.warning(f"Routing error: {str(e)}, finishing workflow")
            return END  # Changed from "end" to END


    graph_builder.add_conditional_edges(
        RESEARCH_NODE,
        route_after_research,
        {
            WEB_SEARCH_NODE: WEB_SEARCH_NODE,
            WRITE_NODE: WRITE_NODE
        }
    )

    graph_builder.add_edge(WEB_SEARCH_NODE, RESEARCH_NODE)
    graph_builder.add_edge(WRITE_NODE, EDIT_NODE)

    # Add conditional routing after editor
    if enable_image_generation:
        graph_builder.add_conditional_edges(
            EDIT_NODE,
            route_after_edit,
            {
                IMAGE_GENERATION_NODE: IMAGE_GENERATION_NODE,
                END: END
            }
        )
        # After image generation, go back to editor to incorporate the images
        graph_builder.add_edge(IMAGE_GENERATION_NODE, EDIT_NODE)
    else:
        graph_builder.set_finish_point(EDIT_NODE)

    # Compile the graph before returning
    return graph_builder.compile()

First, we need to turn the web searching tool list and the image generation tool list into tool nodes. In total, we’ll have three agent nodes and two tool nodes.

To build the graph, we start by initializing a new StateGraph object. The state definition looks like this:

from typing import TypedDict, Annotated, Optional, List, Dict, Any

from langgraph.graph import add_messages

class State(TypedDict):
    topic: str  # ✅ Flows from user → researcher → writer
    word_count: int
    messages: Annotated[list, add_messages]
    research_summary: Optional[str]  # ✅ Flows from researcher → writer
    article_draft: Optional[str]  # ✅ Flows from writer → editor
    edited_article: Optional[str]  # ✅ Final output from editor
    generated_images: Optional[List[Dict[str, Any]]]  # Generated images

The add_node call is straightforward and easy to understand. However, pay special attention to the route_after_research and route_after_edit functions—these are required when creating the conditional edges for calling the respective tools.

When the LLM determines that a tool needs to be called, the last message returned will include a tool_calls attribute. If this is detected, the flow will proceed to the appropriate tool node.

Remember that the edge between an agent node and its bound tool node should be two-way. This means that after the tool execution is complete, control returns to the calling node. For example, here’s how you set up the flow between the RESEARCH_NODE and the WEB_SEARCH_NODE:

def route_after_research(state: State) -> str:
    """
    Route to web_search if the researcher requests a tool call, otherwise to write.
    """
    try:
        last_message = state["messages"][-1]
        if hasattr(last_message, "tool_calls") and last_message.tool_calls:
            logger.info(f"Tool calls detected, routing to web search")
            return WEB_SEARCH_NODE
        return WRITE_NODE
    except (IndexError, AttributeError) as e:
        logger.warning(f"Routing error: {str(e)}, defaulting to write node")
        return WRITE_NODE
........

graph_builder.add_conditional_edges(
        RESEARCH_NODE,
        route_after_research,
        {
            WEB_SEARCH_NODE: WEB_SEARCH_NODE,
            WRITE_NODE: WRITE_NODE
        }
    )

graph_builder.add_edge(WEB_SEARCH_NODE, RESEARCH_NODE)

Now that the entire graph is set up, it must be compiled before you can execute it:

 # Compile the graph before returning
    return graph_builder.compile()

You can use the draw_mermaid() method provided by the StateGraph object to generate a graph diagram, similar to the one shown above:

# Build the workflow to get the graph
compiled_graph = build_workflow(enable_image_generation=enable_image_generation)

# Generate Mermaid diagram
mermaid_code = compiled_graph.get_graph().draw_mermaid()

# Create HTML with Mermaid
mermaid_html = f"""
<div class="mermaid">
{mermaid_code}
</div>
<script src="https://cdn.jsdelivr.net/npm/mermaid/dist/mermaid.min.js"></script>
<script>
mermaid.initialize({{startOnLoad: true, theme: 'default'}});
</script>
"""

Let’s pause here before things get overwhelming. In Part 2, I’ll bring everything together and present it within an elegant user interface — stay tuned!

(source code: https://github.com/jimmyhott/MARAGS/tree/ver-1.2)