my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	import argparse
#2	import logging
#3	import os
#4	from typing import Optional
#5
#6	from embedchain.helpers.json_serializable import register_deserializable
#7
#8	from .base import BaseBot
#9
#10	try:
#11	from fastapi_poe import PoeBot, run
#12	except ModuleNotFoundError:
#13	raise ModuleNotFoundError(
#14	"The required dependencies for Poe are not installed." "Please install with `pip install fastapi-poe==0.0.16`"
#15	) from None
#16
#17
#18	def start_command():
#19	parser = argparse.ArgumentParser(description="EmbedChain PoeBot command line interface")
#20	# parser.add_argument("--host", default="0.0.0.0", help="Host IP to bind")
#21	parser.add_argument("--port", default=8080, type=int, help="Port to bind")
#22	parser.add_argument("--api-key", type=str, help="Poe API key")
#23	# parser.add_argument(
#24	# "--history-length",
#25	# default=5,
#26	# type=int,
#27	# help="Set the max size of the chat history. Multiplies cost, but improves conversation awareness.",
#28	# )
#29	args = parser.parse_args()
#30
#31	# FIXME: Arguments are automatically loaded by Poebot's ArgumentParser which causes it to fail.
#32	# the port argument here is also just for show, it actually works because poe has the same argument.
#33
#34	run(PoeBot(), api_key=args.api_key or os.environ.get("POE_API_KEY"))
#35
#36
#37	@register_deserializable
#38	class PoeBot(BaseBot, PoeBot):
#39	def __init__(self):
#40	self.history_length = 5
#41	super().__init__()
#42
#43	async def get_response(self, query):
#44	last_message = query.query[-1].content
#45	try:
#46	history = (
#47	[f"{m.role}: {m.content}" for m in query.query[-(self.history_length + 1) : -1]]
#48	if len(query.query) > 0
#49	else None
#50	)
#51	except Exception as e:
#52	logging.error(f"Error when processing the chat history. Message is being sent without history. Error: {e}")
#53	answer = self.handle_message(last_message, history)
#54	yield self.text_event(answer)
#55
#56	def handle_message(self, message, history: Optional[list[str]] = None):
#57	if message.startswith("/add "):
#58	response = self.add_data(message)
#59	else:
#60	response = self.ask_bot(message, history)
#61	return response
#62
#63	# def add_data(self, message):
#64	# data = message.split(" ")[-1]
#65	# try:
#66	# self.add(data)
#67	# response = f"Added data from: {data}"
#68	# except Exception:
#69	# logging.exception(f"Failed to add data {data}.")
#70	# response = "Some error occurred while adding data."
#71	# return response
#72
#73	def ask_bot(self, message, history: list[str]):
#74	try:
#75	self.app.llm.set_history(history=history)
#76	response = self.query(message)
#77	except Exception:
#78	logging.exception(f"Failed to query {message}.")
#79	response = "An error occurred. Please try again!"
#80	return response
#81
#82	def start(self):
#83	start_command()
#84
#85
#86	if __name__ == "__main__":
#87	start_command()
#88

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public