Spaces:
Sleeping
Sleeping
File size: 6,170 Bytes
5884015 | 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 | import json
from pathlib import Path
from dotenv import load_dotenv
from openai import OpenAI
from pypdf import PdfReader
from logic.tools import FUNCTION_MAP, TOOLS
load_dotenv(override=True)
BASE_DIR = Path(__file__).resolve().parent.parent
class Me:
def __init__(self):
self.openai = OpenAI()
self.name = "Evison Ndoni"
pdf_path = BASE_DIR / "me" / "evison.pdf"
self.linkedin = ""
try:
if pdf_path.exists():
reader = PdfReader(str(pdf_path))
for page in reader.pages:
text = page.extract_text()
if text:
self.linkedin += text
else:
self.linkedin = "(LinkedIn PDF not found on server.)"
except Exception as e:
self.linkedin = f"(Error reading LinkedIn PDF: {e})"
summary_path = BASE_DIR / "me" / "summary.txt"
try:
with open(summary_path, "r", encoding="utf-8") as f:
self.summary = f.read()
except Exception as e:
self.summary = f"(Summary file not found or unreadable: {e})"
self.extra_context = """
- 26-year-old software engineer from Albania.
- 3+ years experience with React, Next.js, TypeScript, Tailwind CSS and Flutter.
- Currently learning Agentic AI and aiming for AI Engineer roles.
- Values clean, fluid UI/UX and likes to build useful products and SaaS.
- Tries to stay grounded, responsible, and future-oriented while putting God first.
""".strip()
def handle_tool_call(self, tool_calls):
results = []
for tool_call in tool_calls:
tool_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
print(f"Tool called: {tool_name}", flush=True)
tool = FUNCTION_MAP.get(tool_name)
result = tool(**arguments) if tool else {}
results.append(
{
"role": "tool",
"content": json.dumps(result),
"tool_call_id": tool_call.id,
}
)
return results
def system_prompt(self):
system_prompt = (
f"You are acting as {self.name}. You are answering questions on "
f"{self.name}'s personal website, particularly questions related to his "
f"career, background, skills, experience, and values.\n\n"
f"Your responsibility is to represent {self.name} as faithfully as possible. "
f"Be professional, warm, confident, and grounded, as if talking to a "
f"potential client, hiring manager, or collaborator.\n\n"
f"{self.name} is a follower of the Orthodox Christian faith and tries to put "
f"God first while being ambitious and disciplined in his work. When faith "
f"or values come up, you can mention this naturally, but keep the focus on "
f"respectful and professional conversation.\n\n"
f"If you don't know the answer to any question, use your "
f"'record_unknown_question' tool to record the question, even if it's "
f"trivial or unrelated to career.\n\n"
f"If the user seems like a potential employer, client, or collaborator, "
f"gently encourage them to share their email so {self.name} can follow up, "
f"and record it using your 'record_user_details' tool.\n"
)
system_prompt += f"\n## Short Summary\n{self.summary}\n"
system_prompt += f"\n## LinkedIn-style Profile\n{self.linkedin}\n"
system_prompt += f"\n## Additional Personal Context\n{self.extra_context}\n"
system_prompt += (
"\nWith this context, please chat with the user, always staying in "
f"character as {self.name}."
)
return system_prompt
def _run_conversation(self, message, history_messages):
"""
Internal helper that runs the full tool-calling loop and returns
the final assistant message content as a string.
`history_messages` is a list of OpenAI-style dicts: [{role, content}, ...]
(no system message inside; we add it here).
"""
if history_messages is None:
history_messages = []
messages = [
{"role": "system", "content": self.system_prompt()}
] + history_messages + [{"role": "user", "content": message}]
done = False
while not done:
response = self.openai.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=TOOLS,
)
choice = response.choices[0]
if choice.finish_reason == "tool_calls":
message_tool = choice.message
tool_calls = message_tool.tool_calls
results = self.handle_tool_call(tool_calls)
messages.append(message_tool)
messages.extend(results)
else:
done = True
final_message = choice.message
return final_message.content or ""
return ""
def chat_stream(self, message, history_messages):
"""
Generator version for streaming.
Yields *partial assistant content* as a plain string.
Gradio can wrap this to update the Chatbot incrementally.
"""
full_content = self._run_conversation(message, history_messages)
partial = ""
for ch in full_content:
partial += ch
yield partial
def chat(self, message, history_messages):
"""
Non-streaming wrapper kept for compatibility with existing code.
It internally uses `chat_stream` and just returns the final string.
"""
last_chunk = ""
for chunk in self.chat_stream(message, history_messages):
last_chunk = chunk
return last_chunk
|