Build a Social Media Content Agent

Build a LangGraph agent that automates Instagram content with AI review.

Introduction

30 Second Summary

Creating consistent social media content is exhausting. Between researching trends, writing captions, choosing hashtags, and picking the right posting time, a single Instagram post can eat up an hour of your day.

In this project, you will build an AI agent that automates Instagram content creation using a multi-step pipeline with built-in quality controls and a dual-mode review system.

What You'll Build

A multi-node AI content agent powered by LangGraph and Google Gemini (via Vertex AI) that researches trends, generates structured Instagram posts, and routes content through either human review or an AI evaluation agent before publishing.

By the end of this project, you'll have:

  • A working LangGraph StateGraph agent that researches trending topics and generates Instagram content validated by Pydantic models.
  • A dual-mode review system with human-in-the-loop approval for development and an AI evaluation agent for production.
  • A CLI interface with a --mode flag to switch between dev and prod execution paths.
  • Secret Mission: An enhanced evaluation agent with multi-dimensional scoring, configurable confidence thresholds, and three distinct routing paths.

Are there any prerequisites?

You should have Python experience and a Google Cloud Platform account with Vertex AI enabled. You also need the Google Cloud CLI (gcloud) installed on your machine.

Before We Start

Before building the content agent, you need a clean workspace with the right packages and credentials in place.

In this step, you'll create a Python project, install LangGraph and its dependencies, authenticate with Google Cloud, and verify that Vertex AI responds to your requests.

In this step, get ready to:
  • Create a project directory with a Python virtual environment.
  • Install LangGraph, LangChain, and Pydantic packages.
  • Configure environment variables and verify Vertex AI connectivity.
Create your project directory and virtual environment

A virtual environment isolates your project's packages from the rest of your system. This prevents version conflicts between projects.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir social-media-agent
cd social-media-agent
  • Confirm the folder exists by running:
ls

You should see the social-media-agent folder listed in the output.

  • Create a Python virtual environment by running:
python3 -m venv venv
  • Activate the virtual environment:

macOS

  • Activate by running:
source venv/bin/activate

Windows

  • Activate by running:
venv\Scripts\Activate.ps1

You should see (venv) appear at the start of your terminal prompt.

Don't see (venv) in your prompt?

On macOS, make sure you typed source before the path. Without it, the script won't modify your current shell session.

On Windows PowerShell, you may need to allow script execution first. Run Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser then try activating again.

Why isn't my virtual environment activating?

Now you need to authenticate with Google Cloud so your code can access Vertex AI.

  • Authenticate with Application Default Credentials by running:
gcloud auth application-default login

What does this command do?

This command opens your browser and asks you to sign in with your Google account. Once you sign in, it saves credentials locally so your Python code can call Vertex AI without needing an API key in your code.

Your browser opens a Google sign-in page. Sign in with the account linked to your GCP project. When it succeeds, your terminal displays a message starting with Credentials saved to file: followed by a file path.

  • Set your Google Cloud project by running:
gcloud config set project [[GCP_PROJECT="your-project-id-here"]]

Find your project ID

Your GCP project ID appears in the Google Cloud Console at the top of the page, next to the project name dropdown. It looks like my-project-123456 and is different from the project name.

Getting a permission or authentication error?

Make sure you signed in with the Google account that owns the GCP project. If you have multiple accounts, run gcloud auth application-default login again and pick the correct one.

Help me fix my gcloud authentication error

Install dependencies
  • Install the required packages by running:
pip install langgraph==1.2.6 langchain-google-genai==4.2.6 langchain==1.3.11 pydantic==2.11.4 python-dotenv==1.1.0

What are these packages?

  • langgraph is the graph-based agent orchestration framework you'll build your agent with.
  • langchain-google-genai connects LangChain to Google's Gemini models via Vertex AI.
  • langchain provides the core abstractions for tools, prompts, and model interaction.
  • pydantic enforces typed, validated data structures on your agent's outputs.
  • python-dotenv loads environment variables from your .env file.

The install takes about a minute. When it finishes, you see a success message listing the installed packages.

  • Verify the installation by running:
python3 -c "import langgraph; import pydantic; print('All packages installed successfully')"

You should see All packages installed successfully printed in your terminal.

Seeing a ModuleNotFoundError?

Make sure (venv) still appears in your terminal prompt. If it disappeared, re-activate your virtual environment and run the pip install command again.

Help me fix my package installation issue

Configure environment variables and project structure

Your agent needs to know which GCP project and region to use when calling Gemini. You'll store these in a .env file so they stay out of your code.

  • Create a .env file in your social-media-agent folder by running:
cat << 'EOF' > .env
GOOGLE_CLOUD_PROJECT=[[GCP_PROJECT="your-project-id-here"]]
GOOGLE_CLOUD_LOCATION=us-central1
AGENT_MODE=dev
EOF

What are these variables?

  • GOOGLE_CLOUD_PROJECT tells the SDK which GCP project to bill API calls to.
  • GOOGLE_CLOUD_LOCATION sets the region for Vertex AI. us-central1 is a good default with broad model availability.
  • AGENT_MODE controls whether your agent runs in dev mode (human review) or prod mode (AI evaluation).
  • Create the requirements.txt file by running:
cat << 'EOF' > requirements.txt
langgraph==1.2.6
langchain-google-genai==4.2.6
langchain==1.3.11
pydantic==2.11.4
python-dotenv==1.1.0
EOF
  • Create the empty Python files for your project structure by running:
touch config.py models.py tools.py agent.py main.py
  • Confirm all your files exist by running:
ls

You should see .env, requirements.txt, config.py, models.py, tools.py, agent.py, main.py, and a venv folder in the output.

Now verify that your credentials and packages work together by making a test call to Gemini.

  • Run this test script to verify your Vertex AI connection:
python3 -c "
from dotenv import load_dotenv
import os
from langchain.chat_models import init_chat_model

load_dotenv()
os.environ['GOOGLE_CLOUD_PROJECT'] = os.getenv('GOOGLE_CLOUD_PROJECT')
os.environ['GOOGLE_CLOUD_LOCATION'] = os.getenv('GOOGLE_CLOUD_LOCATION', 'us-central1')

model = init_chat_model('google_genai:gemini-2.0-flash', temperature=0.7)
response = model.invoke('Say hello in one sentence.')
print(response.content)
"

You should see a short greeting printed by Gemini. This confirms your authentication, project configuration, and packages all work together.

Getting an authentication or API error?

Double check that your .env file has the correct project ID (the same one you used with gcloud config set project).

Make sure Vertex AI API is enabled in your GCP project. You can check this in the Google Cloud Console under APIs & Services.

Help me fix my Vertex AI connection error

✔️ Awesome, I've got everything!

Great. Double check you've saved your .env and requirements.txt files.

