Skip to content

Commit 2aa84fe

Browse files
Zawwarsami16claude
andcommitted
phase 11.0: brain adapter expansion + OpenAI-compat helper
new shared helper zhub/brains/_openai_compat.py: probe_openai_compat(base_url, api_key) — cheap GET /models probe stream_openai_compat(http, base, key, model, msgs, ...) — async-gen yielding ChatChunk for any OpenAI-compat endpoint, including tool_call deltas and finish_reason. migrated existing adapters to use the helper: Groq, OpenAI, Cerebras each shrinks from ~120 LOC to ~50 LOC. all 12 existing tests still green (httpx.get monkeypatching still works because the helper calls module-level httpx.get). three new adapters (~50 LOC each via the helper): TogetherAdapter — api.together.xyz/v1, Llama 3.3 70B Turbo default, env TOGETHER_API_KEY / TOGETHER_MODEL. MistralAdapter — api.mistral.ai/v1, mistral-large-latest default, env MISTRAL_API_KEY / MISTRAL_MODEL. one standalone (Cohere has its own v2 wire shape): CohereAdapter — api.cohere.com/v2/chat, command-r-plus-08-2024 default. parses newline-JSON with type discriminator (message-start / content-delta / message-end). normalizes finish_reason "complete" → "stop". env COHERE_API_KEY / COHERE_MODEL. REGISTRY now lists 8 adapters in priority order: ollama → groq → openai → cerebras → anthropic → together → mistral → cohere auto-detect picks the first whose env creds + reachability probe both pass; all four major hosted-LLM providers covered. readme: brain table + headline count updated; 8 brains listed. 164/164 pytest now (+9 new for Together/Mistral/Cohere). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1 parent aa30aff commit 2aa84fe

11 files changed

Lines changed: 536 additions & 215 deletions

