Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 27 additions & 8 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -78,23 +78,22 @@ res = interfaze.chat.completions.create(

## Streaming

For live rendering, iterate `text_deltas()` — it yields visible text only, stripping Interfaze's
inline `<think>`/`<precontext>` side-channels:
`.stream()` yields OpenAI-style events (`content.delta`, `content.done`, tool-call events, …) with
Interfaze's inline `<think>`/`<precontext>` side-channels stripped from the content events:

```python
stream = interfaze.chat.completions.stream(
messages=[{"role": "user", "content": "Tell me a story."}],
)
for text in stream.text_deltas():
print(text, end="")
for event in stream:
if event.type == "content.delta":
print(event.delta, end="")
final = stream.get_final_completion()
print(final.reasoning, final.precontext)
```

> Iterating the stream directly (`for chunk in stream`) or the plain `create(stream=True)` path
> yields **raw** chunks whose `delta.content` still contains the `<think>`/`<precontext>` tags — use
> `text_deltas()` for anything user-facing. `.stream()` also accumulates and surfaces
> `reasoning`/`precontext` on `get_final_completion()`.
> Just want clean tokens? `stream.text_deltas()` yields visible text only. Plain
> `create(stream=True)` returns the raw chunk iterator (side-channel tags **not** stripped).

## Inputs

Expand Down Expand Up @@ -152,6 +151,26 @@ precontext with `ChatInterfaze(precontext=[...])`.
intermediary mid-job.
- The underlying OpenAI client is available at `interfaze.openai`.

## Compatibility notes

Drop-in for the OpenAI **chat completions** flow (`create`, response types, errors,
`create(stream=True)`, `models`), with a few behaviors worth knowing when migrating:

- **`.stream()` yields OpenAI-style events**, drop-in with OpenAI's streaming helper — iterate
`event.type` (`"content.delta"` with `.delta`, `"content.done"`,
`"tool_calls.function.arguments.delta"`/`.done`, …), same as `client.chat.completions.stream()`
on the OpenAI SDK. `create(stream=True)` still gives the raw `ChatCompletionChunk` iterator, and
`stream.text_deltas()` gives just the clean visible text if you don't need events.
- **Returned text is lightly post-processed.** `json_object` content is unwrapped from its
```` ```json ```` fence, and streamed `<think>`/`<precontext>` side-channels are pulled into
`reasoning`/`precontext`, so `message.content` may not be byte-identical to the raw wire response.
- **`inputs.*` accept https URLs** (Interfaze fetches them server-side). Those parts are valid for
Interfaze but **not** portable to OpenAI/Azure, which require base64 in `file`/`input_audio` parts.
- **Escape hatch:** anything not on the wrapper — `chat.completions.parse()`, `.with_raw_response`,
`.with_streaming_response` — is on the underlying client at `interfaze.openai`.
- **`tasks.*` return the extracted result** (a `dict`/`list`/`str`), not a `ChatCompletion` — e.g.
`tasks.ocr(...)` returns the OCR dict directly.

## License

MIT
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ dependencies = ["openai>=2,<3"]

[project.urls]
Homepage = "https://interfaze.ai"
Repository = "https://github.com/interfaze/interfaze-py"
Repository = "https://github.com/InterfazeAI/interfaze-python"

[project.optional-dependencies]
langchain = ["langchain-openai>=1.3.5; python_version>='3.10'"]
Expand Down
66 changes: 64 additions & 2 deletions src/interfaze/_chat.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from typing import Any, Dict, Iterable, List, Literal, Optional, Union, cast, overload

from openai import AsyncOpenAI, AsyncStream, OpenAI, Stream
from openai.types.chat import ChatCompletion, ChatCompletionChunk
from openai.types.chat import ChatCompletion, ChatCompletionChunk, ParsedChatCompletion

from ._constants import INTERFAZE_MODEL
from ._errors import InterfazeError
Expand Down Expand Up @@ -63,7 +63,11 @@ def to_interfaze(raw: ChatCompletion, strip_fence: bool) -> InterfazeChatComplet
msg["content"] = strip_json_fence(msg["content"])
except (KeyError, IndexError, TypeError):
pass
return InterfazeChatCompletion.model_validate(data)
result = InterfazeChatCompletion.model_validate(data)
request_id = getattr(raw, "_request_id", None)
if request_id is not None:
result._request_id = request_id
return result


class _CompletionsBase:
Expand Down Expand Up @@ -146,6 +150,35 @@ def stream(
kw, strip = self._kwargs(messages, model, task, guard, response_format, kwargs)
return InterfazeStream(self._client, kw, strip)

def parse(
self,
*,
messages: Iterable[Any],
response_format: Any,
model: str = INTERFAZE_MODEL,
guard: Optional[List[GuardCode]] = None,
**kwargs: Any,
) -> "ParsedChatCompletion[Any]":
"""Structured output via a Pydantic model (delegates to the OpenAI client's ``parse``).

Interfaze extras (``vcache``/``precontext``) are present but untyped here; use ``create`` for
the typed extended completion.
"""
msgs = prepare(messages, model, None, guard, None)[0]
return self._client.chat.completions.parse(
model=model, messages=msgs, response_format=response_format, **kwargs
)

@property
def with_raw_response(self) -> Any:
"""Raw-HTTP escape hatch (delegates to the OpenAI client; bypasses task/guard preprocessing)."""
return self._client.chat.completions.with_raw_response

@property
def with_streaming_response(self) -> Any:
"""Streaming raw-HTTP escape hatch (delegates to the OpenAI client)."""
return self._client.chat.completions.with_streaming_response


class AsyncCompletions(_CompletionsBase):
"""`interfaze.chat.completions` (async)."""
Expand Down Expand Up @@ -211,6 +244,35 @@ def stream(
kw, strip = self._kwargs(messages, model, task, guard, response_format, kwargs)
return AsyncInterfazeStream(self._client, kw, strip)

async def parse(
self,
*,
messages: Iterable[Any],
response_format: Any,
model: str = INTERFAZE_MODEL,
guard: Optional[List[GuardCode]] = None,
**kwargs: Any,
) -> "ParsedChatCompletion[Any]":
"""Structured output via a Pydantic model (delegates to the OpenAI client's ``parse``).

Interfaze extras (``vcache``/``precontext``) are present but untyped here; use ``create`` for
the typed extended completion.
"""
msgs = prepare(messages, model, None, guard, None)[0]
return await self._client.chat.completions.parse(
model=model, messages=msgs, response_format=response_format, **kwargs
)

@property
def with_raw_response(self) -> Any:
"""Raw-HTTP escape hatch (delegates to the OpenAI client; bypasses task/guard preprocessing)."""
return self._client.chat.completions.with_raw_response

@property
def with_streaming_response(self) -> Any:
"""Streaming raw-HTTP escape hatch (delegates to the OpenAI client)."""
return self._client.chat.completions.with_streaming_response


class Chat:
def __init__(self, client: OpenAI) -> None:
Expand Down
105 changes: 65 additions & 40 deletions src/interfaze/_stream.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
from typing import Any, AsyncIterator, Dict, Iterator, List, Optional, Tuple

from openai import AsyncOpenAI, OpenAI
from openai.lib.streaming.chat import ChatCompletionStreamEvent, ChatCompletionStreamState
from openai.types.chat import ChatCompletionChunk

from ._errors import InterfazeError
Expand Down Expand Up @@ -53,10 +54,7 @@ def _suffix_prefix_len(s: str, tag: str) -> int:


class _SideChannelFilter:
"""Strip inline ``<think>``/``<precontext>`` blocks from streamed content, chunk by chunk.

Buffers a trailing partial that may be a split tag; never withholds text that cannot be a tag.
"""
"""Incrementally strips ``<think>``/``<precontext>`` blocks, buffering tags split across chunks."""

def __init__(self) -> None:
self._buf = ""
Expand Down Expand Up @@ -102,6 +100,26 @@ def flush(self) -> str:
return rest


def _clean_for_events(
chunk: ChatCompletionChunk, filt: _SideChannelFilter, role_injected: bool
) -> "Tuple[ChatCompletionChunk, bool]":
"""Clean a chunk for openai's event state: strip side channels from content, inject a role."""
cleaned = chunk.model_copy(deep=True)
if not cleaned.choices:
return cleaned, role_injected
delta = cleaned.choices[0].delta
if not role_injected:
role_injected = True
if not delta.role:
delta.role = "assistant"
piece = filt.feed(delta.content) if isinstance(delta.content, str) else ""
if cleaned.choices[0].finish_reason:
piece += filt.flush()
if isinstance(delta.content, str) or piece:
delta.content = piece
return cleaned, role_injected


class _State:
def __init__(self) -> None:
self.content = ""
Expand All @@ -111,6 +129,8 @@ def __init__(self) -> None:
self.model = ""
self.created = 0
self.tool_calls: Dict[int, Dict[str, str]] = {}
self.usage: Any = None
self.system_fingerprint: Optional[str] = None

def accumulate(self, chunk: ChatCompletionChunk) -> None:
if not self.id and chunk.id:
Expand All @@ -119,6 +139,10 @@ def accumulate(self, chunk: ChatCompletionChunk) -> None:
self.model = chunk.model
if not self.created and chunk.created:
self.created = chunk.created
if chunk.usage:
self.usage = chunk.usage
if chunk.system_fingerprint:
self.system_fingerprint = chunk.system_fingerprint
if not chunk.choices:
return
choice = chunk.choices[0]
Expand Down Expand Up @@ -159,6 +183,10 @@ def build(self, strip_fence: bool = False) -> InterfazeChatCompletion:
],
"vcache": False,
}
if self.usage is not None:
data["usage"] = self.usage.model_dump()
if self.system_fingerprint:
data["system_fingerprint"] = self.system_fingerprint
if reasoning:
data["reasoning"] = reasoning
if precontext:
Expand All @@ -167,13 +195,16 @@ def build(self, strip_fence: bool = False) -> InterfazeChatCompletion:


class InterfazeStream:
"""Sync streaming helper — iterate chunks, then ``get_final_completion()``."""
"""Sync streaming helper — iterate OpenAI-style events, then ``get_final_completion()``."""

def __init__(self, client: OpenAI, kwargs: Dict[str, Any], strip_fence: bool = False) -> None:
self._client = client
self._kwargs = kwargs
self._strip_fence = strip_fence
self._state = _State()
self._openai_state: ChatCompletionStreamState[None] = ChatCompletionStreamState()
self._filter = _SideChannelFilter()
self._role_injected = False
self._started = False
self._done = False

Expand All @@ -183,36 +214,31 @@ def __enter__(self) -> "InterfazeStream":
def __exit__(self, *exc: Any) -> None:
return None

def __iter__(self) -> "Iterator[ChatCompletionChunk]":
def __iter__(self) -> "Iterator[ChatCompletionStreamEvent[None]]":
if self._started:
raise InterfazeError("This stream has already been consumed.")
self._started = True
for chunk in self._client.chat.completions.create(stream=True, **self._kwargs):
self._state.accumulate(chunk)
yield chunk
cleaned, self._role_injected = _clean_for_events(chunk, self._filter, self._role_injected)
yield from self._openai_state.handle_chunk(cleaned)
self._done = True

def text_deltas(self) -> "Iterator[str]":
"""Yield visible text only, stripping ``<think>``/``<precontext>`` across chunk boundaries.

Use this (not raw ``create(stream=True)`` deltas) for live rendering. ``reasoning`` and
``precontext`` remain available on ``get_final_completion()``.
"""
"""Iterate only the visible text (no side channels, no events) — for plain-token consumers."""
if self._started:
raise InterfazeError("This stream has already been consumed.")
self._started = True
filt = _SideChannelFilter()
for chunk in self._client.chat.completions.create(stream=True, **self._kwargs):
self._state.accumulate(chunk)
if chunk.choices:
delta = chunk.choices[0].delta
if delta and isinstance(delta.content, str) and delta.content:
visible = filt.feed(delta.content)
if visible:
yield visible
tail = filt.flush()
if tail:
yield tail
if not chunk.choices:
continue
delta = chunk.choices[0].delta
piece = self._filter.feed(delta.content) if isinstance(delta.content, str) else ""
if chunk.choices[0].finish_reason:
piece += self._filter.flush()
if piece:
yield piece
self._done = True

@property
Expand All @@ -234,13 +260,16 @@ def get_final_completion(self) -> InterfazeChatCompletion:


class AsyncInterfazeStream:
"""Async streaming helper — ``async for`` chunks, then ``await get_final_completion()``."""
"""Async streaming helper — ``async for`` OpenAI-style events, then ``await get_final_completion()``."""

def __init__(self, client: AsyncOpenAI, kwargs: Dict[str, Any], strip_fence: bool = False) -> None:
self._client = client
self._kwargs = kwargs
self._strip_fence = strip_fence
self._state = _State()
self._openai_state: ChatCompletionStreamState[None] = ChatCompletionStreamState()
self._filter = _SideChannelFilter()
self._role_injected = False
self._started = False
self._done = False

Expand All @@ -250,38 +279,34 @@ async def __aenter__(self) -> "AsyncInterfazeStream":
async def __aexit__(self, *exc: Any) -> None:
return None

async def __aiter__(self) -> "AsyncIterator[ChatCompletionChunk]":
async def __aiter__(self) -> "AsyncIterator[ChatCompletionStreamEvent[None]]":
if self._started:
raise InterfazeError("This stream has already been consumed.")
self._started = True
stream = await self._client.chat.completions.create(stream=True, **self._kwargs)
async for chunk in stream:
self._state.accumulate(chunk)
yield chunk
cleaned, self._role_injected = _clean_for_events(chunk, self._filter, self._role_injected)
for event in self._openai_state.handle_chunk(cleaned):
yield event
self._done = True

async def text_deltas(self) -> "AsyncIterator[str]":
"""Yield visible text only, stripping ``<think>``/``<precontext>`` across chunk boundaries.

Use this (not raw ``create(stream=True)`` deltas) for live rendering. ``reasoning`` and
``precontext`` remain available on ``get_final_completion()``.
"""
"""Iterate only the visible text (no side channels, no events) — for plain-token consumers."""
if self._started:
raise InterfazeError("This stream has already been consumed.")
self._started = True
filt = _SideChannelFilter()
stream = await self._client.chat.completions.create(stream=True, **self._kwargs)
async for chunk in stream:
self._state.accumulate(chunk)
if chunk.choices:
delta = chunk.choices[0].delta
if delta and isinstance(delta.content, str) and delta.content:
visible = filt.feed(delta.content)
if visible:
yield visible
tail = filt.flush()
if tail:
yield tail
if not chunk.choices:
continue
delta = chunk.choices[0].delta
piece = self._filter.feed(delta.content) if isinstance(delta.content, str) else ""
if chunk.choices[0].finish_reason:
piece += self._filter.flush()
if piece:
yield piece
self._done = True

async def get_final_completion(self) -> InterfazeChatCompletion:
Expand Down
Loading
Loading