diff --git a/ui.py b/ui.py index 9db31ec..bf31c71 100644 --- a/ui.py +++ b/ui.py @@ -1,7 +1,7 @@ """ Streamlit UI для brojs-agent. -Запуск: python -m streamlit run ui.py +Запуск: streamlit run ui.py """ import asyncio import json @@ -68,16 +68,12 @@ st.markdown(""" .thinking { color: #64748b; font-style: italic; font-size: .82em; padding: 4px 0; } -.status-ok { background:#052e16; border:1px solid #22c55e; color:#4ade80; - padding:10px 16px; border-radius:8px; } -.status-warn { background:#1c1917; border:1px solid #f59e0b; color:#fbbf24; - padding:10px 16px; border-radius:8px; } -.status-err { background:#1c0a0a; border:1px solid #ef4444; color:#f87171; - padding:10px 16px; border-radius:8px; } -.chat-user { background:#1e3a5f; border-radius:12px 12px 2px 12px; - padding:10px 14px; margin:6px 0; } -.chat-agent { background:#1a1a2e; border-radius:12px 12px 12px 2px; - padding:10px 14px; margin:6px 0; } +.status-ok { background:#052e16; border:1px solid #22c55e; color:#4ade80; padding:10px 16px; border-radius:8px; } +.status-warn { background:#1c1917; border:1px solid #f59e0b; color:#fbbf24; padding:10px 16px; border-radius:8px; } +.status-err { background:#1c0a0a; border:1px solid #ef4444; color:#f87171; padding:10px 16px; border-radius:8px; } + +.chat-user { background:#1e3a5f; border-radius:12px 12px 2px 12px; padding:10px 14px; margin:6px 0; } +.chat-agent { background:#1a1a2e; border-radius:12px 12px 12px 2px; padding:10px 14px; margin:6px 0; } """, unsafe_allow_html=True) @@ -93,118 +89,76 @@ st.markdown(""" """, unsafe_allow_html=True) # --------------------------------------------------------------------------- -# Кэш агента — загружается один раз при первом обращении +# Кэш агента # --------------------------------------------------------------------------- -@st.cache_resource(show_spinner="⏳ Инициализация агента (подключение к MCP)...") +@st.cache_resource(show_spinner="Инициализация агента (~30с)...") def get_agent(): from src.agent.agent import homework_direct_agent return homework_direct_agent -@st.cache_resource(show_spinner="⏳ Загрузка pipeline...") +@st.cache_resource(show_spinner="Загрузка pipeline...") def get_pipeline(): from src.agent.graph.pipeline import pipeline return pipeline # --------------------------------------------------------------------------- -# Сборщик событий агента (синхронный — не нужна очередь) +# Callback — перехватывает события агента и шлёт в очередь # --------------------------------------------------------------------------- -_SOLVE_TOOLS = {"validate_teacher_comment", "generate_code_solution"} +_SOLVE_TOOLS = {"validate_teacher_comment", "generate_code_solution"} _JOURNAL_PREFIX = "mcp__journal-bh-professor__" -class AgentEventCollector(BaseCallbackHandler): - """Накапливает события агента в список во время синхронного вызова.""" - - def __init__(self) -> None: - self.events: list[dict] = [] +class AgentCallback(BaseCallbackHandler): + def __init__(self, q: queue.Queue): + self.q = q def _ts(self) -> str: return datetime.now().strftime("%H:%M:%S") - def _add(self, ev: dict) -> None: - self.events.append(ev) - def on_tool_start(self, serialized, input_str, **kwargs): name = serialized.get("name", "?") try: args = json.loads(str(input_str)) if isinstance(input_str, str) else input_str except Exception: args = {} - self._add({"t": "tool_start", "ts": self._ts(), "name": name, "args": args}) + self.q.put({"t": "tool_start", "ts": self._ts(), "name": name, "args": args}) def on_tool_end(self, output, **kwargs): - self._add({"t": "tool_end", "ts": self._ts(), "output": str(output)[:300]}) + self.q.put({"t": "tool_end", "ts": self._ts(), "output": str(output)[:300]}) def on_tool_error(self, error, **kwargs): - self._add({"t": "tool_error", "ts": self._ts(), "msg": str(error)[:200]}) + self.q.put({"t": "tool_error", "ts": self._ts(), "msg": str(error)[:200]}) def on_llm_start(self, *a, **kw): - self._add({"t": "thinking", "ts": self._ts()}) + self.q.put({"t": "thinking", "ts": self._ts()}) def on_llm_end(self, response, **kwargs): try: text = response.generations[0][0].text[:120] - self._add({"t": "llm_end", "ts": self._ts(), "preview": text}) + self.q.put({"t": "llm_end", "ts": self._ts(), "preview": text}) except Exception: pass # --------------------------------------------------------------------------- -# Запуск агента — блокирующий вызов в отдельном потоке -# (избегаем конфликтов с event loop Streamlit) +# Запуск агента в фоне # --------------------------------------------------------------------------- -def _invoke_agent( - agent, messages: dict, config: dict -) -> tuple[dict | None, list[dict], str | None]: - """ - Запускает агента синхронно. Блокирует поток до завершения. - Возвращает (result, events, error_message). - """ - from src.agent.middlewares.retry_on_rate_limit import set_ui_event_queue - - collector = AgentEventCollector() - # Очередь для rate-limit событий из middleware - rl_queue: queue.Queue = queue.Queue() - set_ui_event_queue(rl_queue) - - result: dict | None = None - error: str | None = None - - def _thread_fn() -> None: - nonlocal result, error - - async def _inner() -> None: - nonlocal result, error - try: - r = await agent.ainvoke(messages, {**config, "callbacks": [collector]}) - result = r - except Exception as e: - error = str(e) - - asyncio.run(_inner()) - - t = threading.Thread(target=_thread_fn, daemon=True) - t.start() - t.join() # ждём завершения (UI показывает spinner) - - set_ui_event_queue(None) - - # Переносим rate-limit события в список collector'а - while not rl_queue.empty(): +def _run_agent_thread(agent, messages, config, q: queue.Queue, cb: AgentCallback): + async def _inner(): try: - collector.events.append(rl_queue.get_nowait()) - except Exception: - break - - return result, collector.events, error + result = await agent.ainvoke(messages, {**config, "callbacks": [cb]}) + q.put({"t": "done", "result": result}) + except Exception as e: + q.put({"t": "fatal", "msg": str(e)}) + asyncio.run(_inner()) # --------------------------------------------------------------------------- -# Рендер одного события в HTML +# Рендер одного события в лог # --------------------------------------------------------------------------- def _render_event(ev: dict) -> str: @@ -215,19 +169,28 @@ def _render_event(ev: dict) -> str: return f'
💭 {ts} модель думает...
' if kind == "tool_start": - name = ev["name"] - args = ev.get("args", {}) + name = ev["name"] + args = ev.get("args", {}) short = name.replace(_JOURNAL_PREFIX, "mcp::") - + # Определяем тип инструмента if name in _SOLVE_TOOLS: - cls, icon, label = "tool-subagent", "🧠", f"[субагент] {short}" + cls = "tool-subagent" + icon = "🧠" + label = f"[субагент] {short}" elif "gitea" in name: - cls, icon, label = "tool-call", "📦", short + cls = "tool-call" + icon = "📦" + label = short elif "mcp::" in short or "journal" in name: - cls, icon, label = "tool-call", "📡", short + cls = "tool-call" + icon = "📡" + label = short else: - cls, icon, label = "tool-call", "🔧", short + cls = "tool-call" + icon = "🔧" + label = short + # Показываем ключевые аргументы hint = "" for key in ("taskId", "path", "repo", "repo_name", "name"): if key in args: @@ -244,21 +207,6 @@ def _render_event(ev: dict) -> str: msg = ev["msg"].replace("<", "<") return f'
⚠ {msg}
' - if kind == "rate_limit_wait": - name, pause, attempt, mx = ( - ev.get("name", "?"), ev.get("pause", 30), - ev.get("attempt", 1), ev.get("max", 5), - ) - return (f'
' - f'⏳ {ts} 429 rate limit — {name} · ждал {pause}с ' - f'(попытка {attempt}/{mx})
') - - if kind == "rate_limit_retry": - name = ev.get("name", "?") - attempt = ev.get("attempt", 2) - return (f'
' - f'🔄 {ts} повтор {name} (попытка {attempt})...
') - if kind == "llm_end": preview = ev.get("preview", "").replace("<", "<")[:100] return f'
✏ {ts} {preview}...
' @@ -266,22 +214,11 @@ def _render_event(ev: dict) -> str: return "" -def _show_events(events: list[dict], expanded: bool = False) -> None: - if not events: - return - with st.expander(f"🔍 Лог инструментов ({len(events)} событий)", expanded=expanded): - html = "".join(_render_event(e) for e in events[-80:]) - st.markdown(f'
{html}
', - unsafe_allow_html=True) - - # --------------------------------------------------------------------------- # Вкладки # --------------------------------------------------------------------------- -tab_chat, tab_pipeline, tab_status = st.tabs( - ["💬 Чат с агентом", "⚡ Pipeline", "📊 Статус заданий"] -) +tab_chat, tab_pipeline, tab_status = st.tabs(["💬 Чат с агентом", "⚡ Pipeline", "📊 Статус заданий"]) # ══════════════════════════════════════════════════════════════════════════ @@ -289,8 +226,9 @@ tab_chat, tab_pipeline, tab_status = st.tabs( # ══════════════════════════════════════════════════════════════════════════ with tab_chat: - st.caption("Пиши агенту напрямую. Пока он работает — показывается spinner.") + st.caption("Общайся с агентом: задай вопрос, попроси решить задание или разобрать ситуацию.") + # История сообщений if "chat_history" not in st.session_state: st.session_state.chat_history = [] if "chat_events" not in st.session_state: @@ -298,25 +236,28 @@ with tab_chat: if "chat_thread_id" not in st.session_state: st.session_state.chat_thread_id = f"ui-{int(time.time())}" - # История сообщений + # Показываем историю for msg in st.session_state.chat_history: - role, text = msg["role"], msg["text"] + role = msg["role"] + text = msg["text"] if role == "user": - st.markdown(f'
👤 {text}
', - unsafe_allow_html=True) + st.markdown(f'
👤 {text}
', unsafe_allow_html=True) else: - st.markdown(f'
🤖 {text}
', + st.markdown(f'
🤖 {text}
', unsafe_allow_html=True) + + # Лог событий (раскрывающийся) + if st.session_state.chat_events: + with st.expander(f"🔍 Лог инструментов ({len(st.session_state.chat_events)} событий)", expanded=False): + html = "".join(_render_event(e) for e in st.session_state.chat_events[-80:]) + st.markdown(f'
{html}
', unsafe_allow_html=True) - # Лог предыдущего запроса - _show_events(st.session_state.chat_events) - - # Поле ввода + # Ввод col_input, col_btn = st.columns([5, 1]) with col_input: user_input = st.text_input( "Сообщение", - placeholder='"Реши задание 6a1864f7..." или "Какие задания у меня есть?"', + placeholder='Например: "Реши задание 6a1864f7..." или "Какие задания у меня есть?"', label_visibility="collapsed", key="chat_input", ) @@ -328,46 +269,75 @@ with tab_chat: st.session_state.chat_history.append({"role": "user", "text": msg_text}) st.session_state.chat_events = [] - lc_messages = [ - HumanMessage(content=m["text"]) if m["role"] == "user" - else AIMessage(content=m["text"]) - for m in st.session_state.chat_history - ] + # Строим историю сообщений для агента + lc_messages = [] + for m in st.session_state.chat_history: + if m["role"] == "user": + lc_messages.append(HumanMessage(content=m["text"])) + else: + lc_messages.append(AIMessage(content=m["text"])) + config = {"configurable": {"thread_id": st.session_state.chat_thread_id}} + agent = get_agent() - try: - agent = get_agent() - except Exception as e: - st.error(f"Ошибка инициализации агента: {e}") - st.stop() + # Placeholders для обновления в реальном времени + events_ph = st.empty() + status_ph = st.empty() - with st.spinner("🤖 Агент работает... (это может занять несколько минут)"): - result, events, error = _invoke_agent( - agent, {"messages": lc_messages}, config - ) + evq: queue.Queue = queue.Queue() + cb = AgentCallback(evq) + all_events: list[dict] = [] - st.session_state.chat_events = events + thread = threading.Thread( + target=_run_agent_thread, + args=(agent, {"messages": lc_messages}, config, evq, cb), + daemon=True, + ) + thread.start() - if error: - st.session_state.chat_history.append( - {"role": "agent", "text": f"⚠️ Ошибка: {error}"} - ) - elif result: - msgs = result.get("messages", []) - last = msgs[-1] if msgs else None + final_result = None + fatal = None + + while thread.is_alive() or not evq.empty(): + changed = False + while not evq.empty(): + ev = evq.get_nowait() + if ev["t"] in ("done", "fatal"): + if ev["t"] == "done": + final_result = ev["result"] + else: + fatal = ev["msg"] + else: + all_events.append(ev) + changed = True + + if changed and all_events: + html = "".join(_render_event(e) for e in all_events[-60:]) + events_ph.markdown( + f'
{html}
', + unsafe_allow_html=True, + ) + time.sleep(0.15) + + events_ph.empty() + st.session_state.chat_events = all_events + + if fatal: + st.session_state.chat_history.append({"role": "agent", "text": f"⚠️ Ошибка: {fatal}"}) + elif final_result: + msgs = final_result.get("messages", []) + last = msgs[-1] if msgs else None reply = last.content if last and hasattr(last, "content") else "Готово." st.session_state.chat_history.append({"role": "agent", "text": reply}) - else: - st.session_state.chat_history.append( - {"role": "agent", "text": "Агент завершил работу без ответа."} - ) st.rerun() + # Кнопка очистки if st.session_state.chat_history: if st.button("🗑 Очистить чат"): - st.session_state.chat_history = [] - st.session_state.chat_events = [] + st.session_state.chat_history = [] + st.session_state.chat_events = [] st.session_state.chat_thread_id = f"ui-{int(time.time())}" st.rerun() @@ -377,17 +347,12 @@ with tab_chat: # ══════════════════════════════════════════════════════════════════════════ with tab_pipeline: - st.caption("Запусти конкретное задание по ID или все todo-задания сразу.") - - if "pipe_events" not in st.session_state: - st.session_state.pipe_events = [] - if "pipe_result_text" not in st.session_state: - st.session_state.pipe_result_text = "" + st.caption("Автоматически решает все todo-задания курса по очереди.") col1, col2 = st.columns([3, 1]) with col1: task_id_input = st.text_input( - "Task ID", + "Task ID (оставь пустым — решить все todo)", placeholder="6a1864f7fd30e81cf3146d65", label_visibility="visible", ) @@ -395,48 +360,68 @@ with tab_pipeline: st.write("") run_btn = st.button("▶ Запустить", type="primary", use_container_width=True) - # Показываем результат предыдущего запуска - if st.session_state.pipe_result_text: - st.markdown(st.session_state.pipe_result_text, unsafe_allow_html=True) - - _show_events(st.session_state.pipe_events) - if run_btn: - if not task_id_input.strip(): - st.warning("Введи Task ID") - else: - task_id = task_id_input.strip() + result_ph = st.empty() + events_ph2 = st.empty() + agent = get_agent() + + if task_id_input.strip(): + # Одно задание + task_id = task_id_input.strip() repo_url = f"https://git.brojs.ru/{GITEA_OWNER}/task-{task_id}" prompt = f"Реши задание taskId={task_id} курса 698b49da77cb6d4d2e43ce78" config = {"configurable": {"thread_id": f"pipe-{task_id}-{int(time.time())}"}} messages = {"messages": [HumanMessage(content=prompt)]} + else: + result_ph.info("Pipeline для всех todo-заданий — используй раздел ниже") + st.stop() - st.session_state.pipe_events = [] - st.session_state.pipe_result_text = "" + evq2: queue.Queue = queue.Queue() + cb2 = AgentCallback(evq2) + all_events2: list[dict] = [] - try: - agent = get_agent() - except Exception as e: - st.error(f"Ошибка инициализации агента: {e}") - st.stop() + thread2 = threading.Thread( + target=_run_agent_thread, + args=(agent, messages, config, evq2, cb2), + daemon=True, + ) + thread2.start() - with st.spinner(f"🤖 Решаю задание {task_id[:8]}..."): - result, events, error = _invoke_agent(agent, messages, config) + final2 = None + fatal2 = None - st.session_state.pipe_events = events + with st.spinner(f"Агент решает {task_id[:8]}..."): + while thread2.is_alive() or not evq2.empty(): + while not evq2.empty(): + ev = evq2.get_nowait() + if ev["t"] == "done": + final2 = ev["result"] + elif ev["t"] == "fatal": + fatal2 = ev["msg"] + else: + all_events2.append(ev) - if error: - st.session_state.pipe_result_text = ( - f'
❌ Ошибка: {error[:400]}
' - ) - else: - st.session_state.pipe_result_text = ( - f'
✅ Готово! ' - f'' - f'Открыть репозиторий ↗
' - ) + if all_events2: + html = "".join(_render_event(e) for e in all_events2[-50:]) + events_ph2.markdown( + f'
{html}
', + unsafe_allow_html=True, + ) + time.sleep(0.15) - st.rerun() + if fatal2: + result_ph.markdown( + f'
❌ Ошибка: {fatal2[:300]}
', + unsafe_allow_html=True, + ) + elif final2: + result_ph.markdown( + f'
✅ Готово! ' + f'Открыть репозиторий' + f'
', + unsafe_allow_html=True, + ) st.divider() st.subheader("Запустить все todo-задания") @@ -478,20 +463,19 @@ with tab_status: from src.agent.mcp_client import load_journal_toolsets async def _fetch(): - j = load_journal_toolsets() + j = load_journal_toolsets() + # Инструменты имеют префикс mcp__journal-bh-professor__ tools = {t.name: t for t in j.tasks_submissions_tools} full_name = "mcp__journal-bh-professor__tasks_list" - t = tools.get(full_name) or next( - (v for k, v in tools.items() if "tasks_list" in k), None - ) + t = tools.get(full_name) if not t: - st.warning(f"tasks_list не найден. Доступны: {list(tools.keys())}") + # fallback: ищем по любому имени содержащему tasks_list + t = next((v for k, v in tools.items() if "tasks_list" in k), None) + if not t: + st.warning(f"Инструмент tasks_list не найден. Доступны: {list(tools.keys())}") return [] raw = await t.ainvoke({"courseId": "698b49da77cb6d4d2e43ce78"}) - text = ( - next((x["text"] for x in raw if x.get("type") == "text"), str(raw)) - if isinstance(raw, list) else str(raw) - ) + text = next((x["text"] for x in raw if x.get("type") == "text"), str(raw)) if isinstance(raw, list) else str(raw) data = json.loads(text) return data.get("tasks", data) if isinstance(data, dict) else data @@ -521,7 +505,7 @@ with tab_status: title = t.get("title", t.get("name", "")) counts[status] = counts.get(status, 0) + 1 rows.append({ - "": STATUS_EMOJI.get(status, "❓"), + "": STATUS_EMOJI.get(status, "❓"), "Статус": status, "ID": tid[:12] + "...", "Название": title, @@ -533,6 +517,6 @@ with tab_status: st.divider() cols = st.columns(len(counts)) for col, (s, n) in zip(cols, counts.items()): - col.metric(f"{STATUS_EMOJI.get(s, '❓')} {s}", n) + col.metric(f"{STATUS_EMOJI.get(s,'❓')} {s}", n) else: st.info("Нажми «Обновить статусы» чтобы загрузить данные.")