Spaces:
Running
Running
File size: 4,122 Bytes
8e7f687 |
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 |
# Main chatbot class extracted from app.py
import os
from openai import OpenAI
from content import ContentStore
from notifications.pushover import PushoverService
from tools.definitions import TOOLS
from tools.handler import ToolHandler
from core.router import MessageRouter
from config.prompts import build_system_prompt
class Chatbot:
"""Main chatbot orchestration class"""
def __init__(self, name: str = "Yuelin Liu"):
self.name = name
# Initialize OpenAI client
self.openai = OpenAI(
api_key=os.getenv("GOOGLE_API_KEY"),
base_url="https://generativelanguage.googleapis.com/v1beta/openai/"
)
# Initialize services
self.pushover = PushoverService(
token=os.getenv("PUSHOVER_TOKEN"),
user=os.getenv("PUSHOVER_USER")
)
self.tool_handler = ToolHandler(pushover_service=self.pushover)
self.router = MessageRouter(self.openai)
# Initialize content store
self.content = ContentStore()
# Put career.pdf + summary.txt here (and any other work docs)
self.content.load_folder("me/career", "career")
# Merge everything else (hobby/life/projects/education) into personal/
self.content.load_folder("me/personal", "personal")
# Optional: quick startup log (comment out if noisy)
self._log_loaded_docs()
def build_context_for_mode(self, mode: str) -> str:
"""Build document context for the given mode"""
domain = "career" if mode == "career" else "personal"
return self.content.join_domain_text([domain])
def system_prompt(self, mode: str) -> str:
"""Generate system prompt for the given mode"""
domain_text = self.build_context_for_mode(mode)
return build_system_prompt(self.name, domain_text, mode)
def chat(self, message: str, history: list) -> str:
"""Main chat entrypoint with guarded execution"""
try:
# 1) Route message
route = self.router.classify(message)
intent = route.get("intent", "career")
# Determine mode
if intent == "contact_exchange":
mode = "career" # keep professional context for contact flows
else:
mode = "career" if intent == "career" else "personal"
# 2) Check for immediate responses (boundaries, contact collection, pitch)
immediate_response = self.router.get_response_for_route(message, route, mode)
if immediate_response:
return immediate_response
# 3) Regular chat with tools enabled
messages = [{"role": "system", "content": self.system_prompt(mode)}] \
+ history + [{"role": "user", "content": message}]
while True:
response = self.openai.chat.completions.create(
model="gemini-2.5-flash",
messages=messages,
tools=TOOLS,
temperature=0.2,
top_p=0.9
)
choice = response.choices[0]
if choice.finish_reason == "tool_calls":
results = self.tool_handler.handle_tool_calls(choice.message.tool_calls)
messages.append(choice.message)
messages.extend(results)
continue
return choice.message.content or "Thanks—I've noted that."
except Exception as e:
# Fail-closed, keep UI stable
print(f"[FATAL] Chat turn failed: {e}", flush=True)
return "Oops, something went wrong on my side. Please ask that again—I've reset my context."
def _log_loaded_docs(self):
"""Optional: log loaded documents at startup"""
by_domain = self.content.by_domain
for domain, docs in by_domain.items():
print(f"[LOAD] Domain '{domain}': {len(docs)} document(s)")
for d in docs:
print(f" - {d.title}") |