Spaces:
Sleeping
Sleeping
File size: 2,462 Bytes
29047e2 7036b1e d65be58 7036b1e 2836fdd 320c4bc 2836fdd 7036b1e 29047e2 2836fdd 29047e2 320c4bc 29047e2 adc4d68 29047e2 320c4bc 29047e2 320c4bc 29047e2 7036b1e 29047e2 adc4d68 29047e2 adc4d68 29047e2 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 |
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
|