Spaces:
Sleeping
Sleeping
| import os | |
| from dotenv import load_dotenv | |
| import gradio as gr | |
| from tools import create_agent | |
| from langchain_core.messages import RemoveMessage | |
| from langchain_core.messages import trim_messages | |
| # Global params | |
| AGENT = create_agent() | |
| theme = gr.themes.Default(primary_hue="red", secondary_hue="red") | |
| default_msg = "Bonjour ! Je suis là pour répondre à vos questions sur l'actuariat. Comment puis-je vous aider aujourd'hui ?" | |
| def filter_msg(msg_list:list, keep_n:int) -> list: | |
| """Keep only last keep_n messages from chat history. Preserves structure user msg -> tool msg -> ai msg""" | |
| msg = trim_messages( | |
| msg_list, | |
| strategy="last", | |
| token_counter=len, | |
| max_tokens=keep_n, | |
| start_on="human", | |
| end_on=("tool", "ai"), | |
| include_system=True, | |
| ) | |
| return [m.id for m in msg] | |
| def agent_response(query, config, keep_n=10): | |
| messages = AGENT.get_state(config).values.get("messages", []) | |
| if len(messages) > keep_n: | |
| keep_msg_ids = filter_msg(messages, keep_n) | |
| AGENT.update_state(config, {"messages": [RemoveMessage(id=m.id) for m in messages if m.id not in keep_msg_ids]}) | |
| print("msg removed") | |
| # Generate answer | |
| answer = AGENT.invoke({"messages":query}, config=config) | |
| return answer["messages"][-1].content | |
| js_func = """ | |
| function refresh() { | |
| const url = new URL(window.location); | |
| if (url.searchParams.get('__theme') != 'light') { | |
| url.searchParams.set('__theme', 'light'); | |
| window.location.href = url.href; | |
| } | |
| } | |
| """ | |
| def delete_agent(): | |
| print("del agent") | |
| global AGENT | |
| AGENT = create_agent() | |
| # print(AGENT.get_state(config).values.get("messages"), "\n\n") | |
| with gr.Blocks(theme=theme, js=js_func, title="Dataltist", fill_height=True) as iface: | |
| gr.Markdown("# Dataltist Chatbot 🚀") | |
| chatbot = gr.Chatbot(show_copy_button=True, show_share_button=False, value=[{"role":"assistant", "content":default_msg}], type="messages", scale=1) | |
| msg = gr.Textbox(lines=1, show_label=False, placeholder="Posez vos questions sur l'assurance") # submit_btn=True | |
| # clear = gr.ClearButton([msg, chatbot], value="Effacer 🗑") | |
| config = {"configurable": {"thread_id": "1"}} | |
| def user(user_message, history: list): | |
| return "", history + [{"role": "user", "content": user_message}] | |
| def bot(history: list): | |
| bot_message = agent_response(history[-1]["content"], config) #AGENT.invoke({"messages":history[-1]["content"]}, config=config) | |
| history.append({"role": "assistant", "content": ""}) | |
| for character in bot_message: | |
| history[-1]['content'] += character | |
| # time.sleep(0.005) | |
| yield history | |
| msg.submit(user, [msg, chatbot], [msg, chatbot], queue=False).then( | |
| bot, chatbot, chatbot | |
| ) | |
| iface.unload(delete_agent) | |
| if __name__ == "__main__": | |
| # load_dotenv() | |
| # AUTH_ID = os.environ.get("AUTH_ID") | |
| # AUTH_PASS = os.environ.get("AUTH_PASS") | |
| iface.launch() #share=True, auth=(AUTH_ID, AUTH_PASS) |