ⓧ I'd like to double check the full code

Here is what your .env file should contain:

GOOGLE_CLOUD_PROJECT=your-project-id-here
GOOGLE_CLOUD_LOCATION=us-central1
AGENT_MODE=dev

Here is what your requirements.txt file should contain:

langgraph==1.2.6
langchain-google-genai==4.2.6
langchain==1.3.11
pydantic==2.11.4
python-dotenv==1.1.0

Your workspace is ready. Next up, you'll define the Pydantic models that give your agent structured, validated outputs and build the first node that researches trending topics.

Define Pydantic Models and the Trend Research Node

Your environment is set up and connected to Google Cloud. Now it is time to define the data structures your agent works with and give it the ability to research trends.

AI models return unstructured text by default. For a production agent, you need typed, validated outputs that downstream systems can rely on. Pydantic models solve this by enforcing a schema on every response. In this step, you will also build the agent's first node: a trend researcher that uses Gemini's tool-calling capability to discover what is popular in a given niche.

In this step, get ready to:
  • Define Pydantic models for structured AI outputs.
  • Build a trend research tool with Gemini's tool-calling capability.
  • Test the research node in isolation.
Define Pydantic structured output models

Before writing any agent logic, you need to set up the project configuration and define the data models that Gemini will return. These models guarantee that every AI response has the exact fields your code expects.

  • Open config.py in your editor and add the following code:
import os
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

GOOGLE_CLOUD_PROJECT = os.getenv("GOOGLE_CLOUD_PROJECT")
GOOGLE_CLOUD_LOCATION = os.getenv("GOOGLE_CLOUD_LOCATION", "us-central1")
MODEL_NAME = "google_genai:gemini-2.0-flash"
CONFIDENCE_THRESHOLD = 0.8
MODE = os.getenv("AGENT_MODE", "dev")

What does this code do?

This file centralizes all configuration for the project. It loads your .env values and sets up constants the rest of the codebase imports.

  • MODEL_NAME uses the provider:model format that LangChain's init_chat_model expects. The google_genai prefix routes requests through Vertex AI when ADC credentials are present.
  • CONFIDENCE_THRESHOLD is the score above which the AI evaluator auto-approves content. You will use this in Step 4.
  • MODE reads from your .env and defaults to dev (human review on every post).
  • Save config.py.

Now you will define the Pydantic models that enforce structure on Gemini's responses.

  • Open models.py and add the InstagramPost model:
from typing import Literal
from pydantic import BaseModel, Field


class InstagramPost(BaseModel):
    caption: str = Field(description="Engaging caption with a hook in the first line, max 2200 characters, includes a call-to-action")
    hashtags: list[str] = Field(description="20-30 relevant hashtags mixing popular, medium, and niche tags")
    visual_description: str = Field(description="Detailed description of the ideal image or carousel to pair with this post")
    best_posting_time: str = Field(description="Suggested posting time and day for maximum engagement")
    content_type: Literal["single_image", "carousel", "reel"] = Field(description="The recommended content format")

What does this model do?

InstagramPost defines the exact shape of every piece of content your agent generates. Each field has a Field(description=...) that tells Gemini what to put there.

The content_type field uses a Literal type, which restricts the value to one of three exact strings. If the model tries to return anything else, Pydantic raises a validation error.

  • Below the InstagramPost class, add the ContentQualityScore model:
class ContentQualityScore(BaseModel):
    engagement_score: float = Field(ge=0, le=1, description="Predicted engagement potential from 0 to 1")
    brand_safety: float = Field(ge=0, le=1, description="Brand safety score from 0 to 1")
    platform_compliance: float = Field(ge=0, le=1, description="Platform rules compliance from 0 to 1")
    overall_confidence: float = Field(ge=0, le=1, description="Overall confidence that this post should be published")
    reasoning: str = Field(description="Brief explanation of the scoring rationale")

What does ContentQualityScore do?

This model is for the AI evaluation agent you will build in Step 4. It scores generated content on multiple dimensions before deciding whether to publish or escalate to a human.

The ge=0, le=1 constraints on each float field tell Pydantic to reject any score outside the 0-to-1 range. This prevents the model from returning nonsensical values like 5.0 or -0.3.

  • Save models.py.

Next you will define the agent's state schema and model initialization. These go in agent.py.

  • Open agent.py and add the imports and environment setup:
from typing_extensions import TypedDict

from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt

from config import MODEL_NAME, GOOGLE_CLOUD_PROJECT, GOOGLE_CLOUD_LOCATION, CONFIDENCE_THRESHOLD
from models import InstagramPost, ContentQualityScore
from tools import search_trending_topics

import os

# Set environment variables so the Google GenAI SDK uses Vertex AI via ADC
os.environ["GOOGLE_CLOUD_PROJECT"] = GOOGLE_CLOUD_PROJECT
os.environ["GOOGLE_CLOUD_LOCATION"] = GOOGLE_CLOUD_LOCATION

Why set environment variables in code?

The langchain-google-genai package uses the underlying google-genai SDK. That SDK checks for GOOGLE_CLOUD_PROJECT and GOOGLE_CLOUD_LOCATION in the process environment to route requests through Vertex AI using your ADC credentials. Setting them here ensures the values from your .env propagate to the SDK at runtime.

  • Below the imports, add the AgentState TypedDict and the get_model helper:
class AgentState(TypedDict):
    niche: str
    trending_topics: str
    generated_content: dict
    quality_score: dict
    review_status: str
    feedback: str
    mode: str


def get_model():
    return init_chat_model(MODEL_NAME, temperature=0.7)

What is AgentState?

In LangGraph, every node in the graph reads from and writes to a shared state object. AgentState defines all the fields that flow between nodes.

  • niche holds the user's topic input (e.g. "fitness").
  • trending_topics stores the research node's output.
  • generated_content and quality_score will hold the Pydantic model outputs as dicts.
  • review_status and feedback track the human review loop.
  • mode determines whether the graph runs in dev or prod mode.
  • Save agent.py.

Seeing an import error?

If your editor underlines the from tools import search_trending_topics line, that is expected. You have not written tools.py yet. The error resolves in the next substep.

help me fix import errors in my LangGraph agent project

Build the trend research node with tool calling

Your agent's first capability is trend research. In LangGraph, each capability lives in a "node", a function that reads state, does work, and returns updated fields. The research node uses Gemini's tool-calling feature to structure its search.

  • Open tools.py and define the search tool:
from langchain.tools import tool


@tool
def search_trending_topics(niche: str) -> str:
    """Search for current trending topics and content angles in a specific niche for social media content creation."""
    return f"Search for the latest trending topics, viral content angles, and popular themes in the '{niche}' niche that would perform well on Instagram. Include specific trend names, hashtag movements, and content formats that are currently gaining traction."

