Spaces:
Running
Running
| # 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}") |