Spaces:
Running
Running
File size: 3,831 Bytes
6db8557 71a34b2 6db8557 ebea497 6db8557 817058a 71a34b2 817058a 71a34b2 817058a 71a34b2 817058a 6db8557 71a34b2 6db8557 ebea497 6db8557 ebea497 6db8557 71a34b2 6db8557 71a34b2 ebea497 6db8557 ebea497 |
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 |
from PIL import Image
import sys
import re
import streamlit as st
from vectara_agentic.agent import AgentStatusType
from agent import initialize_agent
from config import get_agent_config
initial_prompt = "How can I help you today?"
def format_log_msg(log_msg: str):
max_log_msg_size = 500
return log_msg if len(log_msg) <= max_log_msg_size else log_msg[:max_log_msg_size]+'...'
def agent_progress_callback(status_type: AgentStatusType, msg: str):
output = f'<span style="color:blue;">{status_type.value}</span>: {msg}'
st.session_state.log_messages.append(output)
if 'status' in st.session_state:
latest_message = ''
if status_type == AgentStatusType.TOOL_CALL:
match = re.search(r"'([^']*)'", msg)
tool_name = match.group(1) if match else "Unknown tool"
latest_message = f"Calling tool {tool_name}..."
elif status_type == AgentStatusType.TOOL_OUTPUT:
latest_message = "Analyzing tool output..."
else:
return
st.session_state.status.update(label=latest_message)
with st.session_state.status:
for log_msg in st.session_state.log_messages:
st.markdown(format_log_msg(log_msg), unsafe_allow_html=True)
@st.dialog(title="Agent logs", width='large')
def show_modal():
for log_msg in st.session_state.log_messages:
st.markdown(format_log_msg(log_msg), unsafe_allow_html=True)
async def launch_bot():
def reset():
st.session_state.messages = [{"role": "assistant", "content": initial_prompt, "avatar": "🦖"}]
st.session_state.log_messages = []
st.session_state.prompt = None
st.session_state.ex_prompt = None
st.session_state.first_turn = True
st.session_state.show_logs = False
if 'agent' not in st.session_state:
st.session_state.agent = initialize_agent(cfg, agent_progress_callback=agent_progress_callback)
else:
st.session_state.agent.clear_memory()
if 'cfg' not in st.session_state:
cfg = get_agent_config()
st.session_state.cfg = cfg
st.session_state.ex_prompt = None
reset()
cfg = st.session_state.cfg
print(f'Configuration: {cfg}')
# left side content
# Display chat messages
for message in st.session_state.messages:
print(f'Message: {message}')
with st.chat_message(message["role"], avatar=message["avatar"]):
st.write(message["content"])
# User-provided prompt
if st.session_state.ex_prompt:
prompt = st.session_state.ex_prompt
else:
prompt = st.chat_input()
if prompt:
st.session_state.messages.append({"role": "user", "content": prompt, "avatar": '🧑💻'})
st.session_state.prompt = prompt
st.session_state.log_messages = []
st.session_state.show_logs = False
with st.chat_message("user", avatar='🧑💻'):
print(f"Starting new question: {prompt}\n")
st.write(prompt)
st.session_state.ex_prompt = None
# Generate a new response if last message is not from assistant
if st.session_state.prompt:
with st.chat_message("assistant", avatar='🤖'):
st.session_state.status = st.status('Processing...', expanded=False)
res = st.session_state.agent.chat(st.session_state.prompt)
#res = escape_dollars_outside_latex(res)
res = str(res)
message = {"role": "assistant", "content": res, "avatar": '🤖'}
st.session_state.messages.append(message)
st.markdown(res)
st.session_state.ex_prompt = None
st.session_state.prompt = None
st.session_state.first_turn = False
st.rerun()
sys.stdout.flush() |