Day 19/30: 3-Agent Content Pipeline

I still remember the day our content generation pipeline fell apart. We had built a system that relied on a single agent to research, write, and edit content. It was a mess - the agent would often get stuck in an infinite loop, trying to perfect a piece of content that was already good enough. Our support team was flooded with complaints about delayed deliveries, and our team was struggling to keep up with the demand. That's when we decided to break down the pipeline into three separate agents: researcher, writer, and editor. Each agent would have a specific role, and they would work in sequence to generate high-quality content.

The researcher agent would be responsible for gathering information on a given topic. It would use LangGraph's StateGraph to build a knowledge graph, which would then be passed to the writer agent. The writer agent would use this graph to generate a draft, which would be reviewed by the editor agent. The editor agent would then refine the draft, checking for grammar, coherence, and overall quality.

Here's an example of how we implemented this pipeline using LangGraph and MCP:

import langgraph
from langgraph import StateGraph
from mcp import ModelContextProtocol

# Define the researcher agent
def researcher_agent(topic):
    # Create a new StateGraph
    graph = StateGraph()

    # Add nodes for the topic and related concepts
    graph.add_node(topic)
    graph.add_node("concept1")
    graph.add_node("concept2")

    # Add conditional edges between the nodes
    graph.add_conditional_edges([
        (topic, "concept1", 0.8),
        (topic, "concept2", 0.4),
        ("concept1", "concept2", 0.6)
    ])

    # Return the StateGraph
    return graph

# Define the writer agent
def writer_agent(graph):
    # Use the StateGraph to generate a draft
    draft = ""
    for node in graph.nodes:
        draft += f"{node} is related to {graph.get_neighbors(node)}\n"

    # Return the draft
    return draft

# Define the editor agent
def editor_agent(draft):
    # Refine the draft
    refined_draft = ""
    for sentence in draft.split("\n"):
        # Check for grammar and coherence
        if sentence:
            refined_draft += sentence + "\n"

    # Return the refined draft
    return refined_draft

# Create an instance of the ModelContextProtocol
mcp = ModelContextProtocol()

# Define the pipeline
def content_pipeline(topic):
    # Researcher agent
    graph = researcher_agent(topic)

    # Writer agent
    draft = writer_agent(graph)

    # Editor agent
    refined_draft = editor_agent(draft)

    # Return the refined draft
    return refined_draft

# Test the pipeline
topic = "agentic AI"
print(content_pipeline(topic))

One practical gotcha we encountered while building this pipeline was ensuring that the agents were properly synchronized. We had to implement a checkpointing system to ensure that the output of one agent was correctly passed to the next agent. This involved using LangGraph's checkpointers module to save the state of each agent after it completed its task.

As we continue to develop our agentic AI systems, we're excited to explore new ways to improve the efficiency and effectiveness of our pipelines. Tomorrow, we'll dive into another critical aspect of building complex AI systems, and we'll see how LangGraph and MCP can help us overcome the challenges that come with it.