Why is the tool returning a prompt string?

In a production agent, this tool would call a real search API like Tavily or SerpAPI. For this project, the tool returns a structured prompt that Gemini interprets using its own knowledge. This demonstrates the tool-calling pattern without adding extra API key complexity.

The @tool decorator from LangChain converts a regular Python function into a tool object that LLMs can discover, choose to call, and invoke with the correct arguments.

  • Save tools.py.

Now you will write the node function that wires the tool into Gemini's tool-calling flow.

  • Switch back to agent.py and add the research_trends function below get_model:
def research_trends(state: AgentState) -> dict:
    model = get_model()
    tools = [search_trending_topics]
    model_with_tools = model.bind_tools(tools)

    messages = [
        SystemMessage(content="You are a social media trend researcher. Identify current trending topics and content angles for the given niche. Be specific about trends, hashtags, and content formats that are performing well."),
        HumanMessage(content=f"Research current trending topics in the '{state['niche']}' niche for Instagram content. Provide 3-5 specific trending angles with brief explanations of why they are trending.")
    ]

    response = model_with_tools.invoke(messages)

    if response.tool_calls:
        tool_results = []
        for tool_call in response.tool_calls:
            tool_result = search_trending_topics.invoke(tool_call["args"])
            tool_results.append(tool_result)

        follow_up = messages + [response, HumanMessage(content=f"Based on this research context: {tool_results}. Now provide 3-5 specific trending content angles for the '{state['niche']}' niche on Instagram.")]
        final_response = model.invoke(follow_up)
        return {"trending_topics": final_response.content}

    return {"trending_topics": response.content}

What does the research_trends node do?

This function follows the standard LangGraph node pattern: it receives the current state and returns a dict of fields to update.

  • model.bind_tools(tools) tells Gemini about available tools. When it decides a tool would help, it returns a tool_calls list instead of plain text.
  • The if response.tool_calls branch executes each tool call, gathers the results, then sends them back to the model in a follow-up message so it can synthesize a final answer.
  • If Gemini answers directly without calling tools, the function returns that response as-is.
  • Save agent.py.

Editor still showing import errors?

Make sure you saved both tools.py and config.py before checking agent.py. Your editor's language server may need a moment to pick up the new files.

help me debug import resolution in my Python LangGraph project

✔️ Awesome, I've got everything!

Great. Double check you have saved all four files: config.py, models.py, tools.py, and agent.py.

ⓧ I'd like to double check the full code

Here is each file in full. Compare yours against these.

import os
from dotenv import load_dotenv

load_dotenv()

GOOGLE_CLOUD_PROJECT = os.getenv("GOOGLE_CLOUD_PROJECT")
GOOGLE_CLOUD_LOCATION = os.getenv("GOOGLE_CLOUD_LOCATION", "us-central1")
MODEL_NAME = "google_genai:gemini-2.0-flash"
CONFIDENCE_THRESHOLD = 0.8
MODE = os.getenv("AGENT_MODE", "dev")
from typing import Literal
from pydantic import BaseModel, Field


class InstagramPost(BaseModel):
    caption: str = Field(description="Engaging caption with a hook in the first line, max 2200 characters, includes a call-to-action")
    hashtags: list[str] = Field(description="20-30 relevant hashtags mixing popular, medium, and niche tags")
    visual_description: str = Field(description="Detailed description of the ideal image or carousel to pair with this post")
    best_posting_time: str = Field(description="Suggested posting time and day for maximum engagement")
    content_type: Literal["single_image", "carousel", "reel"] = Field(description="The recommended content format")


class ContentQualityScore(BaseModel):
    engagement_score: float = Field(ge=0, le=1, description="Predicted engagement potential from 0 to 1")
    brand_safety: float = Field(ge=0, le=1, description="Brand safety score from 0 to 1")
    platform_compliance: float = Field(ge=0, le=1, description="Platform rules compliance from 0 to 1")
    overall_confidence: float = Field(ge=0, le=1, description="Overall confidence that this post should be published")
    reasoning: str = Field(description="Brief explanation of the scoring rationale")
from langchain.tools import tool


@tool
def search_trending_topics(niche: str) -> str:
    """Search for current trending topics and content angles in a specific niche for social media content creation."""
    return f"Search for the latest trending topics, viral content angles, and popular themes in the '{niche}' niche that would perform well on Instagram. Include specific trend names, hashtag movements, and content formats that are currently gaining traction."
from typing_extensions import TypedDict

from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt

from config import MODEL_NAME, GOOGLE_CLOUD_PROJECT, GOOGLE_CLOUD_LOCATION, CONFIDENCE_THRESHOLD
from models import InstagramPost, ContentQualityScore
from tools import search_trending_topics

import os

os.environ["GOOGLE_CLOUD_PROJECT"] = GOOGLE_CLOUD_PROJECT
os.environ["GOOGLE_CLOUD_LOCATION"] = GOOGLE_CLOUD_LOCATION


class AgentState(TypedDict):
    niche: str
    trending_topics: str
    generated_content: dict
    quality_score: dict
    review_status: str
    feedback: str
    mode: str


def get_model():
    return init_chat_model(MODEL_NAME, temperature=0.7)


def research_trends(state: AgentState) -> dict:
    model = get_model()
    tools = [search_trending_topics]
    model_with_tools = model.bind_tools(tools)

    messages = [
        SystemMessage(content="You are a social media trend researcher. Identify current trending topics and content angles for the given niche. Be specific about trends, hashtags, and content formats that are performing well."),
        HumanMessage(content=f"Research current trending topics in the '{state['niche']}' niche for Instagram content. Provide 3-5 specific trending angles with brief explanations of why they are trending.")
    ]

    response = model_with_tools.invoke(messages)

    if response.tool_calls:
        tool_results = []
        for tool_call in response.tool_calls:
            tool_result = search_trending_topics.invoke(tool_call["args"])
            tool_results.append(tool_result)

        follow_up = messages + [response, HumanMessage(content=f"Based on this research context: {tool_results}. Now provide 3-5 specific trending content angles for the '{state['niche']}' niche on Instagram.")]
        final_response = model.invoke(follow_up)
        return {"trending_topics": final_response.content}

    return {"trending_topics": response.content}
Test the research node in isolation

Before wiring nodes into a full graph, it is good practice to test each one standalone. This confirms your Vertex AI connection works and that the node returns meaningful trend data.

  • In your terminal (with your virtual environment activated), test the research node by running:
python -c "
from agent import research_trends

state = {
    'niche': 'fitness',
    'trending_topics': '',
    'generated_content': {},
    'quality_score': {},
    'review_status': '',
    'feedback': '',
    'mode': 'dev'
}

result = research_trends(state)
print(result['trending_topics'][:500])
"
  • You should see a block of text describing 3-5 trending fitness content angles for Instagram. The exact topics will vary since Gemini generates them dynamically.

