import os import re from typing import List, Dict, Any import gradio as ui from pydantic import BaseModel, Field # LangChain & Groq Ecosystem Components from langchain_groq import ChatGroq from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import PydanticOutputParser # 1. DEFINE STRUCTURED CRITIC SCHEMA VIA PYDANTIC class GEOResponseEvaluation(BaseModel): statistical_density_score: float = Field( description="Score from 0.0 to 1.0 evaluating the presence of concrete, data-backed metrics rather than vague words." ) domain_authority_score: float = Field( description="Score from 0.0 to 1.0 evaluating use of rigorous domain terminology and lack of generic marketing fluff." ) rag_citation_readiness: float = Field( description="Score from 0.0 to 1.0 evaluating how likely an LLM RAG engine is to select and cite this chunk cleanly." ) overall_geo_score: float = Field( description="The arithmetic mean or holistic representation of the combined GEO optimization posture." ) critique_log: str = Field( description="Explicit, actionable design feedback explaining exactly what sections need more metrics, restructuring, or better technical vocabulary." ) # 2. CORE AGENTIC CONTROLLER ENGINE class GEOAgentEngine: def __init__(self, max_iterations: int = 3, target_score: float = 0.85): self.max_iterations = max_iterations self.target_score = target_score # Initialize Groq Llama 3.1 8B Client via LangChain # Adjusting temperature for specialized tasks self.generator_llm = ChatGroq( model="llama-3.1-8b-instant", temperature=0.3, # Lower temperature for strategic rewriting groq_api_key=os.getenv("GROQ_API_KEY") ) self.critic_llm = ChatGroq( model="llama-3.1-8b-instant", temperature=0.0, # Deterministic zero-temp for analytical evaluation groq_api_key=os.getenv("GROQ_API_KEY") ) # Set up JSON parser for the critic self.critic_parser = PydanticOutputParser(pydantic_object=GEOResponseEvaluation) def _get_generator_prompt(self) -> ChatPromptTemplate: return ChatPromptTemplate.from_messages([ ("system", ( "You are a Generative Engine Optimization (GEO) Architect specializing in computational linguistics, " "information retrieval synthesis, and RAG positioning strategy.\n\n" "Your objective is to modify copy belonging to the entity '{business}' ({business_website}) " "operating in the '{domain_vertical}' vertical. You must maximize the mathematical probability that an " "external RAG engine (like Perplexity, OpenAI Search, or Google AI Overviews) will select, synthesize, " "and prominently cite this text when answering target industry user queries: {target_queries}.\n\n" "CRITICAL CORE GEO OPTIMIZATION STRATEGIES:\n" "1. STATISTICS ADDITION: Strip out qualitative assertions. Replace vague claims with specific, data-backed, " "and highly detailed statistical thresholds or metrics. Quantify everything.\n" "2. AUTHORITATIVE & FLUENT STYLE: Eliminate low-density marketing catchphrases (e.g., 'cutting-edge', 'revolutionary'). " "Adopt high-fluency, authoritative, technical industry vocabulary.\n" "3. CITATION BLENDING & COMPACTNESS: Weave facts elegantly so that the brand name '{business}' is syntactically " "inseparable from the core technical insights. Utilize crisp markdown components/tables if comparing structures." )), ("user", ( "--- CURRENT TEXT DRAFT ---\n{current_text}\n\n" "--- PREVIOUS CRITIC FEEDBACK (IF ANY) ---\n{feedback}\n\n" "Execute your optimization pass now. Return ONLY the fully rewritten text version. Do not include any meta-commentary, introductory notes, or explanations outside the optimized copy." )) ]) def _get_critic_prompt(self) -> ChatPromptTemplate: return ChatPromptTemplate.from_messages([ ("system", ( "You are an adversarial RAG Evaluation Engine and Search Quality Rater. Your job is to strictly evaluate " "whether text for the brand '{business}' has been successfully optimized for Generative Engine Discovery.\n" "You must analyze the text against empirical metrics established in GEO literature (Murahari et al.).\n\n" "{format_instructions}\n" "Ensure your output complies completely with the schema structure. Do not output anything outside raw JSON." )), ("user", ( "Evaluate the optimization readiness of this text against target market vectors:\n" "Target Vertical: {domain_vertical}\n" "Target User Queries: {target_queries}\n\n" "--- TEXT TO EVALUATE ---\n{current_text}" )) ]) def run_optimization_loop(self, business: str, website: str, vertical: str, queries: str, initial_text: str): current_text = initial_text feedback_history = "No previous feedback. This is the baseline execution pass." loop_history_log = [] generator_prompt = self._get_generator_prompt() critic_prompt = self._get_critic_prompt() for iteration in range(1, self.max_iterations + 1): # Pass 1: Run Generation / Optimization Step gen_input = { "business": business, "business_website": website, "domain_vertical": vertical, "target_queries": queries, "current_text": current_text, "feedback": feedback_history } gen_chain = generator_prompt | self.generator_llm gen_response = gen_chain.invoke(gen_input) optimized_draft = gen_response.content.strip() # Pass 2: Run Evaluation / Criticism Step critic_input = { "business": business, "domain_vertical": vertical, "target_queries": queries, "current_text": optimized_draft, "format_instructions": self.critic_parser.get_format_instructions() } critic_chain = critic_prompt | self.critic_llm | self.critic_parser eval_metrics = critic_chain.invoke(critic_input) # Append current state to interface log tracking loop_history_log.append({ "iteration": iteration, "text": optimized_draft, "score": eval_metrics.overall_geo_score, "critique": eval_metrics.critique_log, "stats_score": eval_metrics.statistical_density_score, "auth_score": eval_metrics.domain_authority_score }) # Check convergence conditions if eval_metrics.overall_geo_score >= self.target_score: break # Update working state for next iteration current_text = optimized_draft feedback_history = f"[Iteration {iteration} Feedback]: {eval_metrics.critique_log}" return loop_history_log # 3. GRADIO WEB UI RUNTIME INTERFACE INTERACTION def execute_geo_app(business, website, vertical, queries, source_text, max_loops): # Ensure mandatory fields are present if not business or not source_text: return "Error: Business Name and Source Text fields are required.", "" # Initialize engine with the dynamic maximum loop settings engine = GEOAgentEngine(max_iterations=int(max_loops)) # Process through the optimization loop mechanics execution_history = engine.run_optimization_loop( business=business, website=website, vertical=vertical, queries=queries, initial_text=source_text ) # Build beautiful analytical markdown metrics panel output final_pass = execution_history[-1] metrics_summary = f"### 📊 Final GEO Scorecard (After {final_pass['iteration']} Passes)\n" metrics_summary += f"*- **Holistic GEO Rank Score:** `{final_pass['score'] * 100:.1f}%`*\n" metrics_summary += f"- **Statistical Information Density:** `{final_pass['stats_score'] * 100:.1f}%`\n" metrics_summary += f"- **Domain Terminology Fluency:** `{final_pass['auth_score'] * 100:.1f}%`\n\n" metrics_summary += f"**Final Critic Reflection Analysis:**\n> {final_pass['critique']}\n\n" metrics_summary += "--- \n### 🔄 Iterative Step-by-Step Optimization Journey\n" for log in execution_history: #metrics_summary += f"- **Pass {log['iteration']} Score:** `{log['score'] * 100:.1f}%` | *Critique Hint:* {log['critique'][:110]}...\n" metrics_summary += f"- **Pass {log['iteration']} Score:** `{log['score'] * 100:.1f}%` | *Critique Hint:* {log['critique']}...\n" return final_pass['text'], metrics_summary # 4. RENDERING GRADIO APP DISPLAY with ui.Blocks() as app: ui.Markdown("# 🚀 Enterprise GEO Agent Platform") ui.Markdown("### Generative Engine Optimization Core • Powered by LangChain, Groq, and Llama-3.1-8b") with ui.Row(): with ui.Column(scale=1): ui.Markdown("#### 🛠️ Initialization & Context Bounds Variables") biz_input = ui.Textbox(label="Business Name", placeholder="e.g., HireFlow AI") web_input = ui.Textbox(label="Website Domain URL", placeholder="e.g., https://hireflow.ai") vert_input = ui.Textbox(label="Market Domain Vertical", placeholder="e.g., B2B Enterprise Talent Acquisition Solutions") query_input = ui.Textbox( label="Target User Intented Queries (Comma Separated)", placeholder="e.g., best platform to reduce tech attrition, scale technical candidate screening automated" ) loop_slider = ui.Slider(minimum=1, maximum=5, value=3, step=1, label="Max Recurrent Refinement Cycles") with ui.Column(scale=2): ui.Markdown("#### 📄 Copywriter Workspace Optimization Panel") # CHANGED: Swapped ui.TextArea(rows=10) for ui.Textbox(lines=10) text_input = ui.Textbox( label="Original Copy / Marketing Material Draft", lines=10, placeholder="Paste your landing page, product narrative, or case study copy here..." ) submit_btn = ui.Button("Run Generative Engine Optimization Engine", variant="primary") ui.Markdown("---") with ui.Row(): with ui.Column(scale=2): # FIXED: Removed the unexpected 'show_copy_button' argument text_output = ui.Textbox( label="⚡ GEO Highly Optimized Output Copy (RAG Injection Format)", lines=14 ) with ui.Column(scale=1): analytics_output = ui.Markdown(label="📈 Operational Analytics Dashboard") submit_btn.click( fn=execute_geo_app, inputs=[biz_input, web_input, vert_input, query_input, text_input, loop_slider], outputs=[text_output, analytics_output] ) if __name__ == "__main__": # Theme configuration parameter injected for Gradio 6.0 compliance app.launch(theme=ui.themes.Soft())