added LangGraph Based workflow and tools for product extraction, routes classification, calculating carbon footprint and score generator

Agent.py

@@ -0,0 +1,271 @@
+import os
+from dotenv import load_dotenv
+from typing import Annotated, List, Dict
+from typing_extensions import TypedDict
+import requests
+import json
+
+from langgraph.graph.message import add_messages
+# from langchain_google_genai import ChatGoogleGenerativeAI
+from huggingface_hub import InferenceClient
+from langchain.prompts import ChatPromptTemplate
+from langgraph.graph import StateGraph, START, END
+
+# from unity_functions import parse_extraction_result, parse_scoring_result
+# from scrapper import scrape_product_info
+
+# Load environment variables from .env file
+load_dotenv()
+
+# Initialize StateGraph
+class EnvironmentalAnalysisState(TypedDict):
+    messages: Annotated[list, add_messages]
+    product_description: str
+    extracted_data: Dict
+    carbon_footprint: float
+    environmental_score: float
+    recommendations: List[str]
+    analysis_complete: bool
+    route: str
+
+# Initialize llm
+llm = InferenceClient(
+    provider="hyperbolic",
+    api_key=os.getenv("HYPERBOLIC_API_KEY"),  # You'll need to add this to your .env file
+)
+
+
+# Product Information Extraction Tool
+def extract_product_from_url(state: EnvironmentalAnalysisState):
+    """
+    Extract product information from URL and update the product description
+    """
+    extraction_prompt = ChatPromptTemplate.from_template("""
+    You are an expert environmental analyst. You are given a URL to a product page. Your job is to extract environmental impact data by simulating access to the product's specifications and making reasonable assumptions if exact data is missing.
+
+    Product URL: {product_description}
+
+    Carefully extract or infer the following in JSON format:
+    {{
+        "material_composition": "main materials used (e.g., plastic, metal, organic cotton)",
+        "manufacturing_location": "country or region of manufacturing",
+        "product_weight": numeric_value_in_kg,
+        "transport_distance": numeric_value_in_km (assumed from manufacturing to India),
+        "packaging_type": "type and material of packaging (e.g., cardboard box, plastic wrap)",
+        "energy_usage": numeric_value_in_kwh (if applicable or estimated),
+        "recyclability": "recyclable / biodegradable / non-recyclable",
+        "durability": "expected lifespan in years",
+        "certifications": "environmental or sustainability certifications (e.g., Energy Star, Fair Trade)"
+    }}
+
+    If some data is not explicitly available on the page, infer based on similar products or industry standards.
+    Use domain knowledge and mention estimates clearly.
+    """)
+
+    messages = [
+        {
+            "role": "user",
+            "content": extraction_prompt
+        }
+    ]
+
+    result = llm.chat.completions.create(
+        model="deepseek-ai/DeepSeek-R1",  # or another suitable model available on Hyperbolic
+        messages=messages,
+        max_tokens=1000,
+        temperature=0.2
+    )
+    extracted_data = parse_extraction_result(result.choices[0].message.content)
+    return {
+        "extracted_data": extracted_data,
+        "messages": [result.choices[0].message]
+    }
+
+def extract_product_info(state: EnvironmentalAnalysisState):
+    extraction_prompt = ChatPromptTemplate.from_template("""
+    Analyze the following product description and extract environmental impact factors.
+    
+    Product: {product_description}
+    
+    Please provide the information in the following JSON format:
+    {{
+        "material_composition": "main materials used",
+        "manufacturing_location": "country or region",
+        "product_weight": numeric_value_in_kg,
+        "transport_distance": numeric_value_in_km,
+        "packaging_type": "packaging materials and type",
+        "energy_usage": numeric_value_in_kwh,
+        "recyclability": "recyclable/biodegradable/non-recyclable",
+        "durability": "estimated lifespan",
+        "certifications": "environmental certifications if any"
+    }}
+    
+    If specific values are not mentioned, make reasonable estimates based on typical products of this type.
+    """)
+
+    messages = [
+        {
+            "role": "user",
+            "content": extraction_prompt
+        }
+    ]
+
+    result = llm.chat.completions.create(
+        model="deepseek-ai/DeepSeek-R1",
+        messages=messages,
+        max_tokens=1000,
+        temperature=0.2
+    )
+
+    extracted_data = parse_extraction_result(result.choices[0].message.content)
+
+    return {
+        "extracted_data": extracted_data,
+        "messages": [result.choices[0].message]
+    }
+
+def classify_route(state: EnvironmentalAnalysisState):
+    """
+    Classify the Route based on the product description if it starts with a URL or text.
+    """
+    if state["product_description"].startswith("http"):
+        state["route"] = "extract_info_from_url"
+    else:
+        state["route"] = "extract_info_from_description"
+    
+    return state
+
+
+def calculate_carbon_footprint(state: EnvironmentalAnalysisState):
+    climatiq_url = "https://api.climatiq.io/v1/estimate"
+    headers = {
+        "Authorization": f"Bearer {os.getenv('CLIMATIQ_API_KEY')}",
+        "Content-Type": "application/json"
+    }
+
+    extracted_data = state["extracted_data"]
+    total_emissions = 0
+
+    # Calculate emissions from different Activites
+    calculations = []
+
+    if "material_weight" in extracted_data:
+        manufacturing_request = {
+            "emission_factor": {
+                "activity_id": "manufacturing-metals-aluminum"
+            },
+            "parameters": {
+                "weight": extracted_data["material_weight"],
+                "weight_unit": "kg"
+            }
+        }
+        
+        response = requests.post(climatiq_url, headers=headers, 
+                               json=manufacturing_request)
+        if response.status_code == 200:
+            result = response.json()
+            total_emissions += result["co2e"]
+            calculations.append(result)
+    
+    # Transportation emissions calculation
+    if "transport_distance" in extracted_data:
+        transport_request = {
+            "emission_factor": {
+                "activity_id": "freight_flight-route_type_international-distance_na-weight_na"
+            },
+            "parameters": {
+                "distance": extracted_data["transport_distance"],
+                "distance_unit": "km",
+                "weight": extracted_data.get("product_weight", 1),
+                "weight_unit": "kg"
+            }
+        }
+        
+        response = requests.post(climatiq_url, headers=headers, 
+                               json=transport_request)
+        if response.status_code == 200:
+            result = response.json()
+            total_emissions += result["co2e"]
+            calculations.append(result)
+    
+    return {
+        "carbon_footprint": total_emissions,
+        "calculation_details": calculations
+    }
+
+def generate_environmental_score(state: EnvironmentalAnalysisState):
+    scoring_prompt = ChatPromptTemplate.from_template("""
+    Based on the following environmental data, generate a comprehensive environmental score (0-100):
+    
+    Product Data: {extracted_data}
+    Carbon Footprint: {carbon_footprint} kg CO2e
+    
+    Consider these factors:
+    1. Carbon emissions intensity
+    2. Recyclability of materials
+    3. Manufacturing sustainability
+    4. Transportation impact
+    5. Product longevity
+    6. End-of-life disposal
+    
+    Provide:
+    - Overall score (0-100, where 100 is most sustainable)
+    - Category breakdown
+    - Improvement recommendations
+    """)
+    
+    messages = [
+        {
+            "role": "user",
+            "content": scoring_prompt
+        }
+    ]
+    
+    result = llm.chat.completions.create(
+        model="deepseek-ai/DeepSeek-R1",
+        messages=messages,
+        max_tokens=1500,
+        temperature=0.2
+    )
+    
+    # Parse score and recommendations
+    score_data = parse_scoring_result(result.choices[0].message.content)
+    
+    return {
+        "environmental_score": score_data["score"],
+        "recommendations": score_data["recommendations"],
+        "analysis_complete": True,
+        "messages": [result.choices[0].message]
+    }
+
+def route_selector(state: EnvironmentalAnalysisState) -> str:
+    return state["route"]
+
+def create_environmental_analyzer():
+    #initialize the state graph
+    graph_builder = StateGraph(EnvironmentalAnalysisState)
+
+    #nodes
+    graph_builder.add_node("classify_route", classify_route)
+    graph_builder.add_node("extract_info_from_description", extract_product_info)
+    graph_builder.add_node("extract_info_from_url", extract_product_from_url)
+    graph_builder.add_node("calculate_carbon", calculate_carbon_footprint)
+    graph_builder.add_node("generate_score", generate_environmental_score)
+    #edges
+    graph_builder.add_edge(START, "classify_route")
+    # graph_builder.add_edge("classify_route", "extract_info_from_description")
+    # graph_builder.add_edge("classify_route", "extract_info_from_url")
+    graph_builder.add_conditional_edges(
+        "classify",
+        route_selector,
+        {
+            "extract_info_from_description": "extract_info_from_description",
+            "extract_info_from_url": "extract_info_from_url"
+        }
+    )
+    graph_builder.add_edge("extract_info_from_description", "calculate_carbon")
+    graph_builder.add_edge("extract_info_from_url", "calculate_carbon")
+    graph_builder.add_edge("calculate_carbon", "generate_score")
+    graph_builder.add_edge("generate_score", END)
+
+    return graph_builder.compile()

requirements.txt

@@ -0,0 +1,6 @@
+python-dotenv
+langgraph
+langchain
+requests
+bs4
+gradio