integrate fast solver into pipeline: run_pipeline.py now fully automated

- solve_task.py: add fetch_todo_tasks(), _parse_todo_tasks(), run_all() - run_all() auto-fetches all todo tasks from BroJS and solves each one - run_pipeline.py: rewrite to just call run_all() from solve_task - supports: python run_pipeline.py (auto), run_pipeline.py <id...> (targeted) - TARGET_IDS list for hardcoded targets without CLI args - no deepagents imports = no double MCP load on startup Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-04 12:44:07 +03:00
parent bd7580c4bc
commit 928dabe10e
2 changed files with 102 additions and 58 deletions
@@ -1,70 +1,34 @@
-"""Запуск пайплайна для выполнения заданий курса."""
+"""Запуск пайплайна для автоматического выполнения заданий курса.
+
+Использование:
+    python run_pipeline.py           # решить все todo-задания автоматически
+    python run_pipeline.py <id1> <id2>  # решить конкретные задания по ID
+"""
 import asyncio
 import os
-import traceback
+import sys

 os.environ["PYTHONIOENCODING"] = "utf-8"
-# Обходим локальный прокси для OpenRouter (иначе SSL-ошибка)
-os.environ["NO_PROXY"] = "openrouter.ai,platform.brojs.ru,git.brojs.ru," + os.environ.get("NO_PROXY", "")
+os.environ["PYTHONUNBUFFERED"] = "1"
+os.environ["NO_PROXY"] = (
+    "openrouter.ai,platform.brojs.ru,git.brojs.ru,"
+    + os.environ.get("NO_PROXY", "")
+)

 from dotenv import load_dotenv
 load_dotenv()

-from src.agent.graph.pipeline import pipeline, TaskInfo, process_one_task, route, PipelineState
-
-# Целевые задания (None = все todo-задания автоматически)
-TARGET_IDS = [
-    "6a1864f78a94f887e50d46da",  # Экзамен: RAG-агент с ChromaDB и веб-поиском
-]
+from solve_task import run_all

+# Задания из командной строки или жёстко заданный список.
+# Оставь пустым [] — тогда агент сам возьмёт все todo-задания с BroJS.
+TARGET_IDS: list[str] = []

 async def main():
-    print("=== Запуск пайплайна BroJS ===")
-
-    if TARGET_IDS:
-        tasks = [TaskInfo(id=tid, title="", status="todo") for tid in TARGET_IDS]
-        print(f"Целевые задания: {TARGET_IDS}")
-        initial_state = {
-            "tasks": tasks,
-            "current_index": 0,
-            "results": [],
-            "errors": [],
-        }
-        from langgraph.graph import StateGraph, START
-        builder = StateGraph(PipelineState)
-        builder.add_node("process_one_task", process_one_task)
-        builder.add_edge(START, "process_one_task")
-        builder.add_conditional_edges(
-            "process_one_task", route,
-            {"process_one_task": "process_one_task", "__end__": "__end__"}
-        )
-        targeted_pipeline = builder.compile()
-        try:
-            result = await targeted_pipeline.ainvoke(
-                initial_state,
-                {"configurable": {"thread_id": "pipeline-targeted"}},
-            )
-        except Exception as e:
-            print(f"КРИТИЧЕСКАЯ ОШИБКА: {e}")
-            traceback.print_exc()
-            return
-    else:
-        try:
-            result = await pipeline.ainvoke(
-                {"tasks": [], "current_index": 0, "results": [], "errors": []},
-                {"configurable": {"thread_id": "pipeline-main"}},
-            )
-        except Exception as e:
-            print(f"КРИТИЧЕСКАЯ ОШИБКА: {e}")
-            traceback.print_exc()
-            return
-
-    print("\n=== Результат пайплайна ===")
-    for r in result.get("results", []):
-        print(f"  Task {r['task_id'][:8]}: {r['status']} (mode={r['mode']}, retries={r['retries']})")
-    for e in result.get("errors", []):
-        print(f"  ОШИБКА: {e}")
-    print("=== Готово ===")
-
+    # Командная строка имеет приоритет над TARGET_IDS
+    cli_ids = sys.argv[1:] if len(sys.argv) > 1 else None
+    ids = cli_ids or TARGET_IDS or None   # None = автоматически все todo
+    await run_all(target_ids=ids)

