LangGraph: stateful-агент на моделях SpeShu.AI

export SPESHU_API_KEY="sk-ваш-ключ"

export SPESHU_MODEL=$(curl -s https://speshu.ai/api/v1/models \
  -H "Authorization: Bearer $SPESHU_API_KEY" \
  | jq -r '.data[0].id')
echo "$SPESHU_MODEL"

import os

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model=os.environ["SPESHU_MODEL"],
    base_url="https://speshu.ai/api/v1",
    api_key=os.environ["SPESHU_API_KEY"],
    timeout=300,       # потолок запроса на стороне API — 300 секунд
    max_retries=2,      # повторы на 429 и 5xx
)

import os

from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import START, MessagesState, StateGraph

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


def call_model(state: MessagesState) -> dict:
    return {"messages": [llm.invoke(state["messages"])]}


builder = StateGraph(MessagesState)
builder.add_node("assistant", call_model)
builder.add_edge(START, "assistant")

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "demo-user"}}

graph.invoke({"messages": [{"role": "user", "content": "Меня зовут Анна."}]}, config)
result = graph.invoke(
    {"messages": [{"role": "user", "content": "Как меня зовут?"}]}, config
)

print(result["messages"][-1].content)

# тот же файл, что и в предыдущем примере: импорты и клиент llm уже созданы
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition


@tool
def get_weather(city: str) -> str:
    """Текущая погода в городе. Возвращает температуру в градусах Цельсия."""
    return f"{city}: 14 °C, облачно"


tools = [get_weather]
llm = llm.bind_tools(tools)


def call_model(state: MessagesState) -> dict:
    return {"messages": [llm.invoke(state["messages"])]}


builder = StateGraph(MessagesState)
builder.add_node("agent", call_model)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)
builder.add_edge("tools", "agent")
graph = builder.compile(checkpointer=InMemorySaver())

result = graph.invoke(
    {"messages": [{"role": "user", "content": "Какая погода в Казани?"}]},
    {"configurable": {"thread_id": "demo-user"}},
)
print(result["messages"][-1].content)

config = {"configurable": {"thread_id": "demo-user"}}

for chunk, meta in graph.stream(
    {"messages": [{"role": "user", "content": "Напиши три абзаца про океан."}]},
    config,
    stream_mode="messages",
):
    print(chunk.content, end="", flush=True)

import os

from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field


class Route(BaseModel):
    intent: str = Field(description="Намерение пользователя")
    urgency: int = Field(description="Срочность от 1 до 5")


router_llm = ChatOpenAI(
    model=os.environ["SPESHU_MODEL"],
    base_url="https://speshu.ai/api/v1",
    api_key=os.environ["SPESHU_API_KEY"],
    timeout=300,
).with_structured_output(Route)

route = router_llm.invoke("Верните деньги за заказ, он не пришёл")
print(route.intent, route.urgency)

import json
import os
import time
import uuid

import requests

BASE = "https://speshu.ai/api/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['SPESHU_API_KEY']}"}


def generate_image(prompt: str) -> str:
    created = requests.post(
        f"{BASE}/async/media/tasks",
        headers={
            **HEADERS,
            "Content-Type": "application/json",
            "Idempotency-Key": str(uuid.uuid4()),
        },
        json={"model": "nano-banana-2", "input": {"prompt": prompt}},
        timeout=60,
    )
    created.raise_for_status()
    task_id = created.json()["data"]["taskId"]

    while True:
        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)

import os

from langchain_openai import OpenAIEmbeddings

# Отдельный провайдер эмбеддингов — не SpeShu.AI.
embeddings = OpenAIEmbeddings(
    model=os.environ["EMBEDDING_MODEL"],
    base_url=os.environ["EMBEDDING_BASE_URL"],
    api_key=os.environ["EMBEDDING_API_KEY"],
)

# Генерация остаётся на SpeShu.AI — отдельный клиент из раздела «Настройка клиента».