"""Question generation agent using Claude.""" import os import json from typing import Generator from anthropic import Anthropic from .context import ConversationContext from .research import search_web, search_news, research_speaker, fact_check, get_topic_trends def get_anthropic_client() -> Anthropic: """Get Anthropic client with API key from environment.""" api_key = os.getenv("ANTHROPIC_API_KEY") if not api_key: raise ValueError("ANTHROPIC_API_KEY environment variable not set") return Anthropic(api_key=api_key) SYSTEM_PROMPT = """You are an expert at helping people ask insightful, thoughtful questions during Q&A sessions at conferences, classes, and talks. Your role is to: 1. Analyze the conversation/talk context provided 2. Identify key themes, claims, and areas worth exploring 3. Generate high-quality questions that: - Show genuine engagement with the material - Demonstrate critical thinking - Could lead to valuable insights - Are respectful and constructive - Help the asker make a positive impression Question Categories: - CLARIFICATION: Questions that seek to understand something better - DEPTH: Questions that explore a topic more deeply - CONNECTION: Questions that connect ideas to other fields or experiences - CHALLENGE: Respectful questions that probe assumptions or claims - PRACTICAL: Questions about real-world application - FORWARD: Questions about future implications or directions When generating questions, consider: - The speaker's expertise and background - Recent trends and developments in the field - Claims that could be fact-checked or explored further - Connections to current events or other domains You have access to research tools. Use them to: - Research the speaker's background - Fact-check interesting claims - Find recent trends in the topic - Discover relevant news or developments Always explain WHY a question is good - this helps users learn to ask better questions themselves.""" TOOLS = [ { "name": "search_web", "description": "Search the web for information on any topic. Use this to research context, verify claims, or find relevant information.", "input_schema": { "type": "object", "properties": { "query": { "type": "string", "description": "The search query" } }, "required": ["query"] } }, { "name": "search_news", "description": "Search for recent news articles on a topic. Use this to find current events or recent developments.", "input_schema": { "type": "object", "properties": { "query": { "type": "string", "description": "The news search query" }, "days": { "type": "integer", "description": "How many days back to search (default: 7)", "default": 7 } }, "required": ["query"] } }, { "name": "research_speaker", "description": "Research a speaker's background, expertise, and recent activity.", "input_schema": { "type": "object", "properties": { "speaker_name": { "type": "string", "description": "Name of the speaker" }, "topic": { "type": "string", "description": "Optional topic context" } }, "required": ["speaker_name"] } }, { "name": "fact_check", "description": "Fact-check a specific claim or statement made in the talk.", "input_schema": { "type": "object", "properties": { "claim": { "type": "string", "description": "The claim to verify" } }, "required": ["claim"] } }, { "name": "get_topic_trends", "description": "Get recent trends and developments in a specific topic area.", "input_schema": { "type": "object", "properties": { "topic": { "type": "string", "description": "The topic to research" } }, "required": ["topic"] } } ] # Tool display names and icons for UI TOOL_DISPLAY = { "search_web": {"icon": "🔍", "name": "Web Search"}, "search_news": {"icon": "📰", "name": "News Search"}, "research_speaker": {"icon": "👤", "name": "Speaker Research"}, "fact_check": {"icon": "✓", "name": "Fact Check"}, "get_topic_trends": {"icon": "📈", "name": "Trend Analysis"}, } def execute_tool(tool_name: str, tool_input: dict) -> str: """Execute a tool and return the result as a string.""" try: if tool_name == "search_web": result = search_web(tool_input["query"]) elif tool_name == "search_news": result = search_news(tool_input["query"], days=tool_input.get("days", 7)) elif tool_name == "research_speaker": result = research_speaker( tool_input["speaker_name"], topic=tool_input.get("topic", "") ) elif tool_name == "fact_check": result = fact_check(tool_input["claim"]) elif tool_name == "get_topic_trends": result = get_topic_trends(tool_input["topic"]) else: return f"Unknown tool: {tool_name}" # Extract relevant info from result if isinstance(result, dict): if "answer" in result: return f"Summary: {result['answer']}\n\nSources: {json.dumps(result.get('results', [])[:3], indent=2)}" return json.dumps(result, indent=2) return str(result) except Exception as e: return f"Error executing {tool_name}: {str(e)}" def format_agent_log(logs: list[dict]) -> str: """Format agent logs as HTML for display.""" if not logs: return "" html = '
' for log in logs: log_type = log.get("type", "info") if log_type == "thinking": html += f'''
🤔 Thinking... {log.get("message", "")}
''' elif log_type == "tool_call": tool_info = TOOL_DISPLAY.get(log.get("tool"), {"icon": "🔧", "name": log.get("tool")}) html += f'''
{tool_info["icon"]} {tool_info["name"]}
Query: {log.get("input", "")}
''' elif log_type == "tool_result": html += f'''
✓ Result received
{log.get("summary", "")[:150]}...
''' elif log_type == "generating": html += f'''
✨ Generating questions...
''' elif log_type == "complete": html += f'''
🎉 Complete! Generated {log.get("count", 0)} questions
''' elif log_type == "error": html += f'''
❌ Error: {log.get("message", "")}
''' html += '
' return html def analyze_and_generate_questions_stream( context: ConversationContext, speaker_name: str = "", num_questions: int = 5, focus_area: str = "" ) -> Generator[tuple[str, str, dict | None], None, None]: """ Analyze context and generate insightful questions with live progress updates. Yields: Tuple of (agent_log_html, questions_html, result_dict or None) - During processing: yields progress updates with None result - At end: yields final state with result dict """ client = get_anthropic_client() logs = [] def add_log(log_type: str, **kwargs): logs.append({"type": log_type, **kwargs}) return format_agent_log(logs) # Initial state yield add_log("thinking", message="Analyzing transcript context..."), "", None # Build the user message user_message = f"""Based on the following context from a talk/presentation, generate {num_questions} insightful questions. {context.to_prompt_context()} """ if speaker_name: user_message += f"\nSpeaker name: {speaker_name} (please research their background)" if focus_area: user_message += f"\nFocus area: {focus_area}" user_message += """ Please: 1. First, use the research tools to gather additional context (speaker background, fact-check claims, find trends) 2. Then generate questions with explanations of why each is valuable 3. Categorize each question (CLARIFICATION, DEPTH, CONNECTION, CHALLENGE, PRACTICAL, FORWARD) 4. Format your response as a JSON object with this structure: { "analysis": "Brief analysis of the talk's key themes", "questions": [ { "question": "The question text", "category": "CATEGORY", "reasoning": "Why this is a good question", "based_on": "What context/research this was based on" } ], "research_summary": "Summary of research conducted" }""" messages = [{"role": "user", "content": user_message}] # Agent loop with tool use max_iterations = 10 iteration = 0 while iteration < max_iterations: iteration += 1 yield add_log("thinking", message=f"Claude is reasoning (iteration {iteration})..."), "", None response = client.messages.create( model="claude-sonnet-4-20250514", max_tokens=4096, system=SYSTEM_PROMPT, tools=TOOLS, messages=messages ) # Check if we need to handle tool use if response.stop_reason == "tool_use": # Process tool calls tool_results = [] assistant_content = response.content for block in response.content: if block.type == "tool_use": # Log tool call tool_input_display = str(block.input.get("query", block.input.get("speaker_name", block.input.get("claim", block.input.get("topic", ""))))) yield add_log("tool_call", tool=block.name, input=tool_input_display), "", None # Execute tool tool_result = execute_tool(block.name, block.input) # Log result result_summary = tool_result[:200] if len(tool_result) > 200 else tool_result yield add_log("tool_result", summary=result_summary), "", None tool_results.append({ "type": "tool_result", "tool_use_id": block.id, "content": tool_result }) # Store research in context if block.name in ["search_web", "search_news", "fact_check", "get_topic_trends"]: context.add_research( query=str(block.input), summary=tool_result[:500] ) # Add assistant response and tool results to messages messages.append({"role": "assistant", "content": assistant_content}) messages.append({"role": "user", "content": tool_results}) else: # No more tool use, extract final response yield add_log("generating"), "", None final_text = "" for block in response.content: if hasattr(block, "text"): final_text += block.text # Try to parse JSON from response try: json_start = final_text.find("{") json_end = final_text.rfind("}") + 1 if json_start >= 0 and json_end > json_start: json_str = final_text[json_start:json_end] result = json.loads(json_str) # Store questions in context for q in result.get("questions", []): context.add_question( question=q["question"], category=q.get("category", "general"), reasoning=q.get("reasoning", "") ) # Format questions as HTML questions_html = format_questions_html(result) yield add_log("complete", count=len(result.get("questions", []))), questions_html, result return except json.JSONDecodeError: pass # Return raw text if JSON parsing fails result = { "analysis": "Could not parse structured response", "questions": [{"question": final_text, "category": "general", "reasoning": ""}], "research_summary": "" } questions_html = format_questions_html(result) yield add_log("complete", count=1), questions_html, result return # Max iterations reached yield add_log("error", message="Max iterations reached"), "", { "analysis": "Max iterations reached", "questions": [], "research_summary": "" } def format_questions_html(result: dict) -> str: """Format questions result as HTML.""" html = f"

Analysis

{result.get('analysis', '')}

" html += "

Suggested Questions

" for i, q in enumerate(result.get("questions", []), 1): category = q.get("category", "GENERAL") question = q.get("question", "") reasoning = q.get("reasoning", "") html += f"""
{category}
{i}. {question}
Why this is good: {reasoning}
""" return html # Keep the non-streaming version for backward compatibility def analyze_and_generate_questions( context: ConversationContext, speaker_name: str = "", num_questions: int = 5, focus_area: str = "" ) -> dict: """Non-streaming version - returns final result only.""" result = None for _, _, r in analyze_and_generate_questions_stream(context, speaker_name, num_questions, focus_area): if r is not None: result = r return result or {"analysis": "No result", "questions": [], "research_summary": ""} def extract_topics_and_claims(context: ConversationContext) -> dict: """ Extract key topics and claims from the transcript. Args: context: The conversation context Returns: Dictionary with topics and claims """ client = get_anthropic_client() transcript = context.get_full_transcript() if not transcript: return {"topics": [], "claims": []} response = client.messages.create( model="claude-sonnet-4-20250514", max_tokens=1024, messages=[{ "role": "user", "content": f"""Analyze this transcript and extract: 1. Key topics being discussed 2. Notable claims or statements that could be fact-checked or explored Transcript: {transcript} Respond in JSON format: {{ "topics": ["topic1", "topic2", ...], "claims": ["claim1", "claim2", ...] }}""" }] ) try: text = response.content[0].text json_start = text.find("{") json_end = text.rfind("}") + 1 if json_start >= 0 and json_end > json_start: result = json.loads(text[json_start:json_end]) # Update context for topic in result.get("topics", []): context.add_topic(topic) for claim in result.get("claims", []): context.add_claim(claim) return result except (json.JSONDecodeError, IndexError): pass return {"topics": [], "claims": []}