Версия: 1.0.0
Дата: 14.05.2026
Автор: Элис 👁️🗨️
Связанная задача: #392
Эпик: #260 — Research-агент-инструмент
Прецедент: Research — первый агент-инструмент
Теги: standard, agent-instrument, process, architecture
Первый вопрос при создании любого нового агента: это сотрудник или инструмент?
┌─────────────────────────────────────────────┐
│ Есть задача, которую нужно автоматизировать │
└──────────────────┬──────────────────────────┘
│
▼
┌─────────────────────────────────────────────┐
│ Нужна ли агенту личность, автономия, │
│ способность принимать решения? │
└──────────────────┬──────────────────────────┘
│
┌──────────┴──────────┐
▼ ▼
┌─────────┐ ┌─────────────┐
│ ДА │ │ НЕТ │
└────┬────┘ └──────┬──────┘
│ │
▼ ▼
┌──────────────┐ ┌────────────────────┐
│ Агент- │ │ Pipeline строгий? │
│ сотрудник │ │ (A→B→C→D, │
└──────────────┘ │ порядок фикси- │
│ │ рован, JSON │
[Agent Creation │ интерфейсы) │
Standard] └──────────┬─────────┘
│
┌──────────┴──────────┐
▼ ▼
┌─────────────┐ ┌────────────────┐
│ ДА │ │ НЕТ │
└──────┬──────┘ └───────┬────────┘
│ │
▼ ▼
┌─────────────┐ ┌────────────────┐
│ Агент- │ │ Скилл агенту- │
│ инструмент │ │ сотруднику │
└─────────────┘ └────────────────┘
Агент-сотрудник, если:
Агент-инструмент, если:
Скилл агенту-сотруднику, если:
| Что | Тип | Почему |
|---|---|---|
| Research | Агент-инструмент | Pipeline A→D→B→C→E, никакой личности, JSON всюду |
| DevOps Денис | Агент-сотрудник | Личность, автономия, сам решает что и как делать |
| knowledge-search | Скилл | Легковесный поиск в 2 запроса (memory_search → wiki-search) |
| Классификация запроса | Скилл или часть инструмента | Один LLM-запрос → скилл. Если часть цепочки → инструмент |
| Роль | Кто | Что делает |
|---|---|---|
| Заказчик | CPO (Юрий), CEO (Роман), любой агент | Формулирует потребность, описывает что инструмент должен делать |
| Архитектор | CTO (Александр) | Проектирует pipeline: число шагов, стыковку OpenClaw ↔ LangGraph, JSON-интерфейсы |
| Prompt Engineer | Егор | Пишет промпты для каждого шага с AT-категоризацией |
| Python-разработчик | Илья | Пишет код StateGraph, FastAPI endpoint |
| DevOps | Денис | Разворачивает LangGraph pipeline на Server 2, Docker, Nginx, Prometheus |
| Координатор | Элис или Вера | Ведёт процесс по фазам, проверяет DoD, эскалирует блокеры |
| Тестировщик | Заказчик + координатор | Проверяет качество, принимает результат |
Заказчик ──→ [Фаза 0] Задача в OP
│
▼
Архитектор ──→ [Фаза 1] Архитектура + согласование
│
▼
Prompt Eng. ──→ [Фаза 2] Промпты + публикация в Wiki.js
│
▼
Python-dev ───→ [Фаза 3] LangGraph StateGraph + FastAPI
│
▼
DevOps ───────→ [Фаза 4] Развёртывание на Server 2
│
▼
Координатор ──→ [Фаза 5] Тестирование + приёмка
│
▼
Все ──────────→ [Фаза 6] Поддержка и мониторинг
Что происходит: Заказчик осознаёт потребность в инструменте и формулирует её.
Кто делает: Заказчик (Юрий, Роман, агент).
Действия:
DoD:
Что происходит: Архитектор проектирует pipeline.
Кто делает: Александр (CTO).
Действия:
Архитектурные принципы (из #337 и #330):
DoD:
Что происходит: Prompt Engineer пишет промпты для каждого шага pipeline.
Кто делает: Егор (Prompt Engineer).
Действия:
kb/research/prompts/{name}Категории промптов (AT-таксономия):
| AT-категория | Описание | Примеры |
|---|---|---|
| EVA-CLASS | Классификация | Prompt A (type → fact_check/deep_research/...) |
| EVA-EXTR | Извлечение, декомпозиция | Prompt D (ключевые слова, подзапросы) |
| GEN-SYNTH | Генерация с синтезом данных | Prompt E (структурированный отчёт) |
| VAL-CROSS | Валидация, cross-check | Prompt C (проверка источников) |
| SRCH-RES | Research / поиск | Prompt B (5 вариантов по типу) |
DoD:
kb/research/prompts/Что происходит: Python-разработчик пишет StateGraph и FastAPI endpoint.
Кто делает: Илья (Python-разработчик).
Действия:
/home/openclaw/langgraph/{instrument-name}/POST /invoke с JSON-входомСтруктура проекта:
/home/openclaw/langgraph/{instrument-name}/
├── main.py # FastAPI app
├── graph.py # StateGraph definition
├── nodes/
│ ├── __init__.py
│ ├── classify.py # Prompt A
│ ├── decompose.py # Prompt D
│ ├── research.py # Prompt B (5 вариантов)
│ ├── validate.py # Prompt C
│ └── synthesize.py # Prompt E
├── state.py # State schema
├── models.py # Pydantic models
├── prometheus.py # Metrics
├── config.py # Config (model, temperature, etc.)
└── requirements.txt
DoD:
POST /invokeЧто происходит: DevOps разворачивает pipeline на Server 2.
Кто делает: Денис (DevOps).
Действия:
{instrument-name}.poreklame.techИнфраструктура (из #339):
localhost:{port}/invoke/metricsDoD:
Что происходит: Координатор и заказчик проверяют качество.
Кто делает: Заказчик + координатор (Элис или Вера).
Действия:
Критерии качества (из Agent Quality Criteria — Tool):
| Параметр | Минимум | Хорошо | Отлично |
|---|---|---|---|
| Число источников | 3 | 5-10 | 10-15 |
| Confidence | 0.5 | 0.7 | 0.9+ |
| Латенция fact_check | <10s | <5s | <2s |
| Латенция deep_research | <120s | <60s | <30s |
| Формат | valid JSON | + description | + metadata |
DoD:
Что происходит: Инструмент работает, обновляется и мониторится.
Кто делает: Распределённо.
Действия по регулярной поддержке:
| Что | Кто | Периодичность |
|---|---|---|
| Обновление промптов | Prompt Engineer (Егор) | По необходимости / по запросу |
| Фикс pipeline если упал | Python-dev (Илья) | Инцидентный |
| Апдейт Docker / SSL | DevOps (Денис) | Ежемесячно / по истечению cert |
| Мониторинг Grafana | Координатор | Ежедневно (check dashboard) |
| Обновление базы источников | Все | По мере появления |
Метрики в Grafana (из #264):
Что делать при падении:
journalctl -u {instrument-name} или Docker logsDoD (поддержка):
| Характеристика | Агент-сотрудник | Агент-инструмент | Скилл (альтернатива) |
|---|---|---|---|
| Личность | ✅ Есть (SOUL, IDENTITY) | ❌ Нет. Чистый pipeline | ❌ Нет |
| Автономия | Высокая — сам решает как | Нулевая — строгий порядок шагов | Нулевая — вызван по команде |
| Качество | Вероятностное (зависит от контекста) | Гарантированное (детерминированные промпты) | Зависит от контекста агента |
| Оркестрация | OpenClaw (main session) | LangGraph StateGraph (state machine) | OpenClaw (скилл в контексте) |
| Промпты | Один system prompt на всё | Цепочка A→B→C→..., каждый под свою задачу | Один-два промпта в скилле |
| Инфраструктура | OpenClaw workspace | LangGraph на Server 2 | Читается в контекст агента |
| Вызов | sessions_spawn | HTTP POST /invoke | Через tool call |
| Размер контекста | Не ограничен | Минимальный (только JSON входа) | Ограничен размером скилла |
| Многократный вызов | Не оптимизирован | Оптимизирован (FastAPI) | Не оптимизирован |
| Метрики | Не собираются | Prometheus exporter | Не собираются |
| Сложность создания | Дни (SOUL + IDENTITY + конфиг) | Дни-недели (промпты + код + деплой) | Часы (написать SKILL.md) |
| Нагрузка на LLM | Высокая (весь контекст) | Средняя (JSON + инструкция) | Низкая (один запрос) |
| Примеры | Элис, Денис, Мила | Research pipeline | knowledge-search |
## 🎯 Цель
{Кратко — что должен делать инструмент}
## 📋 Контекст
{Зачем нужен, какая проблема решается, кто заказчик}
## 🔧 Функциональные требования
- {Требование 1}
- {Требование 2}
- ...
## 📊 Типы входа/выхода
{JSON-схема входных данных}
{JSON-схема выходных данных}
## 🚀 Ожидаемый pipeline
1. {Шаг 1} → {Что делает}
2. {Шаг 2} → {Что делает}
3. ...
## ✅ DoD
- [ ] {Критерий готовности 1}
- [ ] {Критерий готовности 2}
- ...
## ⏱ Ожидаемая латенция
- {Тип запроса}: {время}
- {Тип запроса}: {время}
## 📚 Источники (если есть)
- {Ссылка на стандарт}
- {Ссылка на кейс}
# {prompt_name} — {название на русском}
**prompt_id:** {versioned-id}
**version:** {semver}
**AT-category:** {EVA-CLASS | EVA-EXTR | GEN-SYNTH | VAL-CROSS | SRCH-RES}
**technique:** {Zero-Shot | Few-Shot | Chain-of-Thought | CONSTRAINTS}
**status:** {DRAFT | ACTIVE | DEPRECATED}
**created:** {YYYY-MM-DD}
**author:** {Prompt Engineer}
---
## ROLE
{Описание роли. Кто этот промпт, что делает.}
## GOAL
{Конкретная цель промпта. Что на выходе?}
## CONTEXT
{Контекст: в какой цепочке этот промпт, какие данные на входе.}
## {STRUCTURED_INPUT}
```json
{
"field1": "description",
"field2": "description"
}
{
"field1": "type — description",
"field2": "type — description"
}
Before outputting, verify:
json.loads())Input: {"field": "value"}
Output: {"field": "value"}
Input: {"field": "value"}
Output: {"field": "value"}
| Edge Case | Handling |
|---|---|
### 5.3. Шаблон LangGraph StateGraph
```python
"""LangGraph StateGraph для {instrument-name}"""
import json
import os
from typing import TypedDict, Optional, List
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
# ─── State Schema ──────────────────────────────────────────────────────────
class InstrumentState(TypedDict):
"""Состояние пайплайна для {instrument-name}"""
input_query: str # Исходный запрос
input_data: dict # Входные данные (JSON от OpenClaw)
step_a_output: Optional[dict] # Результат шага A
step_b_output: Optional[dict] # Результат шага B
step_c_output: Optional[dict] # Результат шага C
step_d_output: Optional[dict] # Результат шага D
step_e_output: Optional[dict] # Результат шага E
final_output: Optional[dict] # Итоговый результат
errors: List[str] # Ошибки
# ─── LLM ───────────────────────────────────────────────────────────────────
MODEL = os.getenv("LLM_MODEL", "deepseek/deepseek-chat")
llm = ChatOpenAI(
model=MODEL,
temperature=0.0,
# openai_api_base=...
)
# ─── Nodes ─────────────────────────────────────────────────────────────────
def node_a_classify(state: InstrumentState) -> dict:
"""Шаг A: Классификация типа запроса"""
# Загружаем промпт из файла или переменной
# Вызываем LLM
# Возвращаем { "step_a_output": { "type": "...", "confidence": 0.0 } }
pass
def node_d_decompose(state: InstrumentState) -> dict:
"""Шаг D: Декомпозиция запроса"""
pass
def node_b_research(state: InstrumentState) -> dict:
"""Шаг B: Поиск (выбор под-промпта по типу из step_a)"""
query_type = state["step_a_output"]["type"]
if query_type == "fact_check":
return node_b_fact_check(state)
elif query_type == "deep_research":
return node_b_deep_research(state)
# ... и т.д.
pass
def node_c_validate(state: InstrumentState) -> dict:
"""Шаг C: Валидация источников"""
pass
def node_e_synthesize(state: InstrumentState) -> dict:
"""Шаг E: Синтез ответа"""
pass
# ─── Conditional edges ──────────────────────────────────────────────────────
def route_after_a(state: InstrumentState) -> str:
"""После классификации — всегда на декомпозицию"""
return "decompose"
def route_after_b(state: InstrumentState) -> str:
"""После поиска — на валидацию"""
return "validate"
# ─── Graph ─────────────────────────────────────────────────────────────────
def build_graph() -> StateGraph:
workflow = StateGraph(InstrumentState)
# Регистрируем ноды
workflow.add_node("classify", node_a_classify)
workflow.add_node("decompose", node_d_decompose)
workflow.add_node("research", node_b_research)
workflow.add_node("validate", node_c_validate)
workflow.add_node("synthesize", node_e_synthesize)
# Регистрируем edges
workflow.set_entry_point("classify")
workflow.add_edge("classify", "decompose")
workflow.add_edge("decompose", "research")
workflow.add_edge("research", "validate")
workflow.add_edge("validate", "synthesize")
workflow.add_edge("synthesize", END)
return workflow.compile()
# ─── Invoke helper ─────────────────────────────────────────────────────────
def run_pipeline(input_data: dict) -> dict:
graph = build_graph()
result = graph.invoke({"input_data": input_data, "input_query": input_data.get("query", "")})
return result.get("final_output", {"error": "No output"})
"""FastAPI endpoint для {instrument-name}"""
import os
import time
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from prometheus_client import Counter, Histogram, generate_latest
from starlette.responses import Response
from graph import run_pipeline
app = FastAPI(title="{instrument-name}", version="1.0.0")
# ─── Models ────────────────────────────────────────────────────────────────
class InvokeRequest(BaseModel):
query: str
context: dict = {}
class InvokeResponse(BaseModel):
success: bool
result: dict = {}
error: str = ""
latency_ms: float = 0.0
# ─── Prometheus Metrics ────────────────────────────────────────────────────
REQUEST_COUNT = Counter(
"{instrument_name}_requests_total",
"Total requests",
["status"]
)
REQUEST_LATENCY = Histogram(
"{instrument_name}_request_latency_seconds",
"Request latency in seconds",
buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0, 60.0]
)
SOURCE_COUNT = Histogram(
"{instrument_name}_source_count",
"Number of sources per response",
buckets=[0, 2, 5, 10, 15, 20]
)
# ─── Endpoints ─────────────────────────────────────────────────────────────
@app.post("/invoke", response_model=InvokeResponse)
async def invoke(request: InvokeRequest):
"""Запуск пайплайна."""
start = time.time()
try:
result = run_pipeline(request.dict())
latency = time.time() - start
REQUEST_COUNT.labels(status="success").inc()
REQUEST_LATENCY.observe(latency)
if "sources" in result:
SOURCE_COUNT.observe(len(result["sources"]))
return InvokeResponse(
success=True,
result=result,
latency_ms=round(latency * 1000, 2)
)
except Exception as e:
REQUEST_COUNT.labels(status="error").inc()
latency = time.time() - start
return InvokeResponse(
success=False,
error=str(e),
latency_ms=round(latency * 1000, 2)
)
@app.get("/health")
async def health():
"""Health check."""
return {"status": "ok", "service": "{instrument-name}"}
@app.get("/metrics")
async def metrics():
"""Prometheus metrics."""
return Response(content=generate_latest(), media_type="text/plain")
# ─── Main ──────────────────────────────────────────────────────────────────
if __name__ == "__main__":
import uvicorn
port = int(os.getenv("PORT", "8000"))
uvicorn.run(app, host="0.0.0.0", port=port)
shared/company/TRUSTED_SOURCES.md| WP | Название | Статус |
|---|---|---|
| #392 | Стандарт создания агентов-инструментов | 🆕 New (текущая) |
| #260 | Эпик: Research-агент-инструмент | 🟡 In Progress |
| #337 | Архитектура Research (Александр) | ✅ Closed |
| #338 | Промпты C+D (Егор) | ✅ Closed |
| #342 | Промпты A+B+E (Егор) | ✅ Closed |
| #339 | LangGraph развёртывание (Денис) | ✅ Closed |
| #261 | Найм Python-разработчика (Катя) | ✅ Closed |
| #264 | Research метрики в Grafana | 🔴 New |
| #330 | n8n оркестратор (отвергнут) | 🆕 New |