Seeing an authentication error?

If you see a message about missing credentials or permission denied, verify your ADC login is active by running:

gcloud auth application-default login

Also confirm that the GOOGLE_CLOUD_PROJECT value in your .env file matches a project where Vertex AI is enabled.

help me fix Vertex AI authentication errors in my Python project

Your agent can now research trends and return structured data. Next up, you will build the content creation node that takes these trends and generates a full Instagram post as a validated Pydantic object.

Build the Content Creation Node with Structured Outputs

Your research node can now gather trending topics from Gemini. The next challenge is turning those raw trends into a finished, publication-ready Instagram post.

In this step, you will build a content creation node that uses Pydantic structured outputs to guarantee Gemini returns a validated, typed object. Then you will wire both nodes into a StateGraph and run the full pipeline end-to-end.

In this step, get ready to:
  • Build a content generation node that returns a validated InstagramPost Pydantic object.
  • Wire both nodes into a compiled StateGraph with a checkpointer.
  • Run the pipeline and verify type-safe access to individual fields.
Create the content generation node with Pydantic output

Structured outputs eliminate fragile string parsing. Instead of asking the LLM for text and hoping it follows a format, you pass a Pydantic model directly to Gemini. The model guarantees every response matches your schema or raises a validation error.

  • In agent.py, find the research_trends function. Below it, add the following create_content function:
def create_content(state: AgentState) -> dict:
    model = get_model()
    structured_model = model.with_structured_output(InstagramPost)

    feedback_context = ""
    if state.get("feedback"):
        feedback_context = f"\n\nPrevious feedback to address: {state['feedback']}"

    messages = [
        SystemMessage(content="You are an expert Instagram content creator. Generate engaging, platform-optimized content that drives engagement. Return structured content following the exact schema provided."),
        HumanMessage(content=f"""Based on these trending topics in the '{state['niche']}' niche:

{state['trending_topics']}
{feedback_context}

Generate an Instagram post. Include an engaging caption with a hook in the first line and a call-to-action. Provide 20-30 relevant hashtags. Describe the ideal visual. Suggest the best posting time. Recommend whether this should be a single_image, carousel, or reel.""")
    ]

    post = structured_model.invoke(messages)

    return {
        "generated_content": post.model_dump(),
        "review_status": "pending"
    }

What does this code do?

model.with_structured_output(InstagramPost) wraps Gemini so that every response is automatically parsed and validated against your InstagramPost Pydantic model. If the response does not match the schema, LangChain retries automatically.

The feedback_context check prepares for future revisions. When a reviewer rejects content, their feedback gets injected here so Gemini can address it on the next attempt.

post.model_dump() converts the Pydantic object to a plain dictionary so it can be stored in the graph state. The state tracks this as generated_content and sets review_status to "pending" for the review step you will add later.

  • Save agent.py.

Seeing a red underline on InstagramPost?

Make sure the import from models import InstagramPost, ContentQualityScore is already at the top of agent.py. This was added in the previous step.

help me debug a missing import in agent.py

Wire the basic graph with edges

Nodes on their own do nothing. A StateGraph connects them with edges that define execution order. You will also add a checkpointer so the graph can persist state between runs.

  • In agent.py, below the create_content function, add the following build_graph function:
def build_graph():
    graph = StateGraph(AgentState)

    graph.add_node("research_trends", research_trends)
    graph.add_node("create_content", create_content)

    graph.add_edge(START, "research_trends")
    graph.add_edge("research_trends", "create_content")
    graph.add_edge("create_content", END)

    checkpointer = InMemorySaver()
    return graph.compile(checkpointer=checkpointer)

What does this code do?

StateGraph(AgentState) creates a graph whose nodes all share the AgentState TypedDict as their shared memory. Each node reads from state and returns updates to it.

The edges define the execution flow: START enters research_trends, which flows into create_content, which exits at END. This is a linear pipeline for now. You will add conditional branching in the next step.

InMemorySaver() gives the graph short-term memory. It stores checkpoints after each node so you can resume, replay, or inspect the graph mid-execution.

  • Save agent.py.

Getting a NameError on StateGraph or START?

Confirm that the import line from langgraph.graph import StateGraph, START, END exists at the top of agent.py. Also confirm from langgraph.checkpoint.memory import InMemorySaver is present.

help me fix a NameError in my LangGraph build_graph function

✔️ Awesome, I've got everything!

Great. Double check you have saved agent.py with both new functions.

ⓧ I'd like to double check the full code

from typing_extensions import TypedDict

from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt

from config import MODEL_NAME, GOOGLE_CLOUD_PROJECT, GOOGLE_CLOUD_LOCATION, CONFIDENCE_THRESHOLD
from models import InstagramPost, ContentQualityScore
from tools import search_trending_topics

import os

os.environ["GOOGLE_CLOUD_PROJECT"] = GOOGLE_CLOUD_PROJECT
os.environ["GOOGLE_CLOUD_LOCATION"] = GOOGLE_CLOUD_LOCATION


class AgentState(TypedDict):
    niche: str
    trending_topics: str
    generated_content: dict
    quality_score: dict
    review_status: str
    feedback: str
    mode: str


def get_model():
    return init_chat_model(MODEL_NAME, temperature=0.7)


def research_trends(state: AgentState) -> dict:
    model = get_model()
    tools = [search_trending_topics]
    model_with_tools = model.bind_tools(tools)

    messages = [
        SystemMessage(content="You are a social media trend researcher. Identify current trending topics and content angles for the given niche. Be specific about trends, hashtags, and content formats that are performing well."),
        HumanMessage(content=f"Research current trending topics in the '{state['niche']}' niche for Instagram content. Provide 3-5 specific trending angles with brief explanations of why they are trending.")
    ]

    response = model_with_tools.invoke(messages)

    if response.tool_calls:
        tool_results = []
        for tool_call in response.tool_calls:
            tool_result = search_trending_topics.invoke(tool_call["args"])
            tool_results.append(tool_result)

        follow_up = messages + [response, HumanMessage(content=f"Based on this research context: {tool_results}. Now provide 3-5 specific trending content angles for the '{state['niche']}' niche on Instagram.")]
        final_response = model.invoke(follow_up)
        return {"trending_topics": final_response.content}

    return {"trending_topics": response.content}