README.md

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -100,7 +100,7 @@ The hub is a router. State (publisher registry, in-flight requests, rate-limit w
100100
| **`publish()`** | Turn any chat handler into an OpenAI-compat HTTPS endpoint. Auto-tunneled. `zk_` API key. Persistence-stable across restarts. |
101101
| **`connect()`** | Pair a client to one specific AI. Expose capabilities back; the AI sees them as connected tools and can invoke them. |
102102
| **`expose()`** *(Phase 7.0)* | Device registers capabilities once, **untethered to any AI**. Any AI on the hub can use them via `POST /exposures/<id>/invoke`. |
103-
| **5 brain adapters** | Ollama, Groq, OpenAI, Cerebras, Anthropic — drop-in, streaming-first. Swap brain with a flag, key stays the same. |
103+
| **8 brain adapters** | Ollama, Groq, OpenAI, Cerebras, Anthropic, Together, Mistral, Cohere — drop-in, streaming-first. Swap brain with a flag, key stays the same. |
104104
| **MCP, both directions** | Wrap a zhub AI as an MCP server (`python -m zhub.mcp_server`) for Claude Desktop. Wrap an MCP server as a zhub publisher (`examples/mcp_bridge.py`). |
105105
| **Tool calls** | OpenAI `tool_calls` auto-resolved against connected capabilities + exposures. Parallel resolution. JSON-Schema arg validation. Audit log in `usage.tool_results`. |
106106
| **Federation** | Multiple hubs peer each other. Cross-hub HTTP chat + cross-hub WebSocket connect, transparent to the client. |
@@ -118,7 +118,10 @@ zhub itself is a thin router. Brain dominates total time.
118118
|---|---|---|---|
119119
| **Cerebras Llama 405B** | ~150 ms | **2,000 tok/s** | ~$0.005 |
120120
| **Groq Llama 3.3 70B** | ~200 ms | **700+ tok/s** | ~$0.0006 *(generous free tier)* |
121+
| **Together Llama 3.3 70B Turbo** | ~250 ms | ~250 tok/s | ~$0.0006 |
122+
| **Mistral Large** | ~350 ms | ~80 tok/s | ~$0.006 |
121123
| **OpenAI gpt-4o-mini** | ~400 ms | ~80 tok/s | ~$0.0005 |
124+
| **Cohere Command-R+** | ~450 ms | ~80 tok/s | ~$0.005 |
122125
| **Anthropic Sonnet 4.5** | ~500 ms | ~100 tok/s | ~$0.011 |
123126
| **Ollama Llama 3.2 3B** *(local / $5 VPS)* | ~300 ms | 15–25 tok/s | **$0** |
124127

@@ -352,7 +355,7 @@ CI runs the Python suite on 3.10 / 3.11 / 3.12 plus the JS module test on every
352355
|---|---|
353356
| **~~4.2b~~**| True chunked tool_call delta streaming through SSE (default mode passes deltas through; `auto` mode also resolves+continues) |
354357
| **7.1** | Per-exposure access policies (whitelist of AI names / publisher keys) |
355-
| **More brains** | Cohere, Mistral, Together, Bedrock, Vertex, vLLM-direct |
358+
| ~~**More brains**~~| Phase 11.0: Together, Mistral, Cohere added (8 total). Bedrock + Vertex remain (~80 LOC each via the shared OpenAI-compat helper) |
356359
| ~~**MCP resources + prompts**~~| Phase 9.0: publishers declare `resources=` and `prompts=` in `publish()`; the MCP bridge surfaces them as resources/list, resources/read, prompts/list, prompts/get |
357360
| ~~**Hub UI dashboard**~~| Live view of connected publishers, recent requests, latency, exposed devices — at `/` (Phase 8.0) |
358361
| ~~**Latency percentiles**~~| Phase 10.0: p50/p95/p99 per AI in `/metrics` + dashboard, from a 200-sample ring buffer |

tests/test_brains_registry.py

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,8 @@ def test_registry_holds_classes_in_priority_order():
3737
The four shipped adapters land in this order: Ollama, Groq, OpenAI,
3838
Cerebras."""
3939
names = [cls.name for cls in REGISTRY]
40-
assert names == ["ollama", "groq", "openai", "cerebras", "anthropic"]
40+
assert names == ["ollama", "groq", "openai", "cerebras", "anthropic",
41+
"together", "mistral", "cohere"]
4142

4243

4344
def test_detect_returns_first_available(monkeypatch):
Lines changed: 147 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,147 @@
1+
"""Phase 11.0 — tests for Together, Mistral, Cohere brain adapters."""
2+
3+
import json
4+
from typing import Iterable
5+
6+
import httpx
7+
import pytest
8+
9+
from zhub.brains.base import ChatChunk
10+
from zhub.brains.together import TogetherAdapter
11+
from zhub.brains.mistral import MistralAdapter
12+
from zhub.brains.cohere import CohereAdapter
13+
14+
15+
class _FakeStream:
16+
def __init__(self, lines: Iterable[str]):
17+
self._lines = list(lines)
18+
19+
async def aiter_lines(self):
20+
for line in self._lines:
21+
yield line
22+
23+
async def __aenter__(self):
24+
return self
25+
26+
async def __aexit__(self, *exc):
27+
return None
28+
29+
30+
class _FakeAsyncClient:
31+
def __init__(self, lines: Iterable[str]):
32+
self._lines = list(lines)
33+
self.last_call: dict | None = None
34+
35+
def stream(self, method, url, **kw):
36+
self.last_call = {"method": method, "url": url, **kw}
37+
return _FakeStream(self._lines)
38+
39+
async def aclose(self):
40+
pass
41+
42+
43+
# ---- Together (OpenAI-compat) ------------------------------------------
44+
45+
def test_together_try_init_none_without_key(monkeypatch):
46+
monkeypatch.delenv("TOGETHER_API_KEY", raising=False)
47+
assert TogetherAdapter.try_init() is None
48+
49+
50+
def test_together_try_init_returns_adapter(monkeypatch):
51+
class R:
52+
status_code = 200
53+
monkeypatch.setenv("TOGETHER_API_KEY", "tg_test")
54+
monkeypatch.setattr(httpx, "get", lambda url, headers=None, timeout=None: R())
55+
a = TogetherAdapter.try_init()
56+
assert a is not None and a.name == "together"
57+
assert a.api_key == "tg_test"
58+
59+
60+
@pytest.mark.asyncio
61+
async def test_together_stream_uses_openai_compat():
62+
lines = [
63+
'data: {"choices":[{"delta":{"content":"hi"}}]}',
64+
'data: {"choices":[{"delta":{},"finish_reason":"stop"}]}',
65+
'data: [DONE]',
66+
]
67+
fake = _FakeAsyncClient(lines)
68+
a = TogetherAdapter(api_key="tg_test", model="meta-llama/Llama-3.3-70B-Instruct-Turbo",
69+
http=fake)
70+
out = [c async for c in a.stream([{"role": "user", "content": "x"}])]
71+
assert any(c.delta == "hi" for c in out)
72+
assert out[-1].done is True
73+
assert fake.last_call["headers"]["Authorization"] == "Bearer tg_test"
74+
75+
76+
# ---- Mistral (OpenAI-compat) -------------------------------------------
77+
78+
def test_mistral_try_init_none_without_key(monkeypatch):
79+
monkeypatch.delenv("MISTRAL_API_KEY", raising=False)
80+
assert MistralAdapter.try_init() is None
81+
82+
83+
def test_mistral_try_init_returns_adapter(monkeypatch):
84+
class R:
85+
status_code = 200
86+
monkeypatch.setenv("MISTRAL_API_KEY", "ms_test")
87+
monkeypatch.setattr(httpx, "get", lambda url, headers=None, timeout=None: R())
88+
a = MistralAdapter.try_init()
89+
assert a is not None and a.name == "mistral"
90+
91+
92+
@pytest.mark.asyncio
93+
async def test_mistral_stream_uses_openai_compat():
94+
lines = [
95+
'data: {"choices":[{"delta":{"content":"bonjour"}}]}',
96+
'data: {"choices":[{"delta":{},"finish_reason":"stop"}]}',
97+
'data: [DONE]',
98+
]
99+
fake = _FakeAsyncClient(lines)
100+
a = MistralAdapter(api_key="ms_test", model="mistral-large-latest", http=fake)
101+
out = [c async for c in a.stream([{"role": "user", "content": "x"}])]
102+
assert any(c.delta == "bonjour" for c in out)
103+
assert out[-1].done is True
104+
105+
106+
# ---- Cohere (custom v2 shape) ------------------------------------------
107+
108+
def test_cohere_try_init_none_without_key(monkeypatch):
109+
monkeypatch.delenv("COHERE_API_KEY", raising=False)
110+
assert CohereAdapter.try_init() is None
111+
112+
113+
def test_cohere_try_init_returns_adapter(monkeypatch):
114+
class R:
115+
status_code = 200
116+
monkeypatch.setenv("COHERE_API_KEY", "co_test")
117+
monkeypatch.setattr(httpx, "get", lambda url, headers=None, timeout=None: R())
118+
a = CohereAdapter.try_init()
119+
assert a is not None and a.name == "cohere"
120+
121+
122+
@pytest.mark.asyncio
123+
async def test_cohere_stream_parses_v2_event_shape():
124+
"""Cohere v2 streams newline-JSON with type discriminator."""
125+
lines = [
126+
json.dumps({"type": "message-start", "id": "m_x"}),
127+
json.dumps({"type": "content-delta",
128+
"delta": {"message": {"content": {"text": "Howdy "}}}}),
129+
json.dumps({"type": "content-delta",
130+
"delta": {"message": {"content": {"text": "partner"}}}}),
131+
json.dumps({"type": "message-end",
132+
"delta": {"finish_reason": "complete"}}),
133+
]
134+
fake = _FakeAsyncClient(lines)
135+
a = CohereAdapter(api_key="co_test", model="command-r-plus-08-2024",
136+
http=fake)
137+
out = [c async for c in a.stream([{"role": "user", "content": "x"}])]
138+
deltas = [c.delta for c in out if c.delta]
139+
assert "".join(deltas) == "Howdy partner"
140+
assert out[-1].done is True
141+
# `complete` normalized to `stop`
142+
assert out[-1].finish_reason == "stop"
143+
body = fake.last_call["json"]
144+
assert body["model"] == "command-r-plus-08-2024"
145+
assert body["messages"][-1] == {"role": "user", "content": "x"}
146+
headers = fake.last_call["headers"]
147+
assert headers["Authorization"] == "Bearer co_test"

zhub/brains/__init__.py

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,9 @@
2121
from .openai import OpenAIAdapter
2222
from .cerebras import CerebrasAdapter
2323
from .anthropic import AnthropicAdapter
24+
from .together import TogetherAdapter
25+
from .mistral import MistralAdapter
26+
from .cohere import CohereAdapter
2427

2528

2629
REGISTRY: list[type[BrainAdapter]] = [
@@ -29,6 +32,9 @@
2932
OpenAIAdapter,
3033
CerebrasAdapter,
3134
AnthropicAdapter,
35+
TogetherAdapter,
36+
MistralAdapter,
37+
CohereAdapter,
3238
]
3339

3440

@@ -59,4 +65,5 @@ def list_available() -> list[BrainAdapter]:
5965
"detect", "list_available",
6066
"OllamaAdapter", "GroqAdapter", "OpenAIAdapter",
6167
"CerebrasAdapter", "AnthropicAdapter",
68+
"TogetherAdapter", "MistralAdapter", "CohereAdapter",
6269
]

zhub/brains/_openai_compat.py

Lines changed: 103 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,103 @@
1+
"""Shared streaming + probe logic for OpenAI-compatible brain adapters.
2+
3+
OpenAI's chat completions wire format is the de facto standard — Groq,
4+
OpenAI itself, Cerebras, Together, Mistral, and many self-hosted servers
5+
(vLLM, LM Studio, llama.cpp) all speak it identically. This module factors
6+
the common parts so each provider's adapter is a thin shell.
7+
"""
8+
9+
from __future__ import annotations
10+
11+
import json
12+
import os
13+
from typing import Any, AsyncIterator, Optional
14+
15+
import httpx
16+
17+
from .base import BrainAdapter, ChatChunk
18+
19+
20+
_PROBE_TIMEOUT = 1.5
21+
22+
23+
def probe_openai_compat(base_url: str, api_key: str) -> bool:
24+
"""Cheap reachability probe — GET /models with bearer key."""
25+
try:
26+
r = httpx.get(
27+
f"{base_url.rstrip('/')}/models",
28+
headers={"Authorization": f"Bearer {api_key}"},
29+
timeout=_PROBE_TIMEOUT,
30+
)
31+
return r.status_code == 200
32+
except Exception:
33+
return False
34+
35+
36+
async def stream_openai_compat(
37+
http: httpx.AsyncClient,
38+
base_url: str,
39+
api_key: str,
40+
model: str,
41+
messages: list[dict[str, Any]],
42+
*,
43+
system: Optional[str] = None,
44+
temperature: float = 0.7,
45+
max_tokens: int = 2048,
46+
tools: Optional[list[dict[str, Any]]] = None,
47+
extra_headers: Optional[dict[str, str]] = None,
48+
) -> AsyncIterator[ChatChunk]:
49+
"""Stream chat completions from any OpenAI-compatible endpoint, parsing
50+
the SSE shape `data: {"choices":[{"delta":{...}, "finish_reason":...}]}`
51+
terminated by `data: [DONE]`. Surfaces text deltas, tool_call deltas,
52+
and a final done chunk with finish_reason."""
53+
msgs: list[dict[str, Any]] = []
54+
if system:
55+
msgs.append({"role": "system", "content": system})
56+
msgs.extend(messages)
57+
body: dict[str, Any] = {
58+
"model": model,
59+
"messages": msgs,
60+
"stream": True,
61+
"temperature": temperature,
62+
"max_tokens": max_tokens,
63+
}
64+
if tools:
65+
body["tools"] = tools
66+
headers = {
67+
"Authorization": f"Bearer {api_key}",
68+
"Content-Type": "application/json",
69+
}
70+
if extra_headers:
71+
headers.update(extra_headers)
72+
async with http.stream(
73+
"POST", f"{base_url.rstrip('/')}/chat/completions",
74+
json=body, headers=headers,
75+
) as response:
76+
async for line in response.aiter_lines():
77+
line = line.rstrip("\r")
78+
if not line.startswith("data:"):
79+
continue
80+
payload = line[5:].strip()
81+
if payload == "[DONE]":
82+
yield ChatChunk(delta="", done=True, finish_reason="stop")
83+
return
84+
try:
85+
data = json.loads(payload)
86+
except json.JSONDecodeError:
87+
continue
88+
choices = data.get("choices") or []
89+
if not choices:
90+
continue
91+
choice = choices[0]
92+
delta_obj = choice.get("delta") or {}
93+
content = delta_obj.get("content") or ""
94+
finish = choice.get("finish_reason")
95+
for tcd in delta_obj.get("tool_calls") or []:
96+
yield ChatChunk(delta="", done=False,
97+
tool_call_delta=tcd, raw=data)
98+
if content:
99+
yield ChatChunk(delta=content, done=False, raw=data)
100+
if finish:
101+
yield ChatChunk(delta="", done=True,
102+
finish_reason=finish, raw=data)
103+
return

0 commit comments

Comments
 (0)