diff --git a/src/agent/graph/pipeline.py b/src/agent/graph/pipeline.py
index 05b1301..2acb93d 100644
--- a/src/agent/graph/pipeline.py
+++ b/src/agent/graph/pipeline.py
@@ -8,6 +8,7 @@ import re
from typing import TypedDict
from langchain_core.messages import HumanMessage
+from langchain_core.runnables import RunnableConfig
from langgraph.graph import START, StateGraph
from src.agent.agent import homework_direct_agent, journal as _journal_toolsets, rework_agent
@@ -236,11 +237,16 @@ def _is_rate_limit(exc: Exception) -> bool:
return "429" in msg or "rate" in msg.lower() or "rate_limit" in msg.lower()
-async def _invoke_with_retry(agent, messages, config):
- """Вызывает агента с автоматическим retry при 429."""
+async def _invoke_with_retry(agent, messages, config, callbacks=None):
+ """Вызывает агента с автоматическим retry при 429.
+ callbacks — список LangChain callback-объектов (например AgentCallback из UI).
+ """
+ run_config = dict(config)
+ if callbacks:
+ run_config["callbacks"] = callbacks
for attempt in range(1, RATE_LIMIT_RETRIES + 1):
try:
- return await agent.ainvoke(messages, config)
+ return await agent.ainvoke(messages, run_config)
except Exception as e:
if _is_rate_limit(e) and attempt < RATE_LIMIT_RETRIES:
wait = RATE_LIMIT_PAUSE * attempt
@@ -276,8 +282,8 @@ async def fetch_tasks(state: PipelineState) -> dict:
return {"tasks": coding, "current_index": 0, "results": [], "errors": []}
-async def process_one_task(state: PipelineState) -> dict:
- """Выполняет одно задание."""
+async def process_one_task(state: PipelineState, config: RunnableConfig | None = None) -> dict:
+ """Выполняет одно задание. config может содержать callbacks из UI."""
if state["current_index"] >= len(state["tasks"]):
return state
@@ -316,6 +322,9 @@ async def process_one_task(state: PipelineState) -> dict:
)
agent_to_use = homework_direct_agent
+ # Извлекаем callbacks из LangGraph config (переданы из UI)
+ callbacks = (config or {}).get("callbacks") or []
+
try:
print(f"[pipeline] Задание {task_id[:8]} — {'пересдача' if is_rework else 'первая сдача'}: "
f"{task.get('title','')[:50]}")
@@ -324,6 +333,7 @@ async def process_one_task(state: PipelineState) -> dict:
agent_to_use,
{"messages": [HumanMessage(content=prompt)]},
{"configurable": {"thread_id": f"pipeline-task-{task_id}"}},
+ callbacks=callbacks,
)
last = (result.get("messages") or [{}])[-1]
output = getattr(last, "content", str(last))
@@ -343,6 +353,7 @@ async def process_one_task(state: PipelineState) -> dict:
agent_to_use,
{"messages": [HumanMessage(content=fix_msg)]},
{"configurable": {"thread_id": f"pipeline-task-{task_id}-retry-{retries}"}},
+ callbacks=callbacks,
)
verification = await _verify_repo(repo_name)
diff --git a/ui.py b/ui.py
index bf31c71..09f3d44 100644
--- a/ui.py
+++ b/ui.py
@@ -426,28 +426,69 @@ with tab_pipeline:
st.divider()
st.subheader("Запустить все todo-задания")
if st.button("⚡ Запустить pipeline для всех заданий", use_container_width=True):
- pl = get_pipeline()
- with st.spinner("Pipeline работает... (может занять несколько минут)"):
- try:
- res = asyncio.run(pl.ainvoke(
- {"tasks": [], "current_index": 0, "results": [], "errors": []}
- ))
- results = res.get("results", [])
- errors = res.get("errors", [])
+ pl = get_pipeline()
+ evq_pl = queue.Queue()
+ cb_pl = AgentCallback(evq_pl)
+ all_pl_events: list[dict] = []
+ pl_result = None
+ pl_error = None
- md = [f"### Результат: {len(results)} заданий\n"]
- for r in results:
- tid = r.get("task_id", "")
- url = f"https://git.brojs.ru/{GITEA_OWNER}/task-{tid}"
- icon = "✅" if r.get("status") == "ok" else "❌"
- md.append(f"- {icon} `{tid[:8]}...` — [{r.get('status','')}]({url})")
- if errors:
- md.append(f"\n**Ошибки ({len(errors)}):**")
- for e in errors:
- md.append(f"- {e}")
- st.markdown("\n".join(md))
- except Exception as e:
- st.error(str(e))
+ def _run_pipeline():
+ nonlocal pl_result, pl_error
+ async def _inner():
+ nonlocal pl_result, pl_error
+ try:
+ pl_result = await pl.ainvoke(
+ {"tasks": [], "current_index": 0, "results": [], "errors": []},
+ {"callbacks": [cb_pl]},
+ )
+ except Exception as e:
+ pl_error = str(e)
+ asyncio.run(_inner())
+
+ t_pl = threading.Thread(target=_run_pipeline, daemon=True)
+ t_pl.start()
+
+ events_pl_ph = st.empty()
+ with st.spinner("Pipeline работает... (может занять несколько минут)"):
+ while t_pl.is_alive() or not evq_pl.empty():
+ while not evq_pl.empty():
+ ev = evq_pl.get_nowait()
+ if ev["t"] not in ("done", "fatal"):
+ all_pl_events.append(ev)
+ if all_pl_events:
+ html = "".join(_render_event(e) for e in all_pl_events[-60:])
+ events_pl_ph.markdown(
+ f'{html}
',
+ unsafe_allow_html=True,
+ )
+ time.sleep(0.15)
+
+ events_pl_ph.empty()
+
+ if all_pl_events:
+ with st.expander(f"🔍 Лог pipeline ({len(all_pl_events)} событий)", expanded=False):
+ html = "".join(_render_event(e) for e in all_pl_events[-80:])
+ st.markdown(f'{html}
',
+ unsafe_allow_html=True)
+
+ if pl_error:
+ st.error(pl_error)
+ elif pl_result:
+ results = pl_result.get("results", [])
+ errors = pl_result.get("errors", [])
+ md = [f"### Результат: {len(results)} заданий\n"]
+ for r in results:
+ tid = r.get("task_id", "")
+ url = f"https://git.brojs.ru/{GITEA_OWNER}/task-{tid}"
+ icon = "✅" if r.get("status") == "ok" else "❌"
+ md.append(f"- {icon} `{tid[:8]}...` — [{r.get('status','')}]({url})")
+ if errors:
+ md.append(f"\n**Ошибки ({len(errors)}):**")
+ for e in errors:
+ md.append(f"- {e}")
+ st.markdown("\n".join(md))
# ══════════════════════════════════════════════════════════════════════════