def create_content(state: AgentState) -> dict:
    model = get_model()
    structured_model = model.with_structured_output(InstagramPost)

    feedback_context = ""
    if state.get("feedback"):
        feedback_context = f"\n\nPrevious feedback to address: {state['feedback']}"

    messages = [
        SystemMessage(content="You are an expert Instagram content creator. Generate engaging, platform-optimized content that drives engagement. Return structured content following the exact schema provided."),
        HumanMessage(content=f"""Based on these trending topics in the '{state['niche']}' niche:

{state['trending_topics']}
{feedback_context}

Generate an Instagram post. Include an engaging caption with a hook in the first line and a call-to-action. Provide 20-30 relevant hashtags. Describe the ideal visual. Suggest the best posting time. Recommend whether this should be a single_image, carousel, or reel.""")
    ]

    post = structured_model.invoke(messages)

    return {
        "generated_content": post.model_dump(),
        "review_status": "pending"
    }


def build_graph():
    graph = StateGraph(AgentState)

    graph.add_node("research_trends", research_trends)
    graph.add_node("create_content", create_content)

    graph.add_edge(START, "research_trends")
    graph.add_edge("research_trends", "create_content")
    graph.add_edge("create_content", END)

    checkpointer = InMemorySaver()
    return graph.compile(checkpointer=checkpointer)
Verify structured output

The graph is wired. Time to run the full pipeline and confirm Gemini returns a validated Pydantic object with accessible, typed fields.

  • Run the pipeline end-to-end by executing this command in your terminal:
python -c "
import uuid
from agent import build_graph

graph = build_graph()
config = {'configurable': {'thread_id': str(uuid.uuid4())}}

result = graph.invoke({
    'niche': 'fitness',
    'trending_topics': '',
    'generated_content': {},
    'quality_score': {},
    'review_status': '',
    'feedback': '',
    'mode': 'dev'
}, config)

print('Content type:', type(result['generated_content']))
print('Keys:', list(result['generated_content'].keys()))
print('Caption preview:', result['generated_content']['caption'][:100])
print('Hashtag count:', len(result['generated_content']['hashtags']))
print('Content format:', result['generated_content']['content_type'])
"

This may take 10-20 seconds as Gemini processes both the research and content generation nodes sequentially.

  • You should see output showing a dict type, five keys (caption, hashtags, visual_description, best_posting_time, content_type), a caption preview, a hashtag count between 20 and 30, and a content format of either single_image, carousel, or reel.

Why is this type safety valuable?

Without structured outputs, you would need to parse free-text responses with regex or string splitting. One unexpected format change from the LLM would break your pipeline. Pydantic guarantees every field exists and matches the expected type before your code ever touches it.

In production, this means you can safely serialize the content to a database, pass it to a frontend, or feed it into a downstream service without defensive null checks everywhere.

Seeing a validation error or empty output?

If you see a Pydantic validation error, confirm your models.py has the InstagramPost class with all five fields (caption, hashtags, visual_description, best_posting_time, content_type).

If you see an authentication error, run gcloud auth application-default login again and confirm your .env file has the correct GOOGLE_CLOUD_PROJECT value.

help me debug my LangGraph structured output pipeline

Your pipeline now researches trends and generates validated, type-safe content in a single invocation. Next up, you will add a dual-mode review system so content can be approved by a human in development or scored by an AI evaluator in production.

Add Dual-Mode Review

Your agent can research trends and generate structured posts. But in a real content workflow, publishing without review is risky.

In this step, you will build a dual-mode review system. In development mode, a human always reviews content before publishing. In production mode, an AI evaluation agent auto-approves high-quality content and only routes uncertain posts to a human. This is the pattern companies like LinkedIn and Uber use to scale AI systems without removing human oversight entirely.

In this step, get ready to:
  • Build a human review node that pauses the graph with an interrupt.
  • Build an AI evaluation agent that scores content quality.
  • Wire the dual-mode graph with conditional routing and a CLI runner.
Build the human review node

The human review node is the safety net in development mode. It pauses the entire graph, surfaces the generated content to the user, and waits for either an approval or feedback. If the user rejects the content, the graph loops back to regenerate.

What is an interrupt?

In LangGraph, interrupt() pauses execution at a specific node and surfaces a payload (like the generated content) to the calling code. The graph stays frozen in that state until you resume it with a Command(resume=...) object. Whatever value you pass to resume becomes the return value of interrupt() inside the node.

This requires a checkpointer (like InMemorySaver) so the graph can persist its state while waiting. You already have one configured from the previous step.

  • In agent.py, find the build_graph() function at the bottom of the file, then add the following human_review function above it:
def human_review(state: AgentState) -> dict:
    content = state["generated_content"]
    quality_context = ""
    if state.get("quality_score"):
        qs = state["quality_score"]
        quality_context = f"\n\nAI Evaluation Score: {qs['overall_confidence']:.2f} (Engagement: {qs['engagement_score']:.2f}, Safety: {qs['brand_safety']:.2f}, Compliance: {qs['platform_compliance']:.2f})\nReasoning: {qs['reasoning']}"

    review = interrupt({
        "content": content,
        "quality_context": quality_context,
        "action": "Please review the generated content. Respond with 'approve' or provide feedback for revision."
    })

    if isinstance(review, str) and review.lower().strip() == "approve":
        return {"review_status": "approved", "feedback": ""}
    else:
        return {"review_status": "rejected", "feedback": str(review)}

What does this code do?

  • quality_context conditionally builds a string with the AI evaluation scores. In dev mode this will be empty. In prod mode (when the evaluator routes here), it gives the human reviewer scoring context to help them decide faster.
  • interrupt() pauses the graph and surfaces the content plus context to the caller. The graph freezes until you resume it.
  • After resume, the returned value is checked. If it's the string "approve", the review passes. Anything else is treated as feedback, and review_status is set to "rejected" so the graph can loop back for regeneration.

You won't see a change yet. The next substep adds the AI evaluator that works alongside this node.

Build the AI evaluation agent

The evaluation agent is the production-mode gatekeeper. It scores generated content on multiple dimensions and returns a structured confidence score. If confidence is high enough, the content auto-publishes without human involvement.

  • In agent.py, add the following evaluate_content function above the human_review function you just wrote:
def evaluate_content(state: AgentState) -> dict:
    model = get_model()
    eval_model = model.with_structured_output(ContentQualityScore)

    content = state["generated_content"]

    messages = [
        SystemMessage(content="You are a social media quality evaluator. Score the following Instagram post on engagement potential, brand safety, and platform compliance. Be critical and objective."),
        HumanMessage(content=f"""Evaluate this Instagram post:\n\nCaption: {content['caption']}\n\nHashtags: {', '.join(content['hashtags'])}\n\nVisual: {content['visual_description']}\n\nContent type: {content['content_type']}\n\nScore each dimension from 0 to 1 and provide an overall confidence score for whether this should be published.""")
    ]

    score = eval_model.invoke(messages)

    return {"quality_score": score.model_dump()}

What does this code do?

  • model.with_structured_output(ContentQualityScore) forces Gemini to return a response that matches your Pydantic model exactly. No string parsing needed.
  • The evaluator receives the full generated post (caption, hashtags, visual, content type) and scores it on engagement, brand safety, and platform compliance.
  • score.model_dump() converts the Pydantic object to a dictionary so it can be stored in the graph's state.
  • Save agent.py.
  • Verify both new functions have no syntax errors by running this in your terminal:
