Spaces:
Sleeping
Sleeping
| """Streamlit UI for the content generation agent. | |
| Flow: | |
| Sidebar: provider + API key + model + mock toggle (user pastes any key) | |
| 1. Connect website -> crawl + Brand Brain | |
| 2. Prompt helper -> suggestions + rough note -> editable brief | |
| 3. Generate -> ranked variants per channel with guardrail badges | |
| 4. PPT export -> download .pptx | |
| """ | |
| import os | |
| import tempfile | |
| import streamlit as st | |
| from core.config import (get_settings, MODEL_CHOICES, DEFAULT_MODELS) | |
| from core.pipeline import ContentAgent | |
| from core.schemas import Brief | |
| st.set_page_config(page_title="Content Agent", page_icon="✍️", layout="wide") | |
| def get_agent() -> ContentAgent: | |
| s = st.session_state | |
| settings = get_settings( | |
| provider=s.get("provider"), | |
| api_key=s.get("api_key") or None, | |
| model=s.get("model") or None, | |
| mock_mode=s.get("mock_mode", True), | |
| ) | |
| return ContentAgent(settings) | |
| # ---------------- Sidebar: model / key config ---------------- | |
| with st.sidebar: | |
| st.header("⚙️ Model settings") | |
| st.caption("Paste your own API key and pick a model, or run in mock mode.") | |
| st.session_state["mock_mode"] = st.toggle( | |
| "Mock mode (no key needed)", value=st.session_state.get("mock_mode", True)) | |
| provider = st.selectbox("Provider", ["groq", "openrouter"], | |
| index=0, disabled=st.session_state["mock_mode"]) | |
| st.session_state["provider"] = provider | |
| st.session_state["api_key"] = st.text_input( | |
| "API key", type="password", disabled=st.session_state["mock_mode"], | |
| placeholder="Paste Groq or OpenRouter key") | |
| model_options = MODEL_CHOICES.get(provider, []) | |
| chosen = st.selectbox("Model", model_options, | |
| index=0, disabled=st.session_state["mock_mode"]) | |
| custom = st.text_input("Custom model (optional)", | |
| disabled=st.session_state["mock_mode"], | |
| placeholder="override model id") | |
| st.session_state["model"] = custom or chosen | |
| if st.session_state["mock_mode"]: | |
| st.info("Running in mock mode — deterministic sample output, $0 cost.") | |
| elif not st.session_state["api_key"]: | |
| st.warning("Enter an API key or enable mock mode.") | |
| st.title("✍️ Content Generation Agent") | |
| st.caption("Daily product & marketing content for founders — grounded, guardrailed, multi-channel.") | |
| # ---------------- Step 1: website ---------------- | |
| st.subheader("1️⃣ Connect your website") | |
| col1, col2 = st.columns([3, 1]) | |
| with col1: | |
| website = st.text_input("Product website URL", placeholder="https://yourproduct.com") | |
| with col2: | |
| st.write("") | |
| st.write("") | |
| crawl_btn = st.button("Crawl & analyze", type="primary", use_container_width=True) | |
| if crawl_btn and website: | |
| with st.spinner("Crawling website and building Brand Brain…"): | |
| try: | |
| agent = get_agent() | |
| brand = agent.ingest_website(website) | |
| st.session_state["brand"] = brand.to_dict() | |
| st.success(f"Brand Brain ready for {brand.product_name}") | |
| except Exception as e: | |
| st.error(f"Failed: {e}") | |
| if st.session_state.get("brand"): | |
| b = st.session_state["brand"] | |
| with st.expander("🧠 Brand Brain (review the grounded facts)", expanded=False): | |
| st.write(f"**{b['product_name']}** — {b['one_liner']}") | |
| c1, c2 = st.columns(2) | |
| c1.write("**Value props**") | |
| c1.markdown("\n".join(f"- {x}" for x in b.get("value_props", [])) if b.get("value_props") else "_None_") | |
| c1.write("**Features**") | |
| c1.markdown("\n".join(f"- {x}" for x in b.get("features", [])) if b.get("features") else "_None_") | |
| c2.write("**Differentiators**") | |
| c2.markdown("\n".join(f"- {x}" for x in b.get("differentiators", [])) if b.get("differentiators") else "_None_") | |
| c2.write("**Cannot claim (guardrail)**") | |
| c2.markdown("\n".join(f"- {x}" for x in b.get("forbidden_claims", [])) if b.get("forbidden_claims") else "_None_") | |
| # ---------------- Step 2: prompt helper + brief ---------------- | |
| if st.session_state.get("brand"): | |
| from core.schemas import BrandBrain | |
| brand_obj = BrandBrain(**st.session_state["brand"]) | |
| st.subheader("2️⃣ Describe today's update") | |
| if st.button("💡 Suggest prompts"): | |
| with st.spinner("Thinking of prompt ideas…"): | |
| st.session_state["suggestions"] = get_agent().suggest_prompts(brand_obj) | |
| for sug in st.session_state.get("suggestions", []): | |
| st.caption("→ " + sug) | |
| raw = st.text_area("Your rough note", placeholder="e.g. shipped one-click integrations today") | |
| cc1, cc2, cc3 = st.columns(3) | |
| channels = cc1.multiselect("Channels", ["linkedin", "instagram", "whatsapp"], | |
| default=["linkedin", "instagram", "whatsapp"]) | |
| num_variants = cc2.slider("Variants per channel", 1, 5, 3) | |
| language = cc3.selectbox("Language", ["en", "hi (coming soon)"], index=0) | |
| lang_code = "en" if language.startswith("en") else "hi" | |
| if st.button("Build editable brief"): | |
| if not channels: | |
| st.error("Please select at least one channel in Step 2.") | |
| else: | |
| with st.spinner("Structuring your brief…"): | |
| built = get_agent().build_brief(raw, brand_obj, channels, num_variants, lang_code) | |
| st.session_state["brief"] = built["brief"].to_dict() | |
| st.session_state["suggested_prompt"] = built["suggested_prompt"] | |
| if st.session_state.get("brief"): | |
| st.subheader("3️⃣ Review & customize the brief") | |
| br = st.session_state["brief"] | |
| pcol1, pcol2 = st.columns(2) | |
| br["objective"] = pcol1.selectbox( | |
| "Objective", ["daily_update", "feature", "launch", "learning", "hiring", "metric"], | |
| index=max(0, ["daily_update", "feature", "launch", "learning", "hiring", "metric"].index(br["objective"]) if br["objective"] in ["daily_update", "feature", "launch", "learning", "hiring", "metric"] else 0)) | |
| br["tone"] = pcol2.selectbox( | |
| "Tone", ["professional", "founder_story", "casual", "punchy"], | |
| index=max(0, ["professional", "founder_story", "casual", "punchy"].index(br["tone"]) if br["tone"] in ["professional", "founder_story", "casual", "punchy"] else 0)) | |
| br["audience"] = pcol1.text_input("Audience", value=br.get("audience", "")) | |
| br["cta"] = pcol2.text_input("CTA", value=br.get("cta", "")) | |
| st.session_state["suggested_prompt"] = st.text_area( | |
| "Suggested prompt (edit freely)", value=st.session_state.get("suggested_prompt", "")) | |
| br["raw_input"] = st.session_state["suggested_prompt"] | |
| st.session_state["brief"] = br | |
| if st.button("🚀 Generate content", type="primary"): | |
| if not br.get("channels"): | |
| st.error("Please select at least one channel in Step 2 before generating content.") | |
| else: | |
| with st.spinner("Generating grounded, guardrailed content…"): | |
| agent = get_agent() | |
| brand_obj = BrandBrain(**st.session_state["brand"]) | |
| brief_obj = Brief(**br) | |
| result = agent.generate(brief_obj, brand_obj) | |
| st.session_state["result"] = result | |
| # ---------------- Step 4: results + PPT ---------------- | |
| if st.session_state.get("result"): | |
| result = st.session_state["result"] | |
| st.subheader("4️⃣ Generated content") | |
| if not result.variants_by_channel: | |
| st.warning("⚠️ No channels were selected or generated.") | |
| else: | |
| tabs = st.tabs([c.capitalize() for c in result.variants_by_channel.keys()]) | |
| for tab, (channel, variants) in zip(tabs, result.variants_by_channel.items()): | |
| with tab: | |
| for i, v in enumerate(variants): | |
| badge = "✅ clean" if v.guardrails.passed else "⚠️ " + str(len(v.guardrails.all_flags())) + " flag(s)" | |
| st.markdown(f"**Variant {i+1}** · score {v.score} · {badge}") | |
| st.text_area(f"{channel}_{i}", value=v.text, height=140, label_visibility="collapsed") | |
| if v.hashtags: | |
| st.caption(" ".join(v.hashtags)) | |
| if not v.guardrails.passed: | |
| for f in v.guardrails.all_flags(): | |
| st.caption("⚠️ " + f) | |
| st.divider() | |
| st.subheader("📊 Export PPT") | |
| if st.button("Generate PPT deck"): | |
| with st.spinner("Building deck…"): | |
| out = os.path.join(tempfile.gettempdir(), "content_update.pptx") | |
| path = get_agent().export_ppt(result.fact_pack, result.brief, out) | |
| with open(path, "rb") as fh: | |
| st.download_button("⬇️ Download .pptx", fh.read(), | |
| file_name="content_update.pptx", | |
| mime="application/vnd.openxmlformats-officedocument.presentationml.presentation") | |