Spaces:
Sleeping
Sleeping
| import streamlit as st | |
| from crewai import Agent, Task, Crew | |
| import os | |
| from langchain_groq import ChatGroq | |
| from langchain_openai import ChatOpenAI | |
| from fpdf import FPDF | |
| import pandas as pd | |
| import plotly.express as px | |
| import tempfile | |
| import time | |
| import ast | |
| import logging | |
| # Setup logging | |
| logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') | |
| # Title and Application Introduction | |
| st.title("Patent Strategy and Innovation Consultant") | |
| st.sidebar.write( | |
| "This application uses AI to provide actionable insights and comprehensive analysis for patent-related strategies." | |
| ) | |
| # User Input Section | |
| st.sidebar.header("User Inputs") | |
| patent_area = st.text_input("Enter Patent Technology Area", value="Transparent Antennas for Windshields") | |
| stakeholder = st.text_input("Enter Stakeholder", value="Patent Attorneys") | |
| # Initialize LLM | |
| llm = None | |
| # Model Selection | |
| model_choice = st.radio("Select LLM", ["GPT-4o", "llama-3.3-70b"], index=1, horizontal=True) | |
| # API Key Validation and LLM Initialization | |
| groq_api_key = os.getenv("GROQ_API_KEY") | |
| openai_api_key = os.getenv("OPENAI_API_KEY") | |
| #llm = ChatGroq(groq_api_key=os.getenv("GROQ_API_KEY"), model="groq/llama-3.3-70b-versatile") | |
| if model_choice == "llama-3.3-70b": | |
| if not groq_api_key: | |
| st.error("Groq API key is missing. Please set the GROQ_API_KEY environment variable.") | |
| llm = None | |
| else: | |
| llm = ChatGroq(groq_api_key=groq_api_key, model="groq/llama-3.3-70b-versatile") | |
| elif model_choice == "GPT-4o": | |
| if not openai_api_key: | |
| st.error("OpenAI API key is missing. Please set the OPENAI_API_KEY environment variable.") | |
| llm = None | |
| else: | |
| llm = ChatOpenAI(api_key=openai_api_key, model="gpt-4o") | |
| # Advanced Options | |
| st.sidebar.header("Advanced Options") | |
| enable_advanced_analysis = st.sidebar.checkbox("Enable Advanced Analysis", value=True) | |
| enable_custom_visualization = st.sidebar.checkbox("Enable Custom Visualizations", value=True) | |
| # Agent Customization | |
| st.sidebar.header("Agent Customization") | |
| with st.sidebar.expander("Customize Agent Goals", expanded=False): | |
| enable_customization = st.checkbox("Enable Custom Goals") | |
| if enable_customization: | |
| planner_goal = st.text_area( | |
| "Planner Goal", | |
| value="Research trends in patent filings and technological innovation, identify key players, and provide strategic recommendations." | |
| ) | |
| writer_goal = st.text_area( | |
| "Writer Goal", | |
| value="Craft a professional insights document summarizing trends, strategies, and actionable outcomes for stakeholders." | |
| ) | |
| analyst_goal = st.text_area( | |
| "Analyst Goal", | |
| value=( | |
| "Perform detailed statistical analysis of patent filings, growth trends, and innovation distribution. " | |
| "Identify top assignees/companies in the transparent antenna industry. " | |
| "Provide structured output in a list of dictionaries with 'Category' and 'Values' keys for clear data presentation." | |
| ) | |
| ) | |
| else: | |
| planner_goal = "Research trends in patent filings and technological innovation, identify key players, and provide strategic recommendations." | |
| writer_goal = "Craft a professional insights document summarizing trends, strategies, and actionable outcomes for stakeholders." | |
| analyst_goal = ( | |
| "Perform detailed statistical analysis of patent filings, growth trends, and innovation distribution. " | |
| "Identify top assignees/companies in the transparent antenna industry. " | |
| "Provide structured output in a list of dictionaries with 'Category' and 'Values' keys for clear data presentation." | |
| ) | |
| # Agent Definitions | |
| planner = Agent( | |
| role="Patent Research Consultant", | |
| goal=planner_goal, | |
| backstory=( | |
| "You're tasked with researching {topic} patents and identifying key trends and players. Your work supports the Patent Writer and Data Analyst." | |
| ), | |
| allow_delegation=False, | |
| verbose=True, | |
| llm=llm | |
| ) | |
| writer = Agent( | |
| role="Patent Insights Writer", | |
| goal=writer_goal, | |
| backstory=( | |
| "Using the research from the Planner and data from the Analyst, craft a professional document summarizing patent insights for {stakeholder}." | |
| ), | |
| allow_delegation=False, | |
| verbose=True, | |
| llm=llm | |
| ) | |
| analyst = Agent( | |
| role="Patent Data Analyst", | |
| goal=analyst_goal, | |
| backstory=( | |
| "Analyze patent filing data and innovation trends in {topic} to provide statistical insights. Your analysis will guide the Writer's final report." | |
| ), | |
| allow_delegation=False, | |
| verbose=True, | |
| llm=llm | |
| ) | |
| # Task Definitions | |
| plan = Task( | |
| description=( | |
| "1. Research recent trends in {topic} patent filings and innovation.\n" | |
| "2. Identify key players and emerging technologies.\n" | |
| "3. Provide recommendations for stakeholders on strategic directions.\n" | |
| "4. Identify key statistics such as top regions, top players, and hot areas of innovation.\n" | |
| "5. Limit the output to 600 words." | |
| ), | |
| expected_output="A research document with structured insights, strategic recommendations, and key statistics.", | |
| agent=planner | |
| ) | |
| write = Task( | |
| description=( | |
| "1. Use the Planner's and Analyst's outputs to craft a professional patent insights document.\n" | |
| "2. Include key findings, visual aids, and actionable strategies.\n" | |
| "3. Suggest strategic directions and highlight untapped innovation areas.\n" | |
| "4. Incorporate summarized tables for key statistics and example inventions.\n" | |
| "5. Limit the document to 600 words." | |
| ), | |
| expected_output="A polished, stakeholder-ready patent insights document with actionable recommendations.", | |
| agent=writer | |
| ) | |
| analyse = Task( | |
| description=( | |
| "1. Conduct a comprehensive statistical analysis of patent filing trends, innovation hot spots, and future growth projections in the transparent antenna industry.\n" | |
| "2. Identify and rank the top regions, leading assignees/companies driving innovation.\n" | |
| "3. Highlight regional innovation trends and the distribution of emerging technologies across different geographies.\n" | |
| "4. Provide actionable insights and strategic recommendations based on the data.\n" | |
| "5. Deliver structured output in a list of dictionaries with 'Category' and 'Values' fields:\n" | |
| " - 'Values' can be:\n" | |
| " a) A dictionary with counts for quantitative data (e.g., {{'Region A': 120, 'Region B': 95}}),\n" | |
| " b) A list of key items (technologies, companies, inventors), or\n" | |
| " c) Descriptive text for qualitative insights.\n" | |
| "6. Example Output Format:\n" | |
| "[\n" | |
| " {{'Category': 'Top Regions', 'Values': {{'North America': 120, 'Europe': 95, 'Asia-Pacific': 85}}}},\n" | |
| " {{'Category': 'Top Assignees', 'Values': {{'Company A': 40, 'Company B': 35}}}},\n" | |
| " {{'Category': 'Emerging Technologies', 'Values': ['Graphene Antennas', '5G Integration']}},\n" | |
| " {{'Category': 'Strategic Insights', 'Values': 'Collaborations between automotive and material science industries are accelerating innovation.'}}\n" | |
| "]\n" | |
| "7. Ensure that the output is clean, well-structured, and formatted for use in visualizations and tables." | |
| ), | |
| expected_output="A structured, well-organized dataset with numeric, list-based, and descriptive insights for comprehensive visual and tabular reporting.", | |
| agent=analyst | |
| ) | |
| crew = Crew( | |
| agents=[planner, analyst, writer], | |
| tasks=[plan, analyse, write], | |
| verbose=True | |
| ) | |
| # PDF Report Generation | |
| def generate_pdf_report(result, charts=None, table_data=None, metadata=None): | |
| with tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as temp_pdf: | |
| pdf = FPDF() | |
| pdf.add_page() | |
| pdf.set_font("Arial", size=12) | |
| pdf.set_auto_page_break(auto=True, margin=15) | |
| pdf.set_font("Arial", size=16, style="B") | |
| pdf.cell(200, 10, txt="Patent Strategy and Innovation Report", ln=True, align="C") | |
| pdf.ln(10) | |
| if metadata: | |
| pdf.set_font("Arial", size=10) | |
| for key, value in metadata.items(): | |
| pdf.cell(200, 10, txt=f"{key}: {value}", ln=True) | |
| pdf.set_font("Arial", size=12) | |
| pdf.multi_cell(0, 10, txt=result) | |
| if charts: | |
| for chart_path in charts: | |
| try: | |
| pdf.add_page() | |
| pdf.image(chart_path, x=10, y=20, w=180) | |
| logging.info(f"Successfully included chart: {chart_path}") | |
| except Exception as e: | |
| logging.error(f"Failed to include chart in PDF: {chart_path}. Error: {e}") | |
| if table_data: | |
| pdf.add_page() | |
| pdf.set_font("Arial", size=10) | |
| pdf.cell(200, 10, txt="Consolidated Table:", ln=True, align="L") | |
| for row in table_data: | |
| pdf.cell(200, 10, txt=str(row), ln=True) | |
| pdf.output(temp_pdf.name) | |
| return temp_pdf.name | |
| # Data Validation | |
| def validate_analyst_output(analyst_output): | |
| if not analyst_output: | |
| st.warning("No data available for analysis.") | |
| return None | |
| if not isinstance(analyst_output, list) or not all(isinstance(item, dict) for item in analyst_output): | |
| st.warning("Analyst output must be a list of dictionaries.") | |
| return None | |
| required_keys = {'Category', 'Values'} | |
| if not all(required_keys.issubset(item.keys()) for item in analyst_output): | |
| st.warning(f"Each dictionary must contain keys: {required_keys}") | |
| return None | |
| return analyst_output | |
| # Visualization and Table Display | |
| def create_visualizations(analyst_output): | |
| chart_paths = [] | |
| validated_data = validate_analyst_output(analyst_output) | |
| if validated_data: | |
| for item in validated_data: | |
| category = item["Category"] | |
| values = item["Values"] | |
| try: | |
| # Handle dictionary data | |
| if isinstance(values, dict): | |
| df = pd.DataFrame(list(values.items()), columns=["Label", "Count"]) | |
| # Choose Pie Chart for fewer categories, else Bar Chart | |
| if len(df) <= 5: | |
| chart = px.pie(df, names="Label", values="Count", title=f"{category} Distribution") | |
| else: | |
| chart = px.bar(df, x="Label", y="Count", title=f"{category} Analysis") | |
| # Handle list data | |
| elif isinstance(values, list): | |
| # Convert the list into a frequency count without dummy values | |
| df = pd.DataFrame(values, columns=["Label"]) | |
| df = df["Label"].value_counts().reset_index() | |
| df.columns = ["Label", "Count"] | |
| # Plot as a bar chart or pie chart | |
| if len(df) <= 5: | |
| chart = px.pie(df, names="Label", values="Count", title=f"{category} Distribution") | |
| else: | |
| chart = px.bar(df, x="Label", y="Count", title=f"{category} Frequency") | |
| # Handle text data | |
| elif isinstance(values, str): | |
| st.subheader(f"{category} Insights") | |
| st.table(pd.DataFrame({"Insights": [values]})) | |
| continue # No chart for text data | |
| else: | |
| st.warning(f"Unsupported data format for category: {category}") | |
| continue | |
| # Display the chart in Streamlit | |
| st.plotly_chart(chart) | |
| # Save the chart for PDF export | |
| with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as temp_chart: | |
| chart.write_image(temp_chart.name) | |
| chart_paths.append(temp_chart.name) | |
| except Exception as e: | |
| st.error(f"Failed to generate visualization for {category}: {e}") | |
| logging.error(f"Error in {category} visualization: {e}") | |
| return chart_paths | |
| def display_table(analyst_output): | |
| table_data = [] | |
| validated_data = validate_analyst_output(analyst_output) | |
| if validated_data: | |
| for item in validated_data: | |
| category = item["Category"] | |
| values = item["Values"] | |
| # Error handling to prevent crashes | |
| try: | |
| # Handle dictionary data (Table View) | |
| if isinstance(values, dict): | |
| df = pd.DataFrame(list(values.items()), columns=["Label", "Count"]) | |
| st.subheader(f"{category} (Table View)") | |
| st.dataframe(df) | |
| table_data.extend(df.to_dict(orient="records")) | |
| # Handle list data (List View) | |
| elif isinstance(values, list): | |
| df = pd.DataFrame(values, columns=["Items"]) | |
| st.subheader(f"{category} (List View)") | |
| st.dataframe(df) | |
| table_data.extend(df.to_dict(orient="records")) | |
| # Handle text data (Summary View) | |
| elif isinstance(values, str): | |
| st.subheader(f"{category} (Summary)") | |
| st.table(pd.DataFrame({"Insights": [values]})) | |
| table_data.append({"Category": category, "Values": values}) | |
| else: | |
| st.warning(f"Unsupported data format for category: {category}") | |
| except Exception as e: | |
| logging.error(f"Error processing {category}: {e}") | |
| st.error(f"Failed to display {category} as a table due to an error.") | |
| return table_data | |
| def parse_analyst_output(raw_output): | |
| structured_data = [] | |
| current_category = None | |
| current_values = [] | |
| # Split raw output by line | |
| lines = raw_output.split('\n') | |
| for line in lines: | |
| line = line.strip() | |
| # Detect the start of a new category | |
| if line.startswith("Category:"): | |
| # Save the previous category and its values | |
| if current_category and current_values: | |
| structured_data.append({ | |
| "Category": current_category, | |
| "Values": current_values if len(current_values) > 1 else current_values[0] | |
| }) | |
| # Start processing the new category | |
| current_category = line.replace("Category:", "").strip() | |
| current_values = [] | |
| # Skip 'Values:' header | |
| elif line.startswith("Values:"): | |
| continue | |
| # Process the values under the current category | |
| elif line and current_category: | |
| try: | |
| # Attempt to convert the line into Python data (dict/list) | |
| parsed_value = ast.literal_eval(line) | |
| current_values.append(parsed_value) | |
| except (ValueError, SyntaxError): | |
| # If parsing fails, treat it as plain text | |
| current_values.append(line) | |
| # Save the last processed category | |
| if current_category and current_values: | |
| structured_data.append({ | |
| "Category": current_category, | |
| "Values": current_values if len(current_values) > 1 else current_values[0] | |
| }) | |
| return structured_data | |
| # Main Execution Block | |
| if st.button("Generate Patent Insights"): | |
| with st.spinner('Processing...'): | |
| try: | |
| start_time = time.time() | |
| results = crew.kickoff(inputs={"topic": patent_area, "stakeholder": stakeholder}) | |
| elapsed_time = time.time() - start_time | |
| writer_output = getattr(results.tasks_output[2], "raw", "No details available.") | |
| if writer_output: | |
| st.markdown("### Final Report") | |
| st.write(writer_output) | |
| else: | |
| st.warning("No final report available.") | |
| with st.expander("Explore Detailed Insights"): | |
| tab1, tab2 = st.tabs(["Planner's Insights", "Analyst's Analysis"]) | |
| with tab1: | |
| planner_output = getattr(results.tasks_output[0], "raw", "No details available.") | |
| st.write(planner_output) | |
| with tab2: | |
| analyst_output = getattr(results.tasks_output[1], "raw", "No details available.") | |
| st.write(analyst_output) | |
| # Convert raw text to structured data | |
| if isinstance(analyst_output, str): | |
| analyst_output = parse_analyst_output(analyst_output) | |
| st.subheader("Structured Analyst Output") | |
| st.write(analyst_output) | |
| charts = [] | |
| if enable_advanced_analysis: | |
| charts = create_visualizations(analyst_output) | |
| table_data = display_table(analyst_output) | |
| st.success(f"Analysis completed in {elapsed_time:.2f} seconds.") | |
| pdf_path = generate_pdf_report(writer_output, charts=charts, table_data=table_data, metadata={"Technology Area": patent_area, "Stakeholder": stakeholder}) | |
| with open(pdf_path, "rb") as report_file: | |
| st.download_button("Download Report", data=report_file, file_name="Patent_Strategy_Report.pdf") | |
| except Exception as e: | |
| logging.error(f"An error occurred during execution: {e}") | |
| st.error(f"An error occurred during execution: {e}") |