+if __name__ == "__main__":
    asyncio.run(main())
@@ -317,6 +317,86 @@ async def generate(task_text: str, retries=5) -> dict:
 # Основная логика
 # ---------------------------------------------------------------------------

+COURSE_ID = "698b49da77cb6d4d2e43ce78"
+
+# ---------------------------------------------------------------------------
+# Автоматическая выборка todo-заданий
+# ---------------------------------------------------------------------------
+
+def _parse_todo_tasks(raw: str) -> list[str]:
+    """Парсит ответ tasks_list и возвращает ID заданий со статусом todo/in_progress."""
+    try:
+        data = json.loads(raw)
+    except (json.JSONDecodeError, TypeError):
+        return []
+    items = data.get("tasks", data) if isinstance(data, dict) else data
+    if not isinstance(items, list):
+        return []
+    result = []
+    for item in items:
+        t      = item.get("task", item) if isinstance(item, dict) else {}
+        tid    = t.get("id", "")
+        status = item.get("status", "")
+        if tid and status in ("todo", "in_progress", "", None):
+            title = t.get("title", t.get("name", ""))
+            result.append((tid, title))
+    return result
+
+
+async def fetch_todo_tasks(course_id: str = COURSE_ID) -> list[tuple[str, str]]:
+    """Возвращает список (task_id, title) незакрытых заданий курса."""
+    print(f"[auto] Получаем список заданий курса {course_id}...")
+    raw = await mcp_call("tasks_list", {"courseId": course_id})
+    tasks = _parse_todo_tasks(raw)
+    print(f"[auto] Найдено todo-заданий: {len(tasks)}")
+    for tid, title in tasks:
+        print(f"  - {tid[:8]}... {title}")
+    return tasks
+
+
+# ---------------------------------------------------------------------------
+# Полный автоматический прогон
+# ---------------------------------------------------------------------------
+
+async def run_all(target_ids: list[str] | None = None, course_id: str = COURSE_ID):
+    """Решает все todo-задания курса (или только target_ids если указан список).
+
+    Это точка входа для run_pipeline.py — никакого ручного вызова не нужно.
+    """
+    if target_ids:
+        tasks = [(tid, "") for tid in target_ids]
+        print(f"[auto] Целевые задания: {target_ids}")
+    else:
+        tasks = await fetch_todo_tasks(course_id)
+
+    if not tasks:
+        print("[auto] Нет заданий для выполнения.")
+        return
+
+    results = []
+    for i, (task_id, title) in enumerate(tasks, 1):
+        print(f"\n[auto] Задание {i}/{len(tasks)}: {task_id[:8]}... {title}")
+        try:
+            repo_url = await solve(task_id)
+            results.append({"task_id": task_id, "status": "ok", "url": repo_url})
+        except Exception as e:
+            print(f"[auto] ОШИБКА при решении {task_id[:8]}: {e}")
+            results.append({"task_id": task_id, "status": "error", "error": str(e)})
+        # Пауза между заданиями
+        if i < len(tasks):
+            print("[auto] Пауза 15с перед следующим заданием...")
+            await asyncio.sleep(15)
+
+    print(f"\n{'='*60}")
+    print("ИТОГ:")
+    for r in results:
+        status_icon = "✅" if r["status"] == "ok" else "❌"
+        detail = r.get("url") or r.get("error", "")
+        print(f"  {status_icon} {r['task_id'][:8]}... → {detail}")
+    print('='*60)
+    return results
+
+
 async def solve(task_id: str):
    print(f"\n{'='*60}")
    print(f"Задание: {task_id}")