File size: 8,682 Bytes
ed07e8e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 |
import os
import time
import json
import traceback
from datetime import datetime
import streamlit as st
from dotenv import load_dotenv
from langchain_groq import ChatGroq
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import WebBaseLoader
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
# Initialize environment variables
load_dotenv()
# --------------- Enhanced Logging System ---------------
def log_interaction(user_input: str, bot_response: str, context: str):
"""Log user interactions with context and error handling"""
try:
log_entry = {
"timestamp": datetime.now().isoformat(),
"user_input": user_input,
"bot_response": bot_response,
"context": context,
"model": "llama-3.3-70b-versatile",
"kb_version": st.session_state.kb_info.get('version', '1.0')
}
os.makedirs("chat_history", exist_ok=True)
log_path = os.path.join("chat_history", "chat_logs.json")
# Atomic write operation with UTF-8 encoding
with open(log_path, "a", encoding="utf-8") as f:
f.write(json.dumps(log_entry, ensure_ascii=False) + "\n")
except Exception as e:
error_msg = f"Logging error: {str(e)}\n{traceback.format_exc()}"
print(error_msg)
st.error("Error saving interaction log. Please contact support.")
# --------------- Page Configuration ---------------
st.set_page_config(
page_title="Status Law Assistant",
page_icon="⚖️",
layout="wide",
menu_items={
'About': "### Legal AI Assistant powered by Status.Law"
}
)
# --------------- Knowledge Base Management ---------------
VECTOR_STORE_PATH = "vector_store"
URLS = [
"https://status.law",
"https://status.law/about",
"https://status.law/careers",
"https://status.law/tariffs-for-services-of-protection-against-extradition",
"https://status.law/challenging-sanctions",
"https://status.law/law-firm-contact-legal-protection"
"https://status.law/cross-border-banking-legal-issues",
"https://status.law/extradition-defense",
"https://status.law/international-prosecution-protection",
"https://status.law/interpol-red-notice-removal",
"https://status.law/practice-areas",
"https://status.law/reputation-protection",
"https://status.law/faq"
]
def init_models():
"""Initialize AI models with caching"""
llm = ChatGroq(
model_name="llama-3.3-70b-versatile",
temperature=0.6,
api_key=os.getenv("GROQ_API_KEY")
)
embeddings = HuggingFaceEmbeddings(
model_name="intfloat/multilingual-e5-large-instruct"
)
return llm, embeddings
def build_knowledge_base(embeddings):
"""Create or update the vector knowledge base"""
start_time = time.time()
documents = []
with st.status("Building knowledge base..."):
for url in URLS:
try:
loader = WebBaseLoader(url)
documents.extend(loader.load())
except Exception as e:
st.error(f"Failed to load {url}: {str(e)}")
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=100
)
chunks = text_splitter.split_documents(documents)
vector_store = FAISS.from_documents(chunks, embeddings)
vector_store.save_local(VECTOR_STORE_PATH)
# Update version information
st.session_state.kb_info.update({
'build_time': time.time() - start_time,
'size': sum(os.path.getsize(f) for f in os.listdir(VECTOR_STORE_PATH)) / (1024 ** 2),
'version': datetime.now().strftime("%Y%m%d-%H%M%S")
})
return vector_store
# --------------- Chat Interface ---------------
def main():
llm, embeddings = init_models()
# Initialize or load knowledge base
if not os.path.exists(VECTOR_STORE_PATH):
if st.button("Initialize Knowledge Base"):
with st.spinner("Creating knowledge base..."):
st.session_state.vector_store = build_knowledge_base(embeddings)
st.rerun()
return
if 'vector_store' not in st.session_state:
st.session_state.vector_store = FAISS.load_local(
VECTOR_STORE_PATH, embeddings, allow_dangerous_deserialization=True
)
# Display chat history
if 'messages' not in st.session_state:
st.session_state.messages = []
for msg in st.session_state.messages:
st.chat_message(msg["role"]).write(msg["content"])
# Process user input
if user_input := st.chat_input("Ask your legal question"):
# Display user message
st.chat_message("user").write(user_input)
with st.chat_message("assistant"):
with st.spinner("Analyzing your question..."):
try:
# Retrieve relevant context
context_docs = st.session_state.vector_store.similarity_search(user_input)
context_text = "\n".join(d.page_content for d in context_docs)
# Generate response
prompt_template = PromptTemplate.from_template("""
You are a helpful and polite legal assistant at Status Law.
You answer in the language in which the question was asked.
Answer the question based on the context provided.
If you cannot answer based on the context, say so politely and offer to contact Status Law directly via the following channels:
- For all users: +32465594521 (landline phone).
- For English and Swedish speakers only: +46728495129 (available on WhatsApp, Telegram, Signal, IMO).
- Provide a link to the contact form: [Contact Form](https://status.law/law-firm-contact-legal-protection/).
If the user has questions about specific services and their costs, suggest they visit the page https://status.law/tariffs-for-services-of-protection-against-extradition-and-international-prosecution/ for detailed information.
Ask the user additional questions to understand which service to recommend and provide an estimated cost. For example, clarify their situation and needs to suggest the most appropriate options.
Also, offer free consultations if they are available and suitable for the user's request.
Answer professionally but in a friendly manner.
Example:
Q: How can I challenge the sanctions?
A: To challenge the sanctions, you should consult with our legal team, who specialize in this area. Please contact us directly for detailed advice. You can fill out our contact form here: [Contact Form](https://status.law/law-firm-contact-legal-protection/).
Context: {context}
Question: {question}
Response Guidelines:
1. Answer in the user's language
2. Cite sources when possible
3. Offer contact options if unsure
""")
response = (prompt_template | llm | StrOutputParser()).invoke({
"context": context_text,
"question": user_input
})
# Display and log interaction
st.write(response)
log_interaction(user_input, response, context_text)
st.session_state.messages.extend([
{"role": "user", "content": user_input},
{"role": "assistant", "content": response}
])
except Exception as e:
error_msg = f"Processing error: {str(e)}\n{traceback.format_exc()}"
st.error("Error processing request. Please try again.")
print(error_msg)
log_interaction(user_input, "SYSTEM_ERROR", context_text)
if __name__ == "__main__":
main() |