CrewAI: подключение мультиагентного фреймворка к SpeShu.AI

export SPESHU_API_KEY="sk-ваш-ключ"
export SPESHU_MODEL=$(curl -s https://speshu.ai/api/v1/models | jq -r '.data[0].id')
echo "$SPESHU_MODEL"

import os

from crewai import LLM

llm = LLM(
    model=os.environ["SPESHU_MODEL"],
    base_url="https://speshu.ai/api/v1",
    api_key=os.environ["SPESHU_API_KEY"],
    temperature=0.2,
    timeout=300,
)

import os

from crewai import Agent, Crew, LLM, Task

llm = LLM(
    model=os.environ["SPESHU_MODEL"],
    base_url="https://speshu.ai/api/v1",
    api_key=os.environ["SPESHU_API_KEY"],
    temperature=0.2,
)

researcher = Agent(
    role="Исследователь",
    goal="Собрать проверяемые тезисы по теме и не домысливать факты",
    backstory="Ты пишешь только то, что можешь обосновать. Непроверенное помечаешь явно.",
    llm=llm,
)

writer = Agent(
    role="Редактор",
    goal="Превратить тезисы в короткий понятный отчёт",
    backstory="Ты редактор. Короткие абзацы, конкретные формулировки, без воды.",
    llm=llm,
)

research_task = Task(
    description="Составь пять тезисов о том, зачем компании нужен RAG по документам.",
    expected_output="Список из пяти коротких тезисов.",
    agent=researcher,
)

report_task = Task(
    description="На основе тезисов напиши отчёт на два абзаца.",
    expected_output="Короткий отчёт на русском языке.",
    agent=writer,
)

crew = Crew(agents=[researcher, writer], tasks=[research_task, report_task])
result = crew.kickoff()

print(result.raw)

from pydantic import BaseModel, Field

from crewai import Task


class Report(BaseModel):
    headline: str = Field(description="Заголовок отчёта")
    findings: list[str] = Field(description="Тезисы, каждый — одно предложение")
    risks: list[str] = Field(default_factory=list, description="Открытые вопросы")


report_task = Task(
    description="На основе тезисов собери отчёт.",
    expected_output="Структурированный отчёт.",
    agent=writer,
    output_pydantic=Report,
)

import os

import requests

from crewai.tools import tool


@tool("find_cheapest_model")
def find_cheapest_model() -> str:
    """Найти самую дешёвую текстовую модель в каталоге и вернуть её идентификатор."""
    catalog = requests.get(
        "https://speshu.ai/api/v1/models",
        headers={"Authorization": f"Bearer {os.environ['SPESHU_API_KEY']}"},
        timeout=30,
    ).json()["data"]
    return min(catalog, key=lambda m: float(m["cost_completion"]))["id"]

import json
import os
import time
from uuid import uuid4

import requests
from crewai.tools import tool

BASE = "https://speshu.ai/api/v1"
HEADERS = {
    "Authorization": f"Bearer {os.environ['SPESHU_API_KEY']}",
    "Content-Type": "application/json",
}
# Идентификаторы медиа-моделей стабильны, поэтому здесь допустим литерал.
IMAGE_MODEL = "nano-banana-2"


@tool("generate_image")
def generate_image(prompt: str) -> str:
    """Сгенерировать изображение по описанию и вернуть ссылку на файл."""
    created = requests.post(
        f"{BASE}/async/media/tasks",
        headers={**HEADERS, "Idempotency-Key": uuid4().hex},
        json={"model": IMAGE_MODEL, "input": {"prompt": prompt}},
        timeout=60,
    )
    created.raise_for_status()
    task_id = created.json()["data"]["taskId"]

    for _ in range(120):
        task = requests.get(
            f"{BASE}/async/media/tasks/{task_id}", headers=HEADERS, timeout=30
        ).json()["data"]
        if task["state"] == "success":
            return json.loads(task["resultJson"])["resultUrls"][0]
        if task["state"] == "fail":
            raise RuntimeError(f"{task['failCode']}: {task['failMsg']}")
        time.sleep(3)

    raise TimeoutError(task_id)

crew = Crew(agents=[researcher, writer], tasks=[research_task, report_task], stream=True)

result = crew.kickoff()
for chunk in result.stream:
    print(chunk, end="", flush=True)

import os

import httpx

# Векторы — отдельный провайдер. Это не SpeShu.AI.
EMBEDDING_BASE_URL = os.environ["EMBEDDING_BASE_URL"]
EMBEDDING_API_KEY = os.environ["EMBEDDING_API_KEY"]
EMBEDDING_MODEL = os.environ["EMBEDDING_MODEL"]

# Генерация при этом идёт через SpeShu.AI — см. LLM(...) выше.


def embed(texts: list[str]) -> list[list[float]]:
    response = httpx.post(
        f"{EMBEDDING_BASE_URL}/embeddings",
        headers={"Authorization": f"Bearer {EMBEDDING_API_KEY}"},
        json={"model": EMBEDDING_MODEL, "input": texts},
        timeout=60,
    )
    response.raise_for_status()
    return [item["embedding"] for item in response.json()["data"]]