In my previous post, I showed the output of a multi-agent application I wrote to create blog posts (not to worry, it is for demonstration purposes only). In this post, I will begin the process of working through the code, line by line.
This application is written in Python, in a Colab notebook, using (among other things) LangChain and LangGraph. To follow along you will need to obtain an API key from OpenAI and a key from Tavily.
If you are a C# programmer with little or no Python experience, don’t panic! Python is pretty readable, and I’ll explain any part that is potentially obscure or confusing.
This will be a multi-agent application. The agents we’ll create will be:
- Blogger which will orchestrate the others
- Researcher, which will search the web for relevant information
- Author, which will write drafts of the blog post
- Reviewer, which will evaluate the drafts and suggest improvements
As a general rule, I try to limit the number of agents to 3-5. Any more than that can get terribly complicated with diminishing returns. Your mileage may vary.
Set Up
The program begins by importing the necessary libraries.
!pip install -q openai==1.66.3 \
langchain==0.3.20 \
langchain-openai==0.3.9 \
langchain_experimental==0.3.4 \
langchain-tavily==0.2.4 \
tavily-python==0.5.0 \
langgraph==0.3.21 \
langgraph-supervisor==0.0.18
You then load your OpenAI API key and Tavily API key from either a configuration file (which is what I do here) or from the environment.
import json
import os
from pprint import pprint
file_name = 'config.json'
with open(file_name, 'r') as file:
config = json.load(file)
os.environ['OPENAI_API_KEY'] = config.get("API_KEY")
os.environ["OPENAI_BASE_URL"] = config.get("OPENAI_API_BASE")
os.environ["TAVILY_API_KEY"] = config.get("TAVILY_API_KEY")Next, we set up the model.
from langchain_openai import ChatOpenAI
model_name = 'gpt-4o-mini'
llm = ChatOpenAI(
model = model_name,
temperature=0,
max_tokens=4096
)I’ve opted for the gpt-4o-mini model, as it’s the most cost-effective option. I’ve set the temperature to 0 to get the most consistent results.
Tavily is a tool for searching the web, so let’s set that up as well
from langchain_tavily import TavilySearch
tavily_tool = TavilySearch(
max_results=5,
topic="general",
include_answer=False,
include_raw_content=False,
search_depth="basic"
)You can find all the options for this on the Tavily website.
I’m going to want a state object so that I can pass it among the agents.
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
import operator
class ResearchState(TypedDict):
"""State for the research workflow."""
main_task: str
research_findings: Annotated[List[str], operator.add]
draft: str
review_notes: str
revision_number: int
next_step: str
current_sub_task: strNote, the line research_findings: Annotated [List(str), operator.add) is a Python type annotation. In short, an annotated list has metadata, in this case the function operator.add. This operator, in this context, is used to update state, though it can also be used to reduce the prompt.
Blogger
To get started, we’ll create the (non-trivial) blogger agent. This is actually the most powerful and thus the most complex of the agents.
We begin with the prompt template (this is, essentially, the system prompt)
blogger_prompt_template = """You are a blogger managing a blog post creation workflow.
Current Task: {main_task}
Current State:
- Research Findings: {research_findings}
- Blog Draft: {draft}
- Reviewer Feedback: {review_notes}
- Revision Number: {revision_number}
Your goal is to ensure a clear, engaging, and valuable blog post targeted at software developers.
Decide the next step and respond only with a JSON object (no extra text):
{
"next_step": "researcher" or "author" or "END",
"task_description": "Brief description of what needs to be done next"
}
Decision Rules:
- If no research exists, choose "researcher"
- If research exists but no draft, choose "author"
- If draft exists and reviewer says "APPROVED", choose "END"
- If draft needs revision, choose "author"
- If revision_number >= 4, choose "END"
"""Take a moment to read this over; it sets the parameters and goals of the program. The decision rules are critical, they control the flow, and they set a limit on revisions (in this case 4).
Having told the Blogger what we are trying to accomplish and the general tone of the output we’re ready to create the decision tree that constitutes the workflow for the Blogger. This is pretty long, but much of it is self-explanatory.
def create_blogger_chain():
"""Creates the bloger decision chain."""
def blogger_invoke(state):
research = state.get("research_findings", [])
research_text = "\n".join(research) if research else "No research yet."
revision = state.get("revision_number", 0)
has_research = len(research) > 0
has_draft = bool(state.get("draft", "").strip())
review = state.get("review_notes", "")
if "APPROVED" in review.upper() and has_draft:
print("Blogger: Draft approved, ending workflow")
return {
"next_step": "END",
"task_description": "Report approved and complete"
}
if not has_research:
print("Blogger: No research yet, directing to researcher")
return {
"next_step": "researcher",
"task_description": f"Research the topic: {state.get('main_task', '')}"
}
if has_research and not has_draft:
print("Blogger: Have research, creating first draft")
return {
"next_step": "author",
"task_description": "Write the first draft based on research findings"
}
if has_draft and not review:
print("Blogger: Have draft, sending to reviewer")
return {
"next_step": "reviewer",
"task_description": "Prepare draft for review"
}
if review and "APPROVED" not in review.upper() and revision <= 4:
print(f"Blogger: Revision {revision}, sending back to author")
return {
"next_step": "author",
"task_description": "Revise the draft based on review feedback"
}
if revision >= 4:
print("Blogger: Max revisions reached! Ending")
return {
"next_step": "END",
"task_description": "Maximum revisions reached! Finalizing report"
}
# LLM as fallback
prompt = blogger_prompt_template.format(
main_task=state.get("main_task", ""),
research_findings=research_text,
draft=state.get("draft", "No draft yet."),
review_notes=review if review else "No review yet.",
revision_number=revision
)
try:
response = llm.invoke(prompt)
content = response.content if hasattr(response, 'content') else str(response)
text = content.strip()
if text.startswith("```"):
lines = text.split("\n")
text = "\n".join([l for l in lines if not l.strip().startswith("```")])
text = text.strip()
decision = json.loads(text)
if "next_step" in decision:
return decision
except Exception as e:
print(f"LLM parsing error: {e}")
# Final fallback
print("Blogger: Using final fallback - continuing with author")
return {
"next_step": "author",
"task_description": "Continue with draft creation"
}
return blogger_invoke
# Creating a callable object
blogger_chain = create_blogger_chain()We begin with a nested function.
def create_blogger_chain():
"""Creates the blogger decision chain."""
def blogger_invoke(state):
...When called, Python executes the code inside it, that is, the inner function. In this case, the inner function does the real work.
We define a second function inside create_blogger_chain. We do this to create a closure, that is the inner function can access variables from the outer function. This is a common construct when working with langchain. The inner function has access to the llm without it being passed every time. So, in short, create_blogger_chain is actually a factory function that constructs and configures a callable function and then returns it.
We next set up our variables based on starting values in the state object. With that done, we’re ready to progress through a series of possible conditions. These follow the rules established in the template.
All that’s left for the blogger is to create a node where the decision will be implemented. We’ll use this node, and the others we’ll create, when we implement the workflow (after we define all the agents).
def blogger_node(state: ResearchState) -> dict:
"""Blogger decides the next step."""
print("\n>>>Blogger")
decision = blogger_chain(state)
next_step = decision.get("next_step", "researcher")
task_desc = decision.get("task_description", "Continue work")
print(f"Decision: {next_step}")
print(f"Task: {task_desc}")
return {
"next_step": next_step,
"current_sub_task": task_desc,
}Note The variable decision is assigned as a result of calling the code we just reviewed. With that in hand, we call get on the decision, asking for the next step. If none is returned, we use researcher.
That’s it for blogger. We’ll review the other agents in the next posting.
