diff --git a/src/agent/graph/pipeline.py b/src/agent/graph/pipeline.py index 49ee65d..9355c8a 100644 --- a/src/agent/graph/pipeline.py +++ b/src/agent/graph/pipeline.py @@ -1,8 +1,10 @@ """LangGraph pipeline: последовательно выполняет все незакрытые задания курса.""" from __future__ import annotations +import asyncio import base64 import json +import re from typing import TypedDict from langchain_core.messages import HumanMessage @@ -80,12 +82,25 @@ _CODING_KW = [ "code", "напиши", "реализуй", "python", "langchain", "langgraph", "агент", "agent", "граф", "graph", "файл", "функц", "программ", "скрипт", "алгоритм", "библиотек", "api", "сервер", "модуль", "класс", + # дополнительные ключевые слова для курса KFU-26-1 + "ai", "llm", "rag", "mcp", "stream", "human", "interrupt", "middleware", + "память", "игра", "текст", "fluency", "практическ", "создай", "создайт", + "задание", "deep", "search", "поиск", ] +# Задания, которые точно не требуют кода (теория, чтение) +_NON_CODING_TITLES = [] + def _is_coding(task: TaskInfo) -> bool: title = (task.get("title") or "").lower() - return any(kw in title for kw in _CODING_KW) + if any(nc in title for nc in _NON_CODING_TITLES): + return False + # Если хотя бы одно кодинговое слово — берём задание + if any(kw in title for kw in _CODING_KW): + return True + # Для этого курса все задания — программирование, берём всё + return True async def _task_text(task_id: str) -> str: @@ -162,10 +177,39 @@ def _fix_prompt(task: TaskInfo, repo_name: str, v: dict) -> str: # --------------------------------------------------------------------------- -# Узлы графа +# Вспомогательное: retry при rate-limit 429 # --------------------------------------------------------------------------- MAX_RETRIES = 2 +RATE_LIMIT_RETRIES = 5 # сколько раз повторять при 429 +RATE_LIMIT_PAUSE = 90 # секунд ожидания перед повтором +TASK_PAUSE = 15 # пауза между заданиями (снижает давление на rate limit) + + +def _is_rate_limit(exc: Exception) -> bool: + """Проверяет, является ли исключение ошибкой rate-limit (429).""" + msg = str(exc) + 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.""" + for attempt in range(1, RATE_LIMIT_RETRIES + 1): + try: + return await agent.ainvoke(messages, config) + except Exception as e: + if _is_rate_limit(e) and attempt < RATE_LIMIT_RETRIES: + wait = RATE_LIMIT_PAUSE * attempt + print(f"[pipeline] Rate limit (попытка {attempt}/{RATE_LIMIT_RETRIES}), " + f"жду {wait}с...") + await asyncio.sleep(wait) + else: + raise + + +# --------------------------------------------------------------------------- +# Узлы графа +# --------------------------------------------------------------------------- async def fetch_tasks(state: PipelineState) -> dict: @@ -225,7 +269,11 @@ async def process_one_task(state: PipelineState) -> dict: agent_to_use = homework_direct_agent try: - result = await agent_to_use.ainvoke( + print(f"[pipeline] Задание {task_id[:8]} — {'пересдача' if is_rework else 'первая сдача'}: " + f"{task.get('title','')[:50]}") + + result = await _invoke_with_retry( + agent_to_use, {"messages": [HumanMessage(content=prompt)]}, {"configurable": {"thread_id": f"pipeline-task-{task_id}"}}, ) @@ -243,12 +291,14 @@ async def process_one_task(state: PipelineState) -> dict: while _needs_retry(verification) and retries < MAX_RETRIES: retries += 1 fix_msg = _fix_prompt(task, repo_name, verification) - result = await agent_to_use.ainvoke( + result = await _invoke_with_retry( + agent_to_use, {"messages": [HumanMessage(content=fix_msg)]}, {"configurable": {"thread_id": f"pipeline-task-{task_id}-retry-{retries}"}}, ) verification = await _verify_repo(repo_name) + print(f"[pipeline] Задание {task_id[:8]} — OK (retries={retries})") results.append({ "task_id": task_id, "status": "done", @@ -259,8 +309,13 @@ async def process_one_task(state: PipelineState) -> dict: }) except Exception as e: + print(f"[pipeline] Задание {task_id[:8]} — ОШИБКА: {e}") errors.append(f"Задание {task_id} ({'rework' if is_rework else 'new'}): {e}") + # Пауза между заданиями чтобы не перегружать rate limit + print(f"[pipeline] Пауза {TASK_PAUSE}с перед следующим заданием...") + await asyncio.sleep(TASK_PAUSE) + return { "results": results, "current_index": state["current_index"] + 1, diff --git a/src/agent/prompts.py b/src/agent/prompts.py index 501e755..d9fbedb 100644 --- a/src/agent/prompts.py +++ b/src/agent/prompts.py @@ -81,6 +81,7 @@ Gitea owner = "glevelll" [2] Составь письменный план: - какие файлы нужны (main.py, requirements.txt, etc.) - что реализовать в каждом файле + - какой технический стек использовать (см. раздел ТЕХНИЧЕСКИЕ ПАТТЕРНЫ ниже) [3] gitea_create_repo({"name": "task-", "private": false}) → Создай репозиторий @@ -131,6 +132,470 @@ Gitea owner = "glevelll" - langchain<=1.0.0 в requirements.txt - Пропускать task_update_answer перед task_submit - Писать код только в requirements.txt без main.py + +## ═══════════════════════════════════════════════ +## ТЕХНИЧЕСКИЕ ПАТТЕРНЫ (читай ПЕРЕД написанием кода) +## ═══════════════════════════════════════════════ + +### LLM — ВСЕГДА используй OpenRouter (не Ollama, не hub.pull, не hardcode) +```python +import os +from langchain_openai import ChatOpenAI + +llm = ChatOpenAI( + model="openai/gpt-oss-20b:free", + base_url="https://openrouter.ai/api/v1", + api_key=os.getenv("OPENAI_API_KEY"), + temperature=0.0, +) +``` +requirements.txt: langchain-openai>=0.3.0 + +--- + +### deepagents — правильный паттерн (задания про "deep agent", "deepagent", "deep agents from scratch") +```python +import os, asyncio +from langchain_openai import ChatOpenAI +from langchain_core.messages import HumanMessage +from langchain.tools import tool +from deepagents import create_deep_agent +from deepagents.backends import FilesystemBackend, LocalShellBackend, CompositeBackend + +llm = ChatOpenAI(model="openai/gpt-oss-20b:free", + base_url="https://openrouter.ai/api/v1", + api_key=os.getenv("OPENAI_API_KEY")) + +# Виртуальная ФС + реальная shell среда +backend = CompositeBackend([ + LocalShellBackend(workspace_dir="./workspace"), + FilesystemBackend(), +]) + +@tool +def web_search(query: str) -> str: + """Search the web for information.""" + try: + from duckduckgo_search import DDGS + with DDGS() as ddgs: + results = list(ddgs.text(query, max_results=5)) + return "\\n".join(f"{r['title']}: {r['body']}" for r in results) + except Exception as e: + return f"Search error: {e}" + +agent = create_deep_agent( + llm=llm, + tools=[web_search], + backend=backend, + system_prompt="You are a helpful research agent.", +) + +async def main(): + result = await agent.ainvoke( + {"messages": [HumanMessage(content="Search for Python best practices and save to results.txt")]}, + {"configurable": {"thread_id": "session-1"}}, + ) + print(result["messages"][-1].content) + +if __name__ == "__main__": + asyncio.run(main()) +``` +requirements.txt: deepagents, langchain-openai>=0.3.0, duckduckgo-search + +--- + +### FastMCP сервер — ТОЛЬКО на уровне модуля, НИКОГДА внутри класса +```python +# ПРАВИЛЬНО: +from fastmcp import FastMCP +import json +from pathlib import Path + +mcp = FastMCP("memory-server") +STORAGE = Path("memory.json") + +def _load(): + return json.loads(STORAGE.read_text()) if STORAGE.exists() else {} + +def _save(data): + STORAGE.write_text(json.dumps(data, indent=2, ensure_ascii=False)) + +@mcp.tool() +def save(key: str, value: str) -> bool: + """Save a value by key.""" + data = _load(); data[key] = value; _save(data) + return True + +@mcp.tool() +def get(key: str) -> str: + """Get a value by key.""" + return _load().get(key, "") + +@mcp.tool() +def delete(key: str) -> bool: + """Delete a key.""" + data = _load() + if key in data: + del data[key]; _save(data); return True + return False + +@mcp.tool() +def list_keys() -> list: + """List all keys.""" + return list(_load().keys()) + +if __name__ == "__main__": + mcp.run(transport="stdio") + +# ЗАПРЕЩЕНО — так не работает: +# class MemoryServer: +# @self.mcp.tool() ← NameError: self не существует в теле класса +# def save(self, ...): ... +``` +requirements.txt: fastmcp>=0.1.0, pydantic>=2.0 + +--- + +### LangChain create_agent — НЕ совместим с AgentExecutor +```python +import asyncio, os +from langchain_openai import ChatOpenAI +from langchain_core.messages import HumanMessage +from langchain.agents import create_agent +from langchain.tools import tool + +llm = ChatOpenAI(model="openai/gpt-oss-20b:free", + base_url="https://openrouter.ai/api/v1", + api_key=os.getenv("OPENAI_API_KEY")) + +@tool +def my_tool(query: str) -> str: + """Tool description.""" + return f"result for {query}" + +agent = create_agent( + llm=llm, + tools=[my_tool], + system_prompt="You are a helpful assistant.", +) + +async def main(): + result = await agent.ainvoke( + {"messages": [HumanMessage(content="Hello")]}, + {"configurable": {"thread_id": "t1"}}, + ) + print(result["messages"][-1].content) + +if __name__ == "__main__": + asyncio.run(main()) + +# ЗАПРЕЩЕНО: +# AgentExecutor(agent=create_agent(...), ...) ← несовместимо! +# agent_type=AgentType.ZERO_SHOT_REACT_DESCRIPTION ← не параметр create_agent +``` +requirements.txt: langchain>=1.2.10, langchain-openai>=0.3.0, langgraph>=0.2.0 + +--- + +### Human-in-the-Loop через HumanInTheLoopMiddleware +```python +import asyncio, json, os +from langchain_openai import ChatOpenAI +from langchain_core.messages import HumanMessage +from langchain.agents import create_agent +from langchain.agents.middleware import HumanInTheLoopMiddleware +from langchain.tools import tool +from langgraph.checkpoint.memory import MemorySaver +from langgraph.types import Command + +llm = ChatOpenAI(model="openai/gpt-oss-20b:free", + base_url="https://openrouter.ai/api/v1", + api_key=os.getenv("OPENAI_API_KEY")) + +@tool +def get_weather(city: str) -> str: + """Get weather for a city.""" + return f"Sunny, 22C in {city}" + +memory = MemorySaver() +agent = create_agent( + llm=llm, + tools=[get_weather], + system_prompt="You are a helpful assistant.", + middleware=[HumanInTheLoopMiddleware(interrupt_on={"get_weather": True})], + checkpointer=memory, +) + +def ask_human(interrupt_value): + decisions = [] + for action in interrupt_value.get("action_requests", []): + print(f"Tool: {action['name']}, Args: {action['args']}") + ans = input("Approve? (y/n): ").strip().lower() + decisions.append({"type": "approve" if ans == "y" else "reject"}) + return decisions + +async def main(): + config = {"configurable": {"thread_id": "session-1"}} + result = await agent.ainvoke( + {"messages": [HumanMessage(content="What's the weather in Moscow?")]}, + config, + ) + while "__interrupt__" in result: + decisions = ask_human(result["__interrupt__"][0].value) + result = await agent.ainvoke( + Command(resume={"decisions": decisions}), config + ) + print(result["messages"][-1].content) + +if __name__ == "__main__": + asyncio.run(main()) +``` +requirements.txt: langchain>=1.2.10, langchain-openai>=0.3.0, langgraph>=0.2.0 + +--- + +### LangGraph interrupt (Human-in-the-loop без middleware) +```python +import asyncio, os +from langchain_openai import ChatOpenAI +from langchain_core.messages import HumanMessage +from langchain.agents import create_agent +from langchain.tools import tool +from langgraph.checkpoint.memory import MemorySaver +from langgraph.types import Command + +llm = ChatOpenAI(model="openai/gpt-oss-20b:free", + base_url="https://openrouter.ai/api/v1", + api_key=os.getenv("OPENAI_API_KEY")) + +@tool +def dangerous_action(cmd: str) -> str: + """Execute a dangerous action.""" + return f"Executed: {cmd}" + +memory = MemorySaver() +agent = create_agent(llm=llm, tools=[dangerous_action], + checkpointer=memory, interrupt_before=["tools"]) + +async def main(): + config = {"configurable": {"thread_id": "t1"}} + result = await agent.ainvoke( + {"messages": [HumanMessage(content="Run ls -la")]}, config + ) + # Агент остановился перед вызовом инструмента + snapshot = await agent.aget_state(config) + if snapshot.next: + ans = input(f"Approve tool call? (y/n): ").strip() + if ans == "y": + result = await agent.ainvoke(Command(resume=None), config) + else: + result = await agent.ainvoke( + Command(resume=None, update={"messages": [ + HumanMessage(content="User rejected the action.") + ]}), config + ) + print(result["messages"][-1].content) + +if __name__ == "__main__": + asyncio.run(main()) +``` + +--- + +### RAG-агент с Qdrant (используй OpenRouter для LLM, Qdrant для векторов) +```python +import os, asyncio +from langchain_openai import ChatOpenAI, OpenAIEmbeddings +from langchain_qdrant import QdrantVectorStore +from langchain.tools import tool +from langchain.agents import create_agent +from langchain_core.messages import HumanMessage +from qdrant_client import QdrantClient +from qdrant_client.models import Distance, VectorParams + +llm = ChatOpenAI(model="openai/gpt-oss-20b:free", + base_url="https://openrouter.ai/api/v1", + api_key=os.getenv("OPENAI_API_KEY")) + +# Embeddings через OpenAI-совместимый API (OpenRouter) +embeddings = OpenAIEmbeddings( + model="text-embedding-3-small", + base_url="https://openrouter.ai/api/v1", + api_key=os.getenv("OPENAI_API_KEY"), +) + +# Qdrant in-memory (не требует отдельного сервера) +client = QdrantClient(":memory:") +client.create_collection("knowledge", + vectors_config=VectorParams(size=1536, distance=Distance.COSINE)) +vector_store = QdrantVectorStore(client=client, collection_name="knowledge", + embedding=embeddings) + +@tool +def search_knowledge_base(query: str, max_results: int = 5) -> str: + """Semantic search in the knowledge base.""" + docs = vector_store.similarity_search(query, k=max_results) + if not docs: + return "No relevant documents found." + return "\\n\\n".join(f"{i+1}. {d.page_content}" for i, d in enumerate(docs)) + +@tool +def add_to_knowledge_base(content: str, title: str = "document") -> str: + """Add text to the knowledge base.""" + from langchain_core.documents import Document + vector_store.add_documents([Document(page_content=content, + metadata={"title": title})]) + return f"Added '{title}' to knowledge base." + +agent = create_agent( + llm=llm, + tools=[search_knowledge_base, add_to_knowledge_base], + system_prompt="You are an assistant with access to a knowledge base.", +) + +async def main(): + await add_to_knowledge_base.ainvoke({"content": "Python is a high-level language.", "title": "python-intro"}) + result = await agent.ainvoke( + {"messages": [HumanMessage(content="What do you know about Python?")]}, + {"configurable": {"thread_id": "rag-1"}}, + ) + print(result["messages"][-1].content) + +if __name__ == "__main__": + asyncio.run(main()) +``` +requirements.txt: langchain>=1.2.10, langchain-openai>=0.3.0, langgraph>=0.2.0, + langchain-qdrant, qdrant-client + +--- + +### Stream-режим агента +```python +import asyncio, os +from langchain_openai import ChatOpenAI +from langchain_core.messages import HumanMessage +from langchain.agents import create_agent +from langchain.tools import tool + +llm = ChatOpenAI(model="openai/gpt-oss-20b:free", + base_url="https://openrouter.ai/api/v1", + api_key=os.getenv("OPENAI_API_KEY"), streaming=True) + +@tool +def calculator(expression: str) -> str: + """Evaluate a math expression.""" + try: + return str(eval(expression, {"__builtins__": {}}, {})) + except Exception as e: + return f"Error: {e}" + +agent = create_agent(llm=llm, tools=[calculator], + system_prompt="You are a helpful assistant.") + +async def main(): + config = {"configurable": {"thread_id": "stream-1"}} + # stream_mode="messages" — получаем токены по одному + async for event in agent.astream( + {"messages": [HumanMessage(content="What is 2+2?")]}, + config, + stream_mode="messages", + ): + if isinstance(event, tuple): + msg, metadata = event + if hasattr(msg, "content") and msg.content: + print(msg.content, end="", flush=True) + print() + +if __name__ == "__main__": + asyncio.run(main()) +``` + +--- + +### Задания типа "план / документ" (не чистый кодинг — например ai-fluency) +Если задание просит написать план, документ или пройти курс: +- Создай main.py который ВЫВОДИТ план в консоль +- План должен быть содержательным (минимум 300 слов), структурированным +- Имитируй личный опыт: "я понял, что...", "мой план включает..." +- Опирайся на тему курса из описания задания + +--- + +### Web search без API-ключа (для поисковых агентов) +```python +from duckduckgo_search import DDGS + +def web_search(query: str) -> str: + with DDGS() as ddgs: + results = list(ddgs.text(query, max_results=5)) + return "\\n".join(f"[{r['title']}] {r['body']} ({r['href']})" for r in results) +``` +requirements.txt: duckduckgo-search + +--- + +### LangGraph текстовая игра с interrupt +```python +import asyncio, os +from langchain_openai import ChatOpenAI +from langchain_core.messages import HumanMessage, SystemMessage +from langgraph.graph import StateGraph, START, END +from langgraph.checkpoint.memory import MemorySaver +from langgraph.types import interrupt, Command +from typing import TypedDict, Annotated +from langgraph.graph.message import add_messages + +class GameState(TypedDict): + messages: Annotated[list, add_messages] + location: str + inventory: list + +llm = ChatOpenAI(model="openai/gpt-oss-20b:free", + base_url="https://openrouter.ai/api/v1", + api_key=os.getenv("OPENAI_API_KEY")) + +def game_master(state: GameState) -> dict: + system = SystemMessage(content=( + "You are a text adventure game master. " + f"Player is at: {state.get('location','start')}. " + f"Inventory: {state.get('inventory',[])}. " + "Describe what happens and list 2-3 options." + )) + response = llm.invoke([system] + state["messages"]) + return {"messages": [response]} + +def player_turn(state: GameState) -> Command: + player_input = interrupt("Your action: ") + return Command(goto="game_master", + update={"messages": [HumanMessage(content=player_input)]}) + +memory = MemorySaver() +builder = StateGraph(GameState) +builder.add_node("game_master", game_master) +builder.add_node("player_turn", player_turn) +builder.add_edge(START, "game_master") +builder.add_edge("game_master", "player_turn") +game = builder.compile(checkpointer=memory) + +async def main(): + config = {"configurable": {"thread_id": "game-1"}} + state = {"messages": [HumanMessage(content="Start the adventure!")], + "location": "forest entrance", "inventory": []} + result = await game.ainvoke(state, config) + while True: + last = result["messages"][-1].content + print(f"\\nGame: {last}") + if "__interrupt__" in result: + action = input("\\nYour action: ").strip() + if action.lower() in ("quit", "exit"): + break + result = await game.ainvoke(Command(resume=action), config) + else: + break + +if __name__ == "__main__": + asyncio.run(main()) +``` """ # ---------------------------------------------------------------------------