Asset trace Bot Views.py
Tue Aug 27 2024 04:40:13 GMT+0000 (Coordinated Universal Time)
Saved by @vineethnj
#views.py from django.shortcuts import render, redirect, get_object_or_404 from .models import AssetData, AssetType,AssetImage,Asset_movement_data from .forms import AssetDataForm, AssetTypeForm,AssetMovementForm from django.urls import reverse from .utils import generate_qr_code from authentication.models import Branches from django.http import JsonResponse from django.contrib.auth.decorators import login_required def asset_list(request): user = request.user if user.is_superuser: assets = AssetData.objects.all() asset_type = AssetType.objects.all() elif user.is_superadmin and user.head_office: branches = Branches.objects.filter(head_office=user.head_office) assets = AssetData.objects.filter(branch__in=branches) asset_type = AssetType.objects.filter(assets__branch__in=branches).distinct() elif user.is_admin and user.branch: assets = AssetData.objects.filter(branch=user.branch) asset_type = AssetType.objects.filter(assets__branch=user.branch).distinct() else: assets = AssetData.objects.none() asset_type = AssetType.objects.none() print(assets) asset_data=[] for asset in assets: vessel_url = request.build_absolute_uri(reverse('asset_link', args=[asset.id])) qr_code_url = generate_qr_code(asset.asset_name, vessel_url) asset_data.append({'asset': asset, 'qr_code_url': qr_code_url}) return render(request, 'asset/asset_list.html', {'assets': asset_data,'asset_type':asset_type}) def asset_create(request): if request.method == 'POST': if 'asset_type_submit' in request.POST: return handle_asset_type_submit(request) elif 'asset_submit' in request.POST: return handle_asset_submit(request) else: asset_type_form = AssetTypeForm(prefix='asset_type') asset_form = AssetDataForm(prefix='asset') return render(request, 'asset/asset_form.html', { 'asset_type_form': asset_type_form, 'asset_form': asset_form, }) def handle_asset_type_submit(request): asset_type_form = AssetTypeForm(request.POST, prefix='asset_type') asset_form = AssetDataForm(prefix='asset') if asset_type_form.is_valid(): asset_type_form.save() return redirect('asset_create') return render(request, 'asset/asset_form.html', { 'asset_type_form': asset_type_form, 'asset_form': asset_form, }) def handle_asset_submit(request): asset_form = AssetDataForm(request.POST, request.FILES, prefix='asset') print(f"POST data: {request.POST}") print(f"Form errors: {asset_form.errors}") asset_type_form = AssetTypeForm(prefix='asset_type') if asset_form.is_valid(): asset = asset_form.save(commit=False) asset.save() for image in asset_form.cleaned_data['images']: AssetImage.objects.create(asset=asset, image=image) asset_url = request.build_absolute_uri(reverse('asset_link', args=[asset.id])) asset.qr_code_base64 = generate_qr_code(asset.asset_name, asset_url) asset.save() return redirect('asset_list') return render(request, 'asset/asset_form.html', { 'asset_type_form': asset_type_form, 'asset_form': asset_form, }) def get_branches(request): office_id = request.GET.get('office_id') print(office_id) branches = Branches.objects.filter(head_office_id=office_id).values('id', 'branch_name') print(branches) return JsonResponse(list(branches), safe=False) def asset_detail(request, pk): asset = get_object_or_404(AssetData, pk=pk) asset_url = request.build_absolute_uri(reverse('asset_link', args=[asset.id])) qr_code_base64 = generate_qr_code(asset.asset_name, asset_url) return render(request, 'asset/asset_detail.html', {'asset': asset,'qr_code_base64': qr_code_base64,}) def asset_type_edit(request, pk): asset = get_object_or_404(AssetType, pk=pk) if request.method == 'POST': asset_type_form = AssetTypeForm(request.POST, instance=asset, prefix='asset_type') if 'asset_type_submit' in request.POST and asset_type_form.is_valid(): asset_type_form.save() return redirect('asset_list') else: asset_type_form = AssetTypeForm(instance=asset, prefix='asset_type') return render(request, 'asset/edit_assettype.html', {'asset_type_form': asset_type_form}) def asset_edit(request, pk): asset = get_object_or_404(AssetData, pk=pk) if request.method == 'POST': asset_form = AssetDataForm(request.POST, request.FILES, instance=asset, prefix='asset') if 'asset_submit' in request.POST and asset_form.is_valid(): asset = asset_form.save(commit=False) asset_url = request.build_absolute_uri(reverse('asset_link', args=[asset.id])) asset.qr_code_base64 = generate_qr_code(asset.asset_name, asset_url) asset.save() # Clear existing images if necessary asset.images.all().delete() # Handle image uploads for image in asset_form.cleaned_data['images']: AssetImage.objects.create(asset=asset, image=image) return redirect('asset_list') else: asset_form = AssetDataForm(instance=asset, prefix='asset') return render(request, 'asset/asset_edit.html', { 'asset_form': asset_form, }) def asset_delete(request, pk): asset = get_object_or_404(AssetData, pk=pk) asset.delete() return redirect('asset_list') def asset_link(request, asset_id): asset = get_object_or_404(AssetData, id=asset_id) main_details = [ {'key': 'Name', 'value': asset.asset_name}, {'key': 'Asset Type', 'value': asset.asset_type}, {'key': 'Head Office', 'value': asset.office}, {'key': 'Branch', 'value': asset.branch.branch_name}, {'key': 'Model', 'value': asset.model}, {'key': 'Purchase Date', 'value': asset.purchase_date}, {'key': 'Warranty Info', 'value': asset.warranty_info}, ] images = asset.images.all() context = { 'asset': asset, 'main_details': main_details, 'images': images, } return render(request, 'asset/asset_link.html', context) from django.http import HttpResponse from django.conf import settings def download_qr_code(request, vessel_id): vessel = get_object_or_404(AssetData, id=vessel_id) qr_code_path = f'{settings.MEDIA_ROOT}/qr_codes/{vessel.name}.png' try: with open(qr_code_path, 'rb') as f: response = HttpResponse(f.read(), content_type='image/png') response['Content-Disposition'] = f'attachment; filename="{vessel.name}_qr.png"' return response except FileNotFoundError: return HttpResponse("QR code not found", status=404) from django.db.models import Q def asset_movement_page(request): if request.method == 'POST': form = AssetMovementForm(request.POST) if form.is_valid(): form.save() return redirect('asset_movement_list') else: print(form.errors) else: form = AssetMovementForm() # movements = Asset_movement_data.objects.all().order_by('-movement_date') user = request.user if user.is_superuser: movements = Asset_movement_data.objects.all() elif user.is_superadmin and user.head_office: branches = Branches.objects.filter(head_office=user.head_office) movements = Asset_movement_data.objects.filter( Q(from_branch__in=branches) | Q(to_branch__in=branches) ) elif user.is_admin and user.branch: movements = Asset_movement_data.objects.filter( Q(from_branch=user.branch) | Q(to_branch=user.branch) ) else: movements = Asset_movement_data.objects.none() return render(request, 'asset/asset_movement_list.html', {'form': form, 'movements': movements}) def edit_asset_movement(request, movement_id): movement = get_object_or_404(Asset_movement_data, id=movement_id) if request.method == 'POST': form = AssetMovementForm(request.POST, instance=movement) if form.is_valid(): form.save() return redirect('asset_movement_list') else: form = AssetMovementForm(instance=movement) return render(request, 'asset/edit_asset_movement.html', {'form': form, 'movement': movement}) def asset_movement_success(request): return render(request, 'asset/asset_movement_success.html') def asset_filter_by_branches(request, branch_id): branch = Branches.objects.get(id=branch_id) # Assets currently in the branch, excluding those that have been moved moved_out_assets = Asset_movement_data.objects.filter(from_branch=branch).values_list('asset_id', flat=True) assets = AssetData.objects.filter(branch=branch).exclude(id__in=moved_out_assets) # Assets moved from the branch moved_assets = Asset_movement_data.objects.filter(from_branch=branch).order_by('-movement_date') # Assets received at the branch received_assets = Asset_movement_data.objects.filter(to_branch=branch).order_by('-movement_date') context = { 'branch': branch, 'assets': assets, 'moved_assets': moved_assets, 'received_assets': received_assets, } return render(request, 'asset/assets_by_branch.html', context) @login_required def asset_movement_list(request): user = request.user if user.is_superuser: movements = Asset_movement_data.objects.all() elif user.is_superadmin and user.head_office: branches = Branches.objects.filter(head_office=user.head_office) movements = Asset_movement_data.objects.filter( Q(from_branch__in=branches) | Q(to_branch__in=branches) ) else: movements = Asset_movement_data.objects.none() form = AssetMovementForm() # Assuming you have this form return render(request, 'authentication/asset_movement_list.html', { 'movements': movements, 'form': form }) #views #views.py from django.shortcuts import render, redirect, get_object_or_404 from .models import AssetData, AssetType,AssetImage,Asset_movement_data from .forms import AssetDataForm, AssetTypeForm,AssetMovementForm from django.urls import reverse from .utils import generate_qr_code from authentication.models import Branches from django.http import JsonResponse from django.contrib.auth.decorators import login_required def asset_list(request): user = request.user if user.is_superuser: assets = AssetData.objects.all() asset_type = AssetType.objects.all() elif user.is_superadmin and user.head_office: branches = Branches.objects.filter(head_office=user.head_office) assets = AssetData.objects.filter(branch__in=branches) asset_type = AssetType.objects.filter(assets__branch__in=branches).distinct() elif user.is_admin and user.branch: assets = AssetData.objects.filter(branch=user.branch) asset_type = AssetType.objects.filter(assets__branch=user.branch).distinct() else: assets = AssetData.objects.none() asset_type = AssetType.objects.none() print(assets) asset_data=[] for asset in assets: vessel_url = request.build_absolute_uri(reverse('asset_link', args=[asset.id])) qr_code_url = generate_qr_code(asset.asset_name, vessel_url) asset_data.append({'asset': asset, 'qr_code_url': qr_code_url}) return render(request, 'asset/asset_list.html', {'assets': asset_data,'asset_type':asset_type}) def asset_create(request): if request.method == 'POST': if 'asset_type_submit' in request.POST: return handle_asset_type_submit(request) elif 'asset_submit' in request.POST: return handle_asset_submit(request) else: asset_type_form = AssetTypeForm(prefix='asset_type') asset_form = AssetDataForm(prefix='asset') return render(request, 'asset/asset_form.html', { 'asset_type_form': asset_type_form, 'asset_form': asset_form, }) def handle_asset_type_submit(request): asset_type_form = AssetTypeForm(request.POST, prefix='asset_type') asset_form = AssetDataForm(prefix='asset') if asset_type_form.is_valid(): asset_type_form.save() return redirect('asset_create') return render(request, 'asset/asset_form.html', { 'asset_type_form': asset_type_form, 'asset_form': asset_form, }) def handle_asset_submit(request): asset_form = AssetDataForm(request.POST, request.FILES, prefix='asset') print(f"POST data: {request.POST}") print(f"Form errors: {asset_form.errors}") asset_type_form = AssetTypeForm(prefix='asset_type') if asset_form.is_valid(): asset = asset_form.save(commit=False) asset.save() for image in asset_form.cleaned_data['images']: AssetImage.objects.create(asset=asset, image=image) asset_url = request.build_absolute_uri(reverse('asset_link', args=[asset.id])) asset.qr_code_base64 = generate_qr_code(asset.asset_name, asset_url) asset.save() return redirect('asset_list') return render(request, 'asset/asset_form.html', { 'asset_type_form': asset_type_form, 'asset_form': asset_form, }) def get_branches(request): office_id = request.GET.get('office_id') print(office_id) branches = Branches.objects.filter(head_office_id=office_id).values('id', 'branch_name') print(branches) return JsonResponse(list(branches), safe=False) def asset_detail(request, pk): asset = get_object_or_404(AssetData, pk=pk) asset_url = request.build_absolute_uri(reverse('asset_link', args=[asset.id])) qr_code_base64 = generate_qr_code(asset.asset_name, asset_url) return render(request, 'asset/asset_detail.html', {'asset': asset,'qr_code_base64': qr_code_base64,}) def asset_type_edit(request, pk): asset = get_object_or_404(AssetType, pk=pk) if request.method == 'POST': asset_type_form = AssetTypeForm(request.POST, instance=asset, prefix='asset_type') if 'asset_type_submit' in request.POST and asset_type_form.is_valid(): asset_type_form.save() return redirect('asset_list') else: asset_type_form = AssetTypeForm(instance=asset, prefix='asset_type') return render(request, 'asset/edit_assettype.html', {'asset_type_form': asset_type_form}) def asset_edit(request, pk): asset = get_object_or_404(AssetData, pk=pk) if request.method == 'POST': asset_form = AssetDataForm(request.POST, request.FILES, instance=asset, prefix='asset') if 'asset_submit' in request.POST and asset_form.is_valid(): asset = asset_form.save(commit=False) asset_url = request.build_absolute_uri(reverse('asset_link', args=[asset.id])) asset.qr_code_base64 = generate_qr_code(asset.asset_name, asset_url) asset.save() # Clear existing images if necessary asset.images.all().delete() # Handle image uploads for image in asset_form.cleaned_data['images']: AssetImage.objects.create(asset=asset, image=image) return redirect('asset_list') else: asset_form = AssetDataForm(instance=asset, prefix='asset') return render(request, 'asset/asset_edit.html', { 'asset_form': asset_form, }) def asset_delete(request, pk): asset = get_object_or_404(AssetData, pk=pk) asset.delete() return redirect('asset_list') def asset_link(request, asset_id): asset = get_object_or_404(AssetData, id=asset_id) main_details = [ {'key': 'Name', 'value': asset.asset_name}, {'key': 'Asset Type', 'value': asset.asset_type}, {'key': 'Head Office', 'value': asset.office}, {'key': 'Branch', 'value': asset.branch.branch_name}, {'key': 'Model', 'value': asset.model}, {'key': 'Purchase Date', 'value': asset.purchase_date}, {'key': 'Warranty Info', 'value': asset.warranty_info}, ] images = asset.images.all() context = { 'asset': asset, 'main_details': main_details, 'images': images, } return render(request, 'asset/asset_link.html', context) from django.http import HttpResponse from django.conf import settings def download_qr_code(request, vessel_id): vessel = get_object_or_404(AssetData, id=vessel_id) qr_code_path = f'{settings.MEDIA_ROOT}/qr_codes/{vessel.name}.png' try: with open(qr_code_path, 'rb') as f: response = HttpResponse(f.read(), content_type='image/png') response['Content-Disposition'] = f'attachment; filename="{vessel.name}_qr.png"' return response except FileNotFoundError: return HttpResponse("QR code not found", status=404) from django.db.models import Q def asset_movement_page(request): if request.method == 'POST': form = AssetMovementForm(request.POST) if form.is_valid(): form.save() return redirect('asset_movement_list') else: print(form.errors) else: form = AssetMovementForm() # movements = Asset_movement_data.objects.all().order_by('-movement_date') user = request.user if user.is_superuser: movements = Asset_movement_data.objects.all() elif user.is_superadmin and user.head_office: branches = Branches.objects.filter(head_office=user.head_office) movements = Asset_movement_data.objects.filter( Q(from_branch__in=branches) | Q(to_branch__in=branches) ) elif user.is_admin and user.branch: movements = Asset_movement_data.objects.filter( Q(from_branch=user.branch) | Q(to_branch=user.branch) ) else: movements = Asset_movement_data.objects.none() return render(request, 'asset/asset_movement_list.html', {'form': form, 'movements': movements}) def edit_asset_movement(request, movement_id): movement = get_object_or_404(Asset_movement_data, id=movement_id) if request.method == 'POST': form = AssetMovementForm(request.POST, instance=movement) if form.is_valid(): form.save() return redirect('asset_movement_list') else: form = AssetMovementForm(instance=movement) return render(request, 'asset/edit_asset_movement.html', {'form': form, 'movement': movement}) def asset_movement_success(request): return render(request, 'asset/asset_movement_success.html') def asset_filter_by_branches(request, branch_id): branch = Branches.objects.get(id=branch_id) # Assets currently in the branch, excluding those that have been moved moved_out_assets = Asset_movement_data.objects.filter(from_branch=branch).values_list('asset_id', flat=True) assets = AssetData.objects.filter(branch=branch).exclude(id__in=moved_out_assets) # Assets moved from the branch moved_assets = Asset_movement_data.objects.filter(from_branch=branch).order_by('-movement_date') # Assets received at the branch received_assets = Asset_movement_data.objects.filter(to_branch=branch).order_by('-movement_date') context = { 'branch': branch, 'assets': assets, 'moved_assets': moved_assets, 'received_assets': received_assets, } return render(request, 'asset/assets_by_branch.html', context) @login_required def asset_movement_list(request): user = request.user if user.is_superuser: movements = Asset_movement_data.objects.all() elif user.is_superadmin and user.head_office: branches = Branches.objects.filter(head_office=user.head_office) movements = Asset_movement_data.objects.filter( Q(from_branch__in=branches) | Q(to_branch__in=branches) ) else: movements = Asset_movement_data.objects.none() form = AssetMovementForm() # Assuming you have this form return render(request, 'authentication/asset_movement_list.html', { 'movements': movements, 'form': form }) #views import os import sys import re import json import logging from pathlib import Path from dotenv import load_dotenv from django.http import JsonResponse from django.views.decorators.http import require_http_methods from django.views.decorators.csrf import csrf_exempt from django.db import connection from langchain_community.agent_toolkits import create_sql_agent from langchain_community.utilities import SQLDatabase from langchain_groq import ChatGroq from langchain.tools import BaseTool from langchain.chains import LLMChain from langchain.agents import ZeroShotAgent, Tool, AgentExecutor from langchain.schema import HumanMessage, SystemMessage from langchain_openai import ChatOpenAI from .models import AssetData from langchain.agents import load_tools, initialize_agent, AgentType # Setup Django environment project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.append(project_root) os.environ.setdefault("DJANGO_SETTINGS_MODULE", "asset_management_system.settings") import django django.setup() # Setup logging logger = logging.getLogger(__name__) # Load environment variables and set up API keys load_dotenv() api_key = os.getenv('OPENAI_API_KEY') groq_api_key = os.getenv("GROQ_API_KEY") os.environ["GROQ_API_KEY"] = groq_api_key # Initialize LLM llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini", temperature=0) # Constants WEBAPP_INFO = """ The Asset Management System features a login system with three roles: Master Admin, Super Admin, and Admin, each with specific permissions. The system supports various asset types, with QR codes generated for each asset for easy tracking and management. Assets can be transferred between branches, with movement history tracked for audits. """ offense_words = ["Ninde ama, pati!", "Aaana kunna", "aan-vedi", "achante andi", "Achinga Kunnan", "adichu poli", "Ajoli ka Thajoli", "ammaudi pootil poocha", "Ammente Pooru", "amminhnha", "Andi pidiyan", "anna kunna", "Appi", "appikunna", "Arjun shambu", "arraykku chuttum poorruu ullaval", "avaraathi", "avarathi mone", "Beejam", "chalam nakki", "Chandi", "Chathavante andi oombiya", "Cheekoothi mon", "Cheekoothi mone", "Chokka lingam", "CHotta kunnan", "Chotta Tha", "coondi", "Da patti", "Dushtan", "eli kunna", "Ettukali Patti Pooran", "inchi mola", "johninte andi", "kaatu poori", "kallel oouuki", "kandara oli", "kandi", "kandi theeni", "Kandu", "Kanni", "Kanni mola", "kara vedi", "karim kunna", "karim pooran", "katta thayoli", "kazhappu perutha mairan", "Keepu", "Kettal", "kodam nakiii", "Kolekkeri", "Koothara", "Koothi", "Koothichi", "kotham", "kotham kalakki", "kotham nakku", "Koyimani", "kuch", "kulamavuka", "kundan", "kundaroli poori mone", "Kundi", "Kundi mon", "Kundi oota", "kundimol", "kunji kunnan", "kunna", "Kunna", "kunna chappu", "Kunna Oli", "Kunna paal", "Kunna thayoli", "kunna urunji", "kunnappal", "kushukk", "Lick Me", "malayalam", "Mandu", "Maratel ooki", "Masa", "Mattanga Poore", "matti", "Mayiradi mon", "Mlechan", "mola", "moonchikko", "Mula Adi", "mula chappu", "mula kashakku", "mula mol", "Mundachi", "Myir", "Myre", "Myrru", "Naayi meedan", "naayinte mone", "nayinte mone", "Nayinte Mone", "Nayinte Monne", "ni urru myre", "ninde andi maire", "Ninte ama koondi ishtima", "Ninte ammakku vettu nayinte mone", "Ninte ammaku vetu", "ninte ammeda tar", "Ninte Ammede Kothil.", "ninte ammede pooru", "Ninte ammede thenga mairu", "ninte appante andi", "Odu myre", "ookki", "oomban", "oombi mon", "Oooomb", "Ootan", "paara pooru", "paareel ookki", "Pacha tayolli", "Pachila Pooran", "Paik keriko myra", "paiku vetti", "pala thanthakkundaya thaoli", "pallinedayil kanthu", "pambara kutichi", "pampara thayoli", "panchavarna kunna"] # Create SQLDatabase instance db_engine = connection.settings_dict['ENGINE'] db_name = connection.settings_dict['NAME'] if isinstance(db_name, (str, Path)): db_name = str(db_name) else: raise ValueError(f"Unexpected type for db_name: {type(db_name)}") db_url = f"sqlite:///{db_name}" if 'sqlite' in db_engine else f"{db_engine}://{db_name}" db = SQLDatabase.from_uri(db_url) # Create SQL agent sql_agent = create_sql_agent(llm, db=db, agent_type="openai-tools", verbose=True) class GreetingTool(BaseTool): name = "greeting_tool" description = "Use this to respond to greetings or simple queries that don't require complex reasoning." def _run(self, query: str) -> str: greetings = ["hello", "hi", "hey", "greetings", "good morning", "good afternoon", "good evening"] if any(greeting in query.lower() for greeting in greetings): return "Hello! I'm the Asset Trace AI assistant. How can I help you with asset management today?" return "I'm here to help with asset management. What specific information or assistance do you need?" async def _arun(self, query: str): raise NotImplementedError("This tool does not support async") # Tool classes class QueryValidationTool(BaseTool): name = "query_validation_tool" description = f"""Evaluate the given query to ensure it is appropriate, respectful, and suitable for general audiences. Check for offensive language, hate speech, explicit content, harassment, sensitive topics, and respectfulness.""" def _run(self, query: str) -> str: return "VALID" if len(query) > 5 and all(word.lower() not in query.lower() for word in offense_words) else "INVALID" async def _arun(self, query: str): raise NotImplementedError("This tool does not support async") class CybersecurityTool(BaseTool): name = "cybersecurity_tool" description = """Checks if the query contains any potentially malicious content, including injection attacks, XSS, malware links, obfuscation techniques, and known vulnerabilities.""" def _run(self, query: str) -> str: return "SAFE" if all(word.lower() not in query.lower() for word in ["hack", "exploit", "vulnerability", "attack", "'; DROP TABLE", "<script>"]) else "UNSAFE" async def _arun(self, query: str): raise NotImplementedError("This tool does not support async") class AssetAdditionTool(BaseTool): name = "asset_addition_tool" description = "Provides instructions on how to add an asset to the Asset Management System." def _run(self, query: str) -> str: return """ To add an asset to the Asset Management System: 1. Log in to the system. 2. Navigate to 'Assets' section. 3. Click 'Add New Asset'. 4. Fill in asset details. 5. Click 'Save'. 6. A QR code will be generated for tracking. Contact support for further assistance. """ async def _arun(self, query: str): raise NotImplementedError("This tool does not support async") # Create instances of specialized tools query_validation_tool = QueryValidationTool() cybersecurity_tool = CybersecurityTool() asset_addition_tool = AssetAdditionTool() greeting_tool = GreetingTool() # Define the tools tools = [ Tool(name="Greeting", func=greeting_tool.run, description="Use this to handle greetings and simple queries"), Tool(name="Query Validation", func=query_validation_tool.run, description="Use this to validate user queries"), Tool(name="Cybersecurity Check", func=cybersecurity_tool.run, description="Use this to perform security checks on user queries"), Tool(name="Asset Addition", func=asset_addition_tool.run, description="Use this to get instructions on adding an asset"), Tool(name="Database Query", func=sql_agent.run, description="Use this to query the database for asset information and should not give the table details") ] # Define the agent prefix = """You are an AI assistant for the Asset Trace application. Your role is to provide accurate and helpful information about the application, assist users with queries, and guide them in using the app effectively. Application Details: {WEBAPP_INFO} Available Tools: [Greeting, Query Validation, Cybersecurity Check, Asset Addition, Database Query] Key Instructions: - For simple greetings or queries, use the Greeting tool directly without further reasoning. - For more complex queries related to asset management, use the appropriate tools. - Respond in the same language as the user's query. - Provide concise answers for simple questions, offering to elaborate if needed. - Maintain a professional and friendly tone. - If you're unsure about any information, state that clearly and suggest where the user might find accurate details. - For complex queries, break down your response into clear, manageable steps. - Respect user privacy and data security at all times. Please respond to user queries based on these guidelines and the information provided about the Asset Trace application. """ suffix = """Begin! Remember to focus on the current query and not be overly influenced by past interactions""" prompt = ZeroShotAgent.create_prompt( tools, prefix=prefix, suffix=suffix, input_variables=["input", "agent_scratchpad", "WEBAPP_INFO"] ) llm_chain = LLMChain(llm=llm, prompt=prompt) tool_names = [tool.name for tool in tools] agent = ZeroShotAgent(llm_chain=llm_chain, allowed_tools=tool_names) # Create the agent with a maximum number of iterations agent = initialize_agent( tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True, max_iterations=3, # Limit the number of iterations early_stopping_method="generate", agent_kwargs={ "prefix": prefix, "format_instructions": """Use the following format: Question: the input question you must answer Thought: you should always think about what to do Action: the action to take, should be one of [{tool_names}] Action Input: the input to the action Observation: the result of the action ... (this Thought/Action/Action Input/Observation can repeat N times) Thought: I now know the final answer Final Answer: the final answer to the original input question""" } ) def process_query(query: str) -> str: try: response = agent.run(input=query, WEBAPP_INFO=WEBAPP_INFO) return response except Exception as e: logger.error(f"Error processing query: {str(e)}", exc_info=True) return "I apologize, but I encountered an issue while processing your request. Could you please rephrase your question or try again later?" @csrf_exempt @require_http_methods(["POST"]) def chatbot_response(request): logger.info("Chatbot response view called") try: data = json.loads(request.body) user_message = data.get('message', '').strip() conversation_history = data.get('conversation_history', []) logger.info(f"Received message: {user_message}") system_message = f""" You are an AI assistant for an asset management system. Here's the current context: 1. User's latest message: "{user_message}" 2. System overview: {WEBAPP_INFO} Respond naturally and conversationally. If the user asks about an asset or the system's capabilities, provide relevant information. If they ask about something we don't have info on, politely explain that and offer to help with something else. Always stay focused on asset management and the system's capabilities. """ messages = [SystemMessage(content=system_message)] for msg in conversation_history[-5:]: messages.append(HumanMessage(content=msg['user_message'])) messages.append(SystemMessage(content=msg['ai_response'])) messages.append(HumanMessage(content=user_message)) response = process_query(user_message) conversation_history.append({ 'user_message': user_message, 'ai_response': response }) return JsonResponse({ 'response': response, 'conversation_history': conversation_history }) except json.JSONDecodeError: logger.error("Invalid JSON received") return JsonResponse({'error': 'We encountered an issue processing your request. Please try again.'}, status=400) except Exception as e: logger.error(f"An error occurred: {str(e)}", exc_info=True) return JsonResponse({'error': 'We apologize, but we experienced a temporary issue. Please try again later.'}, status=500) #image_views from django.views.decorators.csrf import csrf_exempt from django.http import JsonResponse from django.core.files.storage import default_storage from django.core.files.base import ContentFile from django.conf import settings import os import torch from PIL import Image from transformers import CLIPProcessor, CLIPModel from .models import AssetData, AssetImage # Load the CLIP model and processor model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32") processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32") def extract_features(image_path): image = Image.open(image_path) inputs = processor(images=image, return_tensors="pt", padding=True) with torch.no_grad(): image_features = model.get_image_features(**inputs) return image_features @csrf_exempt def chatbot_image_response(request): if request.method == 'POST': try: # Get the uploaded image from the request image_file = request.FILES.get('image') if not image_file: return JsonResponse({'error': 'No image file provided'}, status=400) # Save the image temporarily temp_path = default_storage.save('temp_image.jpg', ContentFile(image_file.read())) full_temp_path = os.path.join(settings.MEDIA_ROOT, temp_path) # Extract features from the uploaded image uploaded_features = extract_features(full_temp_path) # Find the top 3 matching images in the database matches = [] for asset_image in AssetImage.objects.all(): db_image_path = os.path.join(settings.MEDIA_ROOT, str(asset_image.image)) db_features = extract_features(db_image_path) similarity = torch.nn.functional.cosine_similarity(uploaded_features, db_features).item() matches.append((asset_image, similarity)) # Sort matches by similarity (descending) and get top 3 top_matches = sorted(matches, key=lambda x: x[1], reverse=True)[:3] if top_matches and top_matches[0][1] > 0.8: response = "I've identified the top 3 assets that match the uploaded image. Here's the information:\n\n" for i, (asset_image, similarity) in enumerate(top_matches, 1): asset = asset_image.asset image_url = request.build_absolute_uri(asset_image.image.url) response += f"{i}. Match {i} (Similarity: {similarity:.2f})\n" response += f" • Asset Name: {asset.asset_name}\n" response += f" • Asset Type: {asset.asset_type}\n" response += f" • Model: {asset.model}\n" response += f" • Serial Number: {asset.serial_number}\n" response += f" • Purchase Date: {asset.purchase_date}\n" response += f" • Warranty Info: {asset.warranty_info}\n" response += f" • Image: [View Image]({image_url})\n\n" # Include the uploaded image URL in the response uploaded_image_url = request.build_absolute_uri(settings.MEDIA_URL + temp_path) response += f"Uploaded Image: [View Uploaded Image]({uploaded_image_url})\n" else: response = "I couldn't find any assets that closely match the uploaded image. Please make sure the image is clear and matches an asset in our database." # Clean up the temporary file os.remove(full_temp_path) return JsonResponse({'response': response}) except Exception as e: return JsonResponse({'error': str(e)}, status=500) return JsonResponse({'error': 'Invalid request method'}, status=405)
Comments