Реализация Research-агента: как всё работает вместе

# 1. Клонируем и устанавливаем зависимости
git clone <repo-url>
cd research-agent
python3 -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"

# 2. Создаём .env из шаблона
cp .env.example .env

# Вариант 1: GateLLM (OpenAI-совместимый, работает в России)
LLM_PROVIDER=gatellm
GATELLM_API_KEY=sk-...
DEFAULT_MODEL=qwen/qwen-2.5-72b-instruct

# Вариант 2: Anthropic Claude
LLM_PROVIDER=anthropic
ANTHROPIC_API_KEY=sk-ant-...
DEFAULT_MODEL=claude-sonnet-4-6

# Вариант 3: Ollama (локально, бесплатно)
LLM_PROVIDER=ollama
OLLAMA_BASE_URL=http://localhost:11434
DEFAULT_MODEL=llama3.2

# Параметры агента
MAX_STEPS=10
REQUEST_TIMEOUT=30

# 3. Проверяем установку
pytest tests/ -x -q
python3 -c "from tools.registry import ToolRegistry; print(ToolRegistry().list_tools())"
# ['search_web', 'fetch_pages', 'summarize_page', 'write_report']

# 4. Первый запуск
python main.py "Что такое RAG и как его построить?"

# Опции
python main.py "тема" --provider deepseek --max-steps 15 --save --verbose

SYSTEM_PROMPT = """\
You are a research assistant. Your job is to research a topic and produce a report.

RULES — follow them strictly:
1. You MUST use tools. Do NOT answer from memory alone.
2. You MUST end every session by calling the write_report tool.
3. Never stop with a plain text response — always call a tool.

WORKFLOW:
Step 1 — Call search_web with a specific query to find sources.
Step 2 — Call fetch_pages on the most promising URLs (pick 2-4).
Step 3 — If any page is very long, call summarize_page on it.
Step 4 — Call write_report with a structured Markdown report and all sources.
"""

async def run(self, query: str) -> AgentState:
    state = AgentState(query=query)
    state.append_message(Message(role="user", content=f"Research topic: {query}"))
    tools = self.registry.get_schemas()   # JSON-схемы для LLM

    while state.step < self.max_steps:   # Инвариант 1: лимит шагов
        state.increment_step()
        log.info("react_step", step=state.step)

        # --- Шаг 1: запрос к LLM ---
        response = await self.llm.complete(
            messages=state.to_api_messages(),  # вся история
            tools=tools,                        # доступные инструменты
            system=SYSTEM_PROMPT,
        )

        # --- Шаг 2: сохраняем ответ LLM в историю ---
        state.append_message(Message(role="assistant", content=response["content"]))

        stop_reason = response.get("stop_reason", "")

        # --- Случай А: LLM не вызвала инструмент ---
        if stop_reason == "end_turn":
            # ... обработка (см. ниже)
            break

        # --- Случай Б: LLM вызывает инструмент ---
        if stop_reason == "tool_use":
            tool_results = []
            done = False

            for block in response["content"]:
                if block.get("type") != "tool_use":
                    continue

                result_str = await self._dispatch_tool(
                    state, block["name"], block["input"], block["id"]
                )

                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block["id"],
                    "content": result_str,
                })

                if block["name"] == "write_report":
                    done = True    # Инвариант 5: write_report завершает
                    break

            # Все результаты — одним сообщением user (требование API)
            if tool_results:
                state.append_message(Message(role="user", content=tool_results))

            if done:
                break

    return state

if stop_reason == "end_turn":
    text_parts = [b["text"] for b in response["content"] if b.get("type") == "text"]

    if text_parts and not state.report:
        # Сценарий 1: LLM написала отчёт как текст, не через write_report
        # Сохраняем как fallback
        state.report = f"# Research: {query}\n\n" + "\n\n".join(text_parts)
        log.warning("fallback_report_from_text")
        break

    if not state.report and state.step < self.max_steps:
        # Сценарий 2: пустой ответ — напоминаем LLM, что нужно написать отчёт
        log.warning("end_turn_no_content_retry")
        state.append_message(Message(
            role="user",
            content="You stopped without calling write_report. "
                    "You MUST call write_report now."
        ))
        continue    # ← ещё одна итерация

    # Сценарий 3: отчёт уже есть или шаги закончились
    break

async def _dispatch_tool(self, state, tool_name, tool_input, tool_use_id) -> str:
    log.info("tool_dispatched", tool=tool_name, step=state.step)  # Инвариант 2

    try:
        result = await self.registry.dispatch(tool_name, **tool_input)

        if tool_name == "write_report":
            state.report = result.content          # сохраняем отчёт
            for src in result.sources:
                state.add_source(Source(...))
            return f"Report written: {result.title}"

        if tool_name == "search_web":
            for r in result:
                state.add_source(Source(url=r["url"], title=r.get("title", "")))

        return json.dumps(result, ensure_ascii=False, default=str)

    except ToolError as e:
        # Инвариант 3: ToolError НИКОГДА не ломает цикл
        log.warning("tool_error", tool=tool_name, error=str(e))
        return f"Error executing {tool_name}: {e}"  # LLM увидит ошибку и адаптируется

class LLMClientProtocol(Protocol):
    """Интерфейс, которому должны соответствовать все клиенты."""
    async def complete(
        self,
        messages: list[dict],
        tools: list[dict],
        system: str = "",
    ) -> LLMResponse: ...

# Все провайдеры возвращают единый формат:
{
    "content": [
        {"type": "text", "text": "Сначала поищу источники."},
        {
            "type": "tool_use",
            "id": "tu_001",
            "name": "search_web",
            "input": {"query": "RAG best practices 2024"}
        }
    ],
    "stop_reason": "tool_use",  # или "end_turn"
    "usage": {"input_tokens": 150, "output_tokens": 30}
}