Metadata-Version: 2.5
Name: river-client
Version: 0.14.1
Summary: Python client for River ML training API
Project-URL: Homepage, https://river.ai
Author-email: River AI <api@river.ai>
License-Expression: Apache-2.0
License-File: LICENSE
Keywords: fine-tuning,grpc,llm,lora,reinforcement-learning,river,training
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Requires-Python: >=3.12
Requires-Dist: grpcio>=1.80.0
Requires-Dist: huggingface-hub>=0.36.0
Requires-Dist: jinja2>=3.1.0
Requires-Dist: jsonschema>=4.26.0
Requires-Dist: numpy>=2.0.0
Requires-Dist: protobuf>=6.31.1
Requires-Dist: referencing>=0.37.0
Requires-Dist: tiktoken>=0.7.0
Requires-Dist: transformers>=5.3.0
Requires-Dist: xxhash>=3.0.0
Provides-Extra: custom-loss
Requires-Dist: sympy; extra == 'custom-loss'
Requires-Dist: torch<2.12,>=2.4; extra == 'custom-loss'
Provides-Extra: examples
Requires-Dist: datasets>=2.14.0; extra == 'examples'
Requires-Dist: pydantic<3,>=2.12; extra == 'examples'
Requires-Dist: wandb>=0.28.0; extra == 'examples'
Provides-Extra: hf-export
Requires-Dist: safetensors>=0.5; extra == 'hf-export'
Requires-Dist: torch>=2.6; extra == 'hf-export'
Description-Content-Type: text/markdown

# river-client

