marketing_graph.py

import os
from langgraph.graph import StateGraph, START, END
from langchain_core.tracers.context import tracing_v2_enabled
from duckduckgo_search import DDGS

# ---------------- DuckDuckGo search ----------------
def duckduckgo_search(query, max_results=3):
    try:
        with DDGS() as ddgs:
            results = list(ddgs.text(query, max_results=max_results))
        if not results:
            return "No results found."
        return "\n\n".join([
            f"• {r['title']}\n{r['body']}\nURL: {r['href']}" for r in results
        ])
    except Exception as e:
        return f"[DuckDuckGo Error] {e}"

# ---------------- Node functions ----------------
def market_research(state):
    product_info = state.get("input_data", "")
    search_query = f"market trends, competitors, and insights for {product_info}"
    web_data = duckduckgo_search(search_query, max_results=5)
    return {
        "MarketResearch": (
            f"Research for: {product_info}\n"
            f"Web Insights:\n{web_data}\n"
            "Summary: Market trends show AI-driven health wearables are booming."
        )
    }

def content_planner(state):
    # Simulated bug to demonstrate error handling
    raise RuntimeError("Simulated bug in content planner")

def post_generator(state):
    plan_output = state.get("ContentPlanner", "")
    return {
        "PostGenerator": f"Generated posts based on plan: {plan_output[:100]}..."
    }

def analytics_agent(state):
    post_output = state.get("PostGenerator", "")
    return {
        "AnalyticsAgent": (
            "Predicted engagement: Instagram > LinkedIn > Twitter.\n"
            f"Based on content: {post_output[:120]}..."
        )
    }

# ---------------- Build graph ----------------
def build_marketing_graph():
    state_schema = ("input_data", "MarketResearch", "ContentPlanner", "PostGenerator", "AnalyticsAgent")
    graph = StateGraph(state_schema)

    # Add nodes directly without safe_execute
    graph.add_node("MarketResearch", market_research)
    graph.add_node("ContentPlanner", content_planner)
    graph.add_node("PostGenerator", post_generator)
    graph.add_node("AnalyticsAgent", analytics_agent)

    # Define edges
    graph.add_edge(START, "MarketResearch")
    graph.add_edge("MarketResearch", "ContentPlanner")
    graph.add_edge("ContentPlanner", "PostGenerator")
    graph.add_edge("PostGenerator", "AnalyticsAgent")
    graph.add_edge("AnalyticsAgent", END)

    return graph