python -c "from agent import evaluate_content, human_review; print('Both nodes imported successfully')"

You should see Both nodes imported successfully printed in your terminal.

Seeing an ImportError or SyntaxError?

  • Check that both functions are defined outside of build_graph(). They should be at the module level, not indented inside another function.
  • Make sure the interrupt import is still at the top of your file: from langgraph.types import interrupt.
  • Confirm your f-string quotes match. The triple-quoted f-string in evaluate_content uses """ on both ends.

Help me debug my agent.py import error

Wire the dual-mode graph with conditional routing

Right now your graph runs a straight line: research, create, end. You need routing functions that decide where content goes after creation, and you need to replace the simple linear graph with conditional edges.

What are conditional edges?

In LangGraph, add_conditional_edges connects a source node to multiple possible targets. Instead of a fixed next node, a routing function inspects the current state and returns the name of the node to visit next.

This is what makes the dual-mode architecture possible. The same graph handles both dev and prod flows depending on what the routing function returns at runtime.

  • In agent.py, add these three routing functions between human_review and build_graph():
def route_after_creation(state: AgentState) -> str:
    if state["mode"] == "prod":
        return "evaluate_content"
    return "human_review"


def route_after_evaluation(state: AgentState) -> str:
    score = state.get("quality_score", {})
    confidence = score.get("overall_confidence", 0)
    if confidence >= CONFIDENCE_THRESHOLD:
        return END
    return "human_review"


def route_after_review(state: AgentState) -> str:
    if state["review_status"] == "approved":
        return END
    return "create_content"

What does each routing function do?

  • route_after_creation is the mode switch. In "prod" mode, content goes to the AI evaluator. In "dev" mode, it goes straight to human review.
  • route_after_evaluation checks the AI evaluator's confidence score. If it meets the threshold (0.8 from your config.py), the content auto-publishes. Otherwise it falls through to human review.
  • route_after_review handles the human's decision. Approved content ends the graph. Rejected content loops back to create_content for regeneration with the feedback.
  • Now replace the existing build_graph() function at the bottom of agent.py with this updated version that registers all four nodes and uses conditional edges:
def build_graph():
    graph = StateGraph(AgentState)

    graph.add_node("research_trends", research_trends)
    graph.add_node("create_content", create_content)
    graph.add_node("evaluate_content", evaluate_content)
    graph.add_node("human_review", human_review)

    graph.add_edge(START, "research_trends")
    graph.add_edge("research_trends", "create_content")
    graph.add_conditional_edges("create_content", route_after_creation, ["evaluate_content", "human_review"])
    graph.add_conditional_edges("evaluate_content", route_after_evaluation, ["human_review", END])
    graph.add_conditional_edges("human_review", route_after_review, ["create_content", END])

    checkpointer = InMemorySaver()
    return graph.compile(checkpointer=checkpointer)

What changed in build_graph?

  • Two new nodes are registered: evaluate_content and human_review.
  • The fixed edge from create_content to END is replaced with add_conditional_edges that route through your routing functions.
  • The third argument to add_conditional_edges is a list of possible target nodes. LangGraph uses this to validate that the routing function only returns valid destinations.
  • Save agent.py.
  • Verify the graph compiles correctly by running:
python -c "from agent import build_graph; g = build_graph(); print('Graph compiled with', len(g.nodes), 'nodes')"

You should see Graph compiled with followed by a node count printed in your terminal.

Seeing a NameError for a routing function?

  • Make sure all three routing functions (route_after_creation, route_after_evaluation, route_after_review) are defined above build_graph() in the file. Python reads top to bottom, so the functions must exist before they are referenced.
  • Confirm CONFIDENCE_THRESHOLD is imported at the top of the file: from config import MODEL_NAME, GOOGLE_CLOUD_PROJECT, GOOGLE_CLOUD_LOCATION, CONFIDENCE_THRESHOLD.

Help me fix my graph compilation error

✔️ Awesome, I've got everything!

Great. Double check you have saved agent.py before moving on.

ⓧ I'd like to double check the full code

Here is the complete agent.py at this point:

from typing_extensions import TypedDict

from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt

from config import MODEL_NAME, GOOGLE_CLOUD_PROJECT, GOOGLE_CLOUD_LOCATION, CONFIDENCE_THRESHOLD
from models import InstagramPost, ContentQualityScore
from tools import search_trending_topics

import os

os.environ["GOOGLE_CLOUD_PROJECT"] = GOOGLE_CLOUD_PROJECT
os.environ["GOOGLE_CLOUD_LOCATION"] = GOOGLE_CLOUD_LOCATION


class AgentState(TypedDict):
    niche: str
    trending_topics: str
    generated_content: dict
    quality_score: dict
    review_status: str
    feedback: str
    mode: str


def get_model():
    return init_chat_model(MODEL_NAME, temperature=0.7)


def research_trends(state: AgentState) -> dict:
    model = get_model()
    tools = [search_trending_topics]
    model_with_tools = model.bind_tools(tools)

    messages = [
        SystemMessage(content="You are a social media trend researcher. Identify current trending topics and content angles for the given niche. Be specific about trends, hashtags, and content formats that are performing well."),
        HumanMessage(content=f"Research current trending topics in the '{state['niche']}' niche for Instagram content. Provide 3-5 specific trending angles with brief explanations of why they are trending.")
    ]

    response = model_with_tools.invoke(messages)

    if response.tool_calls:
        tool_results = []
        for tool_call in response.tool_calls:
            tool_result = search_trending_topics.invoke(tool_call["args"])
            tool_results.append(tool_result)

        follow_up = messages + [response, HumanMessage(content=f"Based on this research context: {tool_results}. Now provide 3-5 specific trending content angles for the '{state['niche']}' niche on Instagram.")]
        final_response = model.invoke(follow_up)
        return {"trending_topics": final_response.content}

    return {"trending_topics": response.content}


def create_content(state: AgentState) -> dict:
    model = get_model()
    structured_model = model.with_structured_output(InstagramPost)

    feedback_context = ""
    if state.get("feedback"):
        feedback_context = f"\n\nPrevious feedback to address: {state['feedback']}"

    messages = [
        SystemMessage(content="You are an expert Instagram content creator. Generate engaging, platform-optimized content that drives engagement. Return structured content following the exact schema provided."),
        HumanMessage(content=f"""Based on these trending topics in the '{state['niche']}' niche:\n\n{state['trending_topics']}\n{feedback_context}\n\nGenerate an Instagram post. Include an engaging caption with a hook in the first line and a call-to-action. Provide 20-30 relevant hashtags. Describe the ideal visual. Suggest the best posting time. Recommend whether this should be a single_image, carousel, or reel.""")
    ]

    post = structured_model.invoke(messages)

    return {
        "generated_content": post.model_dump(),
        "review_status": "pending"
    }


def evaluate_content(state: AgentState) -> dict:
    model = get_model()
    eval_model = model.with_structured_output(ContentQualityScore)

    content = state["generated_content"]

    messages = [
        SystemMessage(content="You are a social media quality evaluator. Score the following Instagram post on engagement potential, brand safety, and platform compliance. Be critical and objective."),
        HumanMessage(content=f"""Evaluate this Instagram post:\n\nCaption: {content['caption']}\n\nHashtags: {', '.join(content['hashtags'])}\n\nVisual: {content['visual_description']}\n\nContent type: {content['content_type']}\n\nScore each dimension from 0 to 1 and provide an overall confidence score for whether this should be published.""")
    ]

    score = eval_model.invoke(messages)

    return {"quality_score": score.model_dump()}


def human_review(state: AgentState) -> dict:
    content = state["generated_content"]
    quality_context = ""
    if state.get("quality_score"):
        qs = state["quality_score"]
        quality_context = f"\n\nAI Evaluation Score: {qs['overall_confidence']:.2f} (Engagement: {qs['engagement_score']:.2f}, Safety: {qs['brand_safety']:.2f}, Compliance: {qs['platform_compliance']:.2f})\nReasoning: {qs['reasoning']}"

    review = interrupt({
        "content": content,
        "quality_context": quality_context,
        "action": "Please review the generated content. Respond with 'approve' or provide feedback for revision."
    })

    if isinstance(review, str) and review.lower().strip() == "approve":
        return {"review_status": "approved", "feedback": ""}
    else:
        return {"review_status": "rejected", "feedback": str(review)}


def route_after_creation(state: AgentState) -> str:
    if state["mode"] == "prod":
        return "evaluate_content"
    return "human_review"


def route_after_evaluation(state: AgentState) -> str:
    score = state.get("quality_score", {})
    confidence = score.get("overall_confidence", 0)
    if confidence >= CONFIDENCE_THRESHOLD:
        return END
    return "human_review"


def route_after_review(state: AgentState) -> str:
    if state["review_status"] == "approved":
        return END
    return "create_content"


def build_graph():
    graph = StateGraph(AgentState)

    graph.add_node("research_trends", research_trends)
    graph.add_node("create_content", create_content)
    graph.add_node("evaluate_content", evaluate_content)
    graph.add_node("human_review", human_review)

    graph.add_edge(START, "research_trends")
    graph.add_edge("research_trends", "create_content")
    graph.add_conditional_edges("create_content", route_after_creation, ["evaluate_content", "human_review"])
    graph.add_conditional_edges("evaluate_content", route_after_evaluation, ["human_review", END])
    graph.add_conditional_edges("human_review", route_after_review, ["create_content", END])

    checkpointer = InMemorySaver()
    return graph.compile(checkpointer=checkpointer)

Now you need a CLI runner that lets you switch between dev and prod mode and handles the interrupt/resume loop.

  • Create a new file called main.py in your project folder and add the following code:
import sys
import uuid
from langgraph.types import Command
from agent import build_graph


def main():
    # Check for --prod flag to set mode
    mode = "dev"
    if "--prod" in sys.argv:
        mode = "prod"

    graph = build_graph()
    thread_id = str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}

    # Prompt the user for a content niche
    niche = input("Enter your content niche (e.g., fitness, tech, food): ").strip()
    if not niche:
        niche = "fitness"

    print(f"\nMode: {mode.upper()}")
    print(f"Starting content generation for niche: '{niche}'")
    print("=" * 60)

    # Initialize the full agent state
    initial_input = {
        "niche": niche,
        "trending_topics": "",
        "generated_content": {},
        "quality_score": {},
        "review_status": "",
        "feedback": "",
        "mode": mode
    }

    stream_input = initial_input

    # Main loop: stream the graph, handle interrupts, resume
    while True:
        result = None
        for event in graph.stream(stream_input, config, stream_mode="values"):
            result = event

        state = graph.get_state(config)

        # If no pending tasks, the graph finished
        if not state.tasks:
            print("\n" + "=" * 60)
            print("Content generation complete!")
            if result and result.get("generated_content"):
                content = result["generated_content"]
                print(f"\nCaption: {content.get('caption', '')[:200]}...")
                print(f"Hashtags: {len(content.get('hashtags', []))} tags")
                print(f"Content type: {content.get('content_type', '')}")
                print(f"Visual: {content.get('visual_description', '')[:100]}...")
                if result.get("quality_score"):
                    qs = result["quality_score"]
                    print(f"\nQuality Score: {qs.get('overall_confidence', 0):.2f}")
                    print(f"  Engagement: {qs.get('engagement_score', 0):.2f}")
                    print(f"  Brand Safety: {qs.get('brand_safety', 0):.2f}")
                    print(f"  Compliance: {qs.get('platform_compliance', 0):.2f}")
            break

        # Handle interrupts (human review)
        for task in state.tasks:
            if task.interrupts:
                interrupt_value = task.interrupts[0].value
                print("\n" + "=" * 60)
                print("CONTENT REVIEW REQUIRED")
                print("=" * 60)

                content = interrupt_value.get("content", {})
                print(f"\nCaption:\n{content.get('caption', 'N/A')}")
                print(f"\nHashtags: {', '.join(content.get('hashtags', []))}")
                print(f"\nVisual: {content.get('visual_description', 'N/A')}")
                print(f"Content type: {content.get('content_type', 'N/A')}")
                print(f"Best time: {content.get('best_posting_time', 'N/A')}")

                quality_ctx = interrupt_value.get("quality_context", "")
                if quality_ctx:
                    print(quality_ctx)

                print(f"\nAction: {interrupt_value.get('action', '')}")
                user_input = input("\nYour review (type 'approve' or provide feedback): ").strip()
                # Resume the graph with the user's decision
                stream_input = Command(resume=user_input)
                break


if __name__ == "__main__":
    main()

What does this code do?

  • The --prod flag in sys.argv sets the mode. Without it, the agent defaults to dev mode (human reviews everything).
  • thread_id gives each run a unique identifier so the checkpointer can persist state across interrupts.
  • The while True loop streams the graph until it finishes. If the graph pauses at an interrupt, the loop detects it via state.tasks, shows the content to the user, collects input, and resumes with Command(resume=user_input).
  • When state.tasks is empty, the graph has reached END and the final content is printed.
  • Save main.py.
  • Test dev mode by running:
python main.py
  • Enter a niche when prompted (try fitness).
  • Wait for the agent to research trends and generate content. After a few seconds, you will see a CONTENT REVIEW REQUIRED prompt with the full generated post.
  • Type approve to accept the content, or type feedback like make it shorter to trigger regeneration.

You should see the graph pause for your review, then complete after you approve. The final output shows the caption, hashtag count, content type, and visual description.

  • Now test prod mode by running:
python main.py --prod

In prod mode, the AI evaluator scores the content first. If the confidence score is 0.8 or above, the content auto-approves and you will see the quality scores printed. If below 0.8, the graph routes to human review just like dev mode.

Agent hangs or no output appears?

  • Make sure your .env file has a valid GOOGLE_CLOUD_PROJECT value. The Vertex AI call will fail silently if the project ID is wrong.
  • Confirm your ADC credentials are still active by running gcloud auth application-default print-access-token. If it fails, re-run gcloud auth application-default login.
  • If you see a PermissionDenied error, make sure the Vertex AI API is enabled in your GCP project.

Help me debug my LangGraph agent that is not producing output

✔️ Awesome, I've got everything!

Great. Double check you have saved both agent.py and main.py.

ⓧ I'd like to double check the full code

Here is the complete main.py:

import sys
import uuid
from langgraph.types import Command
from agent import build_graph


def main():
    mode = "dev"
    if "--prod" in sys.argv:
        mode = "prod"

    graph = build_graph()
    thread_id = str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}

    niche = input("Enter your content niche (e.g., fitness, tech, food): ").strip()
    if not niche:
        niche = "fitness"

    print(f"\nMode: {mode.upper()}")
    print(f"Starting content generation for niche: '{niche}'")
    print("=" * 60)

    initial_input = {
        "niche": niche,
        "trending_topics": "",
        "generated_content": {},
        "quality_score": {},
        "review_status": "",
        "feedback": "",
        "mode": mode
    }

    stream_input = initial_input

    while True:
        result = None
        for event in graph.stream(stream_input, config, stream_mode="values"):
            result = event

        state = graph.get_state(config)

        if not state.tasks:
            print("\n" + "=" * 60)
            print("Content generation complete!")
            if result and result.get("generated_content"):
                content = result["generated_content"]
                print(f"\nCaption: {content.get('caption', '')[:200]}...")
                print(f"Hashtags: {len(content.get('hashtags', []))} tags")
                print(f"Content type: {content.get('content_type', '')}")
                print(f"Visual: {content.get('visual_description', '')[:100]}...")
                if result.get("quality_score"):
                    qs = result["quality_score"]
                    print(f"\nQuality Score: {qs.get('overall_confidence', 0):.2f}")
                    print(f"  Engagement: {qs.get('engagement_score', 0):.2f}")
                    print(f"  Brand Safety: {qs.get('brand_safety', 0):.2f}")
                    print(f"  Compliance: {qs.get('platform_compliance', 0):.2f}")
            break

        for task in state.tasks:
            if task.interrupts:
                interrupt_value = task.interrupts[0].value
                print("\n" + "=" * 60)
                print("CONTENT REVIEW REQUIRED")
                print("=" * 60)

                content = interrupt_value.get("content", {})
                print(f"\nCaption:\n{content.get('caption', 'N/A')}")
                print(f"\nHashtags: {', '.join(content.get('hashtags', []))}")
                print(f"\nVisual: {content.get('visual_description', 'N/A')}")
                print(f"Content type: {content.get('content_type', 'N/A')}")
                print(f"Best time: {content.get('best_posting_time', 'N/A')}")

                quality_ctx = interrupt_value.get("quality_context", "")
                if quality_ctx:
                    print(quality_ctx)

                print(f"\nAction: {interrupt_value.get('action', '')}")
                user_input = input("\nYour review (type 'approve' or provide feedback): ").strip()
                stream_input = Command(resume=user_input)
                break


if __name__ == "__main__":
    main()

Your dual-mode review system is complete. You have a development workflow where humans review every post, and a production workflow where AI handles routine approvals. Next up, you will take this further in the Secret Mission by adding multi-dimensional scoring and configurable thresholds.

Secret mission

Full Evaluation Agent with Scoring

Your evaluation agent currently gives a single confidence score. Real production systems score content across multiple dimensions and route decisions based on configurable thresholds. Upgrade your agent with multi-dimensional scoring, a borderline routing path, and evaluation logging for post-analysis.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing cloud costs. Vertex AI API calls were pay-per-use and covered by your GCP premium tier during the project.

Decide whether to keep your resources for future development, or clean them up entirely.

Resources you used:

  • Local project folder with Python files (agent.py, models.py, tools.py, config.py, main.py, .env, evaluation_log.json).
  • Python virtual environment with installed packages (langgraph, langchain-google-genai, pydantic, etc.).
  • Google Cloud ADC credentials stored locally (no ongoing cost).

Keep everything running

No action needed. Choose this if you plan to continue developing this agent or move on to Project 2 (which adds FastAPI, PostgreSQL, Docker, and Cloud Run deployment).

  • There is no ongoing cost. Your project files sit on your local machine.
  • ADC credentials remain stored locally for future Google Cloud development.
  • Your virtual environment stays ready to activate whenever you return.

Pause - I'll come back to this later

Shut down any running processes to free up terminal space. Your project files and credentials stay intact for when you return.

  • Stop any running Python processes in your terminal by pressing Ctrl+C.
  • Deactivate your virtual environment by running:
deactivate
  • Your files, packages, and ADC credentials remain intact for next time.

Delete - I don't want to use this again

Remove all project resources and revoke your ADC credentials. This fully cleans up everything created during the project.

  • Delete your project folder from the Desktop by running:
cd ~/Desktop
rm -rf [[PROJECT_DIR="your-project-folder-name"]]
  • On Windows, use this instead:
cd ~\Desktop
rmdir /s /q [[PROJECT_DIR="your-project-folder-name"]]
  • Revoke your Google Cloud ADC credentials by running:
gcloud auth application-default revoke

This removes the locally stored credentials file. It does not delete your Google Cloud account or project.

Nice Work!

Nice Work!

Nice work. You just built a production-grade AI content agent powered by LangGraph and Vertex AI that automates Instagram content creation with a dual-mode review pipeline.

You learned how to:

  • Built a multi-node LangGraph agent with stateful graph orchestration, tool calling for trend research, and Pydantic structured outputs for type-safe content generation.
  • Connected to Google Gemini via Application Default Credentials for production-grade authentication without API keys in your code.
  • Designed a dual-mode review system using conditional edges for multi-path routing: human-in-the-loop interrupt/resume for dev mode and an AI evaluation agent for prod mode.
  • (Secret Mission) Built a full evaluation agent with multi-dimensional scoring, configurable confidence thresholds, and three routing paths (auto-approve, borderline-to-human, auto-reject-and-regenerate).

Ready to quiz yourself?