Python client for the [River](https://river.ai) ML training API — sampling,
LoRA and full-weight fine-tuning, and reinforcement learning against River-hosted models.

## Installation

```bash
pip install river-client
```

Requires Python 3.12+.

## Quick start

River provides a low-level API for sampling, training, and checkpointing, plus
an integrated [RL library](#reinforcement-learning) that manages rollouts and
training from your environment and reward function. The example below uses the
low-level API.

```python
import river_client as river

client = river.Client(api_key="your-key", endpoint="api.river.ai")

# Stateless sampling from a base model
samples = client.sample(
    "What is 2+2?",
    base_model="Qwen/Qwen3.6-35B-A3B-FP8",
    max_tokens=50,
)
print(samples[0].text)

# Training with a session
with client.session() as session:
    model = session.create_model(
        base_model="Qwen/Qwen3.6-35B-A3B-FP8",
        lora=river.LoraConfig(rank=16),
    )

    # Forward + backward, then an optimizer step
    result = model.forward_backward(data, loss_fn="cross_entropy")
    model.optim_step(lr=1e-4)

    # Sample from the current weights
    sample_groups = model.sample("Continue:", max_tokens=100)
```

## Capacity hints

Capacity hints are disabled by default. Enable them with
`client.session(capacity_hints=True)` to print capacity changes to stderr:
workers preparing, additional workers starting, or waits for available capacity.
A state must persist for five seconds before a hint appears. Each run and workload
prints at most once per 30 seconds, with no repeated messages for an unchanged
state. A ready message
appears only after a capacity hint was shown.

Structured status remains available with printing disabled:

```python
with client.session() as session:
    # Create models and run your job as usual. In your progress update:
    for status in session.capacity_status:
        print(status.training_run_id, status.workload, status.state)
```

`session.capacity_status` returns an immutable tuple of `river.CapacityStatus`
objects from the latest heartbeat without an extra API call. Each object contains
`training_run_id`, `workload` (`training` or `inference`), `state`, `capacity_wait`,
and `observed_at` (a UTC datetime when available). States are `ready`, `scaling_up`,
`preparing`, `waiting`, `at_limit`, and `unavailable`.

Observations expire after 30 seconds. An empty tuple or `unavailable` means no
fresh status is available; neither means capacity is ready. A scaling limit alone
does not imply queuing, and workload capacity pressure does not explain every
individual request's latency.

These observations require a server that supplies heartbeat capacity status.
They cover the session's training runs and their checkpoint inference, matching
Console. Stateless base-model sampling is outside this scope. Set
`session.capacity_hints = True` to enable hints in an already-running session,
or set it to `False` to silence them.

## Full-weight training

On an endpoint with the GLM 5.3 Flash BF16 pool enabled, select
`full_weight=True` instead of `lora`:

```python
model = session.create_model(
    base_model="zai-org/GLM-5.3-Flash-BF16",
    full_weight=True,
)
model.forward_backward(data, loss_fn="cross_entropy")
model.optim_step(lr=1e-5, beta1=0.9, beta2=0.95, weight_decay=0.01)
checkpoint = model.save_weights("step1", mode="training", immutable=True, timeout=3600)
model.load_weights(checkpoint, load_optimizer=True, timeout=3600)
```

Checkpoints include FP32 master weights, Adam moments, optimizer parameters,
and the training step. Full-weight saves require `immutable=True`; repeated
names create separate checkpoint URIs with independent retention. Only an RL
model such as `zai-org/GLM-5.3-Flash-FW-RL` can sample: its run owns a sampler,
and `river_client.rl` refreshes it with `save_weights(mode="publish")`.
See the [full-weight training guide](../../docs/river-api-design-docs/full-weight-training.md)
for RL, resume semantics and deployment requirements.

For temporary sampling or training checkpoints, use `durable=False`:

```python
checkpoint = model.save_weights("sampling", mode="inference", durable=False)
```

These checkpoints stay on shared NFS for 24 hours by default and never upload
to cloud storage. Pass `ttl=datetime.timedelta(...)` to choose another lifetime
within the server's retention cap. They cannot be restored from cloud storage
after expiry or NFS loss. This option requires server support; the SDK rejects
it on an endpoint without `temporary_checkpoints_v1`. Changing an existing
checkpoint's storage policy requires a new name or `immutable=True`. The RL
trainer uses it for per-step sampling snapshots. Recovery and evaluation saves
remain durable so unfinished evaluations can resume after a long pause. Update the server before using the new RL client.
Custom model implementations must accept the `durable` keyword on `save_weights`.

On trainer resume, unfinished rollouts pinned to temporary snapshots restart
from their environment's initial state, even after a short pause. Recovery
republishes only the current policy; older pins may have expired. Completed
rollout data is preserved, and each restarted rollout selects a fresh policy
without mixing policies within a trajectory.


## Reinforcement learning

`river_client.rl` manages rollouts and training around your environment, tools,
and reward function. It supports multi-turn conversations, images, KV-cache
reuse, and synchronous or bounded asynchronous training.

```python
import operator
import os
from typing import Literal

import river_client as river
from river_client import rl
from river_client.renderers import get_renderer, get_text_content

# Tool schemas come from type hints and docstrings; execution stays in your code.
@rl.tool
async def calculator(a: float, operation: Literal["+", "-", "*", "/"], b: float) -> str:
    """Calculate the result of an arithmetic operation on two numbers."""
    operations = {"+": operator.add, "-": operator.sub, "*": operator.mul, "/": operator.truediv}
    return f"{operations[operation](a, b):g}"

class AnswerEnv(rl.Env):
    tools = [calculator]  # The engine runs tool calls and feeds results back.

    async def reset(self, row):
        return [{"role": "user", "content": (
            row["question"] + " Use the calculator if helpful. Return only the answer."
        )}]

    async def reward(self, traj, row):
        # Score the final answer, excluding the model's reasoning.
        answer = get_text_content(traj.messages[-1]).strip()
        return float(answer == row["answer"])

# Replace this toy dataset and reward with your task.
dataset = [
    {"question": "What is 17 * 23?", "answer": "391"},
    {"question": "What is 144 / 12?", "answer": "12"},
]
base_model = "zai-org/GLM-5.3-Flash"
renderer = get_renderer(base_model)
client = river.Client(api_key=os.environ["RIVER_API_KEY"], endpoint="api.river.ai")

with client.session() as session:
    model = session.create_model(
        base_model=base_model,
        tokenizer=renderer.tokenizer,
        lora=river.LoraConfig(rank=16),
    )
    trainer = rl.AsyncTrainer(
        engine=rl.RolloutEngine(
            model, env=AnswerEnv, renderer=renderer,
            budget=rl.Budget(max_turns=4, max_generated_tokens=2048),
            # Sampling stays on one policy for the whole trajectory by default.
        ),
        optimizer=rl.Adam(lr=1e-5),
        advantage=rl.GroupCentered(),  # Center rewards within each prompt group.
        completion=rl.GroupCompletion(mode="wait"),
        normalize="token",
        groups_per_step=2, group_size=8,  # Two prompts, eight rollouts each.
        max_staleness=0,  # Synchronous; >0 allows bounded sampling ahead.
    )
    rl.run(trainer, dataset, steps=10, on_step=lambda step: print(step.n, step.metrics))
```

See the [bundled agent skill](#ai-agent-skill) for image observations, cache
policies, checkpoint/resume, and evaluation.

## Dedicated streaming inference

**Gated feature — disabled by default.** Contact River to enable dedicated
deployments for your team and the checkpoint's base model before using these
APIs. Use a team API key with that access; personal API keys cannot create
deployments.

`client.create_deployment(checkpoint, ...)` provisions capacity for a
checkpoint and returns a base URL that the standard OpenAI client streams
from unchanged; `list_deployments`, `get_deployment_usage`, `scale_on_target`
and `delete_deployment` manage it from there. See the
[deployment walkthrough](https://docs.river.ai/guides/deployments/)
for a complete example. The bundled agent skill below
carries the full workflow: replica roles, scale-to-zero and resume, streaming
error handling, and usage accounting.

## AI agent skill

The package bundles an agent skill — a `SKILL.md` that teaches AI coding
agents (Claude Code and compatible tools) the current training API:
`train_step` semantics, data formats, RL/SFT/distillation loop patterns,
image uploads and handles, and fault-tolerant auto-recovery. Because it ships
inside the wheel, the skill always matches the installed client version.

Install it into your agent's skills directory:

```bash
python -m river_client.skill --install
```

This copies the skill into `~/.claude/skills/`; pass `--dest` for a
different location (e.g. a project's `.claude/skills/`). Run without
`--install` to print the bundled skill's path instead.

## License

Apache-2.0. See [`LICENSE`](LICENSE).

## Export downloaded checkpoints to Hugging Face

The converter is available in `river-client` 0.13.0 and later. In Console, open
an inference checkpoint and choose **Download HF export script**.
Install the script's dependencies and run it with your API key:

```bash
pip install -U "river-client[hf-export]>=0.13.0" requests
RIVER_API_KEY=... python download.py
```

The script downloads to a temporary directory and writes an HF/PEFT LoRA adapter
under the checkpoint ID. Load that directory with PEFT and the corresponding
base model. Conversion reads the base model's safetensors headers from Hugging
Face; it does not download base weights. Set `HF_TOKEN` for gated base models.

For a checkpoint already on disk, install the export extra and run the converter:

```bash
pip install -U "river-client[hf-export]>=0.13.0"
python -m river_client.hf_export ./river-checkpoint ./hf-adapter
```

The source directory must contain `adapter_config.json` and either
`adapter_model.safetensors` or its sharded index and referenced files. The output
contains `adapter_config.json` and `adapter_model.safetensors` in HF/PEFT adapter
format. The converter uses `base_model_name_or_path` from the adapter
configuration unless you override it:

```bash
python -m river_client.hf_export ./river-checkpoint ./hf-adapter \
  --base-model "<base-model-id>" --revision "<base-model-commit>"
```

Use `--base-model /path/to/local/base` to read a local base model's safetensors
headers. In Python, call the same converter directly:

```python
from river_client.hf_export import export_hf_adapter

output = export_hf_adapter("./river-checkpoint", "./hf-adapter")
# Optional keyword arguments: base_model="...", revision="...".
print(output)
```

The converter checks every target name and tensor shape against the base model.
It restores text-model prefixes, splits supported fused projections, preserves
rank/alpha scaling, and combines indexed adapter shards into PEFT's single
safetensors file. It runs on CPU with memory proportional to adapter size.
The source is preserved, and an existing output directory is never overwritten.
Unsupported layouts fail before an output directory is published. This exports
LoRA adapters, not merged base weights or training/optimizer state.

Conversion preserves adapter values and dtype; inference compatibility also
depends on the loader. In our tests with Transformers 5.8.1 and PEFT 0.18.1,
loading BF16 adapters onto native FP8 layers cast the adapter weights to FP8,
changing their values. Use a loader that preserves BF16 adapters alongside the
FP8 base. The workaround verified in that test was to load the base with
`FineGrainedFP8Config(dequantize=True)` before attaching the adapter. This
requires more memory and does not quantize the adapter.

For unsupported layouts, including NVFP4 expert adapters, choose **Download raw
script** in Console. Run `RIVER_API_KEY=... python download-raw.py` after installing
`requests`. It saves the original files under `<checkpoint-id>-raw` without
conversion; those files are not guaranteed to load in HF/PEFT.
