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("HF_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 """ if state.get("product_description"): scraped_info = scrape_product_info(state["product_description"]) extraction_prompt = ChatPromptTemplate.from_template(""" Analyze the following scraped product information and extract environmental impact factors. Product: {scraped_info} 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. """) rendered_prompt_content = extraction_prompt.format(scraped_info=scraped_info) messages = [ { "role": "user", "content": rendered_prompt_content } ] 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 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. """) rendered_prompt_content = extraction_prompt.format(product_description=state["product_description"]) messages = [ { "role": "user", "content": rendered_prompt_content } ] 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). Aim to score products fairly, recognizing positive efforts and current industry sustainability trends. Be **generous where any notable sustainable practices are observed**, reflecting a positive outlook on efforts towards environmental responsibility. Focus on a **broader assessment rather than scrutinizing every minor detail**, aiming for an overall, balanced sustainability score. Product Data: {extracted_data} Carbon Footprint: {carbon_footprint} kg CO2e Scoring Guide: - 90-100 (Excellent/Leader): Demonstrates exceptional environmental stewardship, innovative green practices, and minimal negative impact across all factors. A true leader in sustainability. - 70-89 (Good/Above Average): Shows strong environmental practices with notable positive attributes and efforts to reduce impact. Generally a good choice for sustainability. - 40-69 (Average/Acceptable): Meets basic environmental considerations but has significant areas for improvement. Represents typical industry practices without outstanding sustainable features. - 0-39 (Poor/Concerning): Significant environmental concerns, unsustainable practices, or a clear lack of attention to environmental impact. Examples to guide scoring perspective: - Product: "Electronic smartphone with aluminum body, manufactured in China, typical global distribution for sales in Europe." Typical Score Range: 50-65 (Reasonable for modern electronics with some recyclable components and efforts, but significant global transport, energy consumption during use, and complex end-of-life disposal.) - Product: "Local handmade wooden furniture using sustainably harvested, certified wood, finished with non-toxic oil, sold within 50km of production." Typical Score Range: 85-95 (Excellent material choice, minimal transport, durable product designed for longevity, non-toxic finish, often supports local economy.) - Product: "Single-use disposable coffee cup made from virgin paper with a plastic lining, widely distributed via international shipping." Typical Score Range: 20-35 (Poor due to single-use nature, material composite making recycling difficult, high transport, and significant waste contribution.) 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 Here are some examples of products and their potential scores to guide your assessment, aiming for higher scores where appropriate: - "Electronic smartphone with aluminum body, manufactured in China": Likely score in the range of 50-65. While it has some recyclable materials, manufacturing and transportation in a global supply chain contribute to a moderate footprint. - "Local handmade wooden furniture using sustainable forest wood": Likely score in the range of 85-95. This product benefits from local production, renewable resources, and often a longer lifespan. - "Single-use plastic water bottle produced with virgin plastics": Likely score in the range of 10-25. High environmental impact due to non-renewable materials, single-use nature, and pollution potential. Provide: - Overall score (0-100, where 100 is most sustainable and 0 is least sustainable) - Improvement recommendations """) rendered_prompt_content = scoring_prompt.format( extracted_data=state["extracted_data"], carbon_footprint=state["carbon_footprint"] ) messages = [ { "role": "user", "content": rendered_prompt_content } ] result = llm.chat.completions.create( model="meta-llama/Meta-Llama-3.1-70B-Instruct", 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", 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()