Upload 5 files

agent_manager.py

@@ -0,0 +1,25 @@
+# agent_manager.py
+from langchain_ollama import OllamaLLM
+
+class AgentManager:
+    def __init__(self, model_name="llama3.2:latest"):
+        self.llm = OllamaLLM(model=model_name)
+        self.logs = []
+
+    def log(self, message):
+        print(message)
+        self.logs.append(message)
+
+    def marketing_fix_node(self, node_name, input_data, error_msg):
+        prompt = f"""
+You are a marketing AI agent. Node '{node_name}' failed with error:
+{error_msg}
+
+Input:
+{input_data}
+
+Generate proper output for this node as if it executed correctly.
+"""
+        fixed_output = self.llm.invoke(prompt)
+        self.log(f"✅ {node_name} recovered:\n{fixed_output}\n")
+        return fixed_output

main.py

@@ -0,0 +1,78 @@
+import gradio as gr
+from agent_manager import AgentManager
+from marketing_graph import build_marketing_graph
+
+def run_marketing_agent(product_name, audience, goal):
+    manager = AgentManager(model_name="llama3.2:latest")
+    graph = build_marketing_graph()
+
+    state = {
+        "input_data": f"""
+**Product:** {product_name}  
+**Target audience:** {audience}  
+**Goal:** {goal}
+"""
+    }
+
+    current_node = "MarketResearch"
+    executed_nodes = []
+
+    while current_node:
+        executed_nodes.append(current_node)
+        node_func = graph.nodes[current_node]
+
+        try:
+            output = node_func(state)
+        except Exception as e:
+            # Self-healing
+            fixed = manager.marketing_fix_node(current_node, state.get("input_data", ""), str(e))
+            output = {current_node: fixed}
+
+        state.update(output)
+
+        # Determine next node
+        next_nodes = [to_node for (from_node, to_node) in graph.edges if from_node == current_node]
+        current_node = next_nodes[0] if next_nodes else None
+
+    # Get Market Research separately
+    market_output = state.get("MarketResearch", "No Market Research generated")
+
+    # Format full workflow output in Markdown
+    sections = []
+    emoji_map = {
+        "MarketResearch": "📝 Market Research",
+        "ContentPlanner": "📅 Content Plan",
+        "PostGenerator": "📣 Generated Posts",
+        "AnalyticsAgent": "📊 Analytics & Suggestions"
+    }
+    for node in executed_nodes:
+        content = state.get(node, "")
+        if content:
+            sections.append(f"## {emoji_map.get(node, node)}")
+            sections.append(content)
+
+    sections.append("## ✨ FINAL MARKETING RESULT ✨")
+    sections.append(f"```json\n{state}\n```")
+
+    full_strategy = "\n\n".join(sections)
+
+    return market_output, full_strategy
+
+# --- Gradio Interface ---
+iface = gr.Interface(
+    fn=run_marketing_agent,
+    inputs=[
+        gr.Textbox(label="Product Name", value="Smart fitness band"),
+        gr.Textbox(label="Target Audience", value="young professionals"),
+        gr.Textbox(label="Goal", value="Increase brand awareness")
+    ],
+    outputs=[
+        gr.Markdown(label="Market Research Output"),
+        gr.Markdown(label="Full Marketing Strategy")
+    ],
+    title="Self-Healing Marketing Agent",
+    description="Enter your product info and get both Market Research and the full marketing strategy."
+)
+
+if __name__ == "__main__":
+    iface.launch()

marketing_graph.py

@@ -0,0 +1,40 @@
+# marketing_graph.py
+from langgraph.graph.state import StateGraph
+
+# Node functions
+def market_research(state):
+    product_info = state.get("input_data", "")
+    return {
+        "MarketResearch": f"Research: {product_info}\nCompetitors: FitTrack, SmartPulse, ZenBand\nTrends: AI fitness, wearable analytics, app integration."
+    }
+
+def content_planner(state):
+    # Simulate failure
+    raise RuntimeError("Simulated bug in content planner")
+
+def post_generator(state):
+    plan_output = state.get("ContentPlanner", "")
+    return {"PostGenerator": f"Generated posts for campaign: 'Get Fit with SmartPulse!'"}
+
+def analytics_agent(state):
+    post_output = state.get("PostGenerator", "")
+    return {"AnalyticsAgent": f"Predicted engagement: Instagram > LinkedIn > Twitter. Suggestion: add CTA."}
+
+def build_marketing_graph():
+    # ✅ Use a tuple/list of node names as state_schema (hashable)
+    state_schema = ("MarketResearch", "ContentPlanner", "PostGenerator", "AnalyticsAgent", "input_data")
+    
+    graph = StateGraph(state_schema)  # positional argument works now
+
+    # Add nodes (function directly)
+    graph.add_node("MarketResearch", market_research)
+    graph.add_node("ContentPlanner", content_planner)
+    graph.add_node("PostGenerator", post_generator)
+    graph.add_node("AnalyticsAgent", analytics_agent)
+
+    # Add edges
+    graph.add_edge("MarketResearch", "ContentPlanner")
+    graph.add_edge("ContentPlanner", "PostGenerator")
+    graph.add_edge("PostGenerator", "AnalyticsAgent")
+
+    return graph