Update app.py

app.py

@@ -173,7 +173,12 @@ class _RC:
         # Handle max_tokens
         _max_tokens = _kwargs.get('max_tokens')
         if _max_tokens is not None and _max_tokens > 0:
-            _params['max_tokens'] = _max_tokens
+            # Replicate Anthropic models often require >= 1024; clamp to avoid 422s
+            try:
+                _mt = int(_max_tokens)
+            except Exception:
+                _mt = 4096
+            _params['max_tokens'] = max(1024, _mt)
         else:
             _params['max_tokens'] = 4096
 
@@ -249,7 +254,7 @@ class _RC:
         return _tool_prompt
         
     def _stream_chat(self, _model_name, _prompt, _system="", **_kwargs):
-        """Stream chat using Replicate's streaming API"""
+        """Stream chat using Replicate's streaming API, yielding only text chunks."""
         _replicate_model = self._get_replicate_model(_model_name)
         _params = self._sanitize_params(**_kwargs)
 
@@ -261,14 +266,73 @@ class _RC:
             "top_p": _params['top_p']
         }
 
+        # pass through stop sequences if provided
+        if 'stop' in _kwargs and _kwargs['stop'] is not None:
+            _input["stop"] = _kwargs['stop']
+
         try:
-            # Use Replicate's streaming method
             for _event in self._client.stream(_replicate_model, input=_input):
-                if _event:
-                    yield str(_event)
+                if not _event:
+                    continue
+
+                # Fast path: plain string/bytes token
+                if isinstance(_event, (str, bytes)):
+                    yield (_event.decode('utf-8', errors='ignore') if isinstance(_event, bytes) else _event)
+                    continue
+
+                # Normalize event interfaces (object, dict, or custom)
+                _etype, _edata = None, None
+                if isinstance(_event, dict):
+                    _etype = _event.get('type') or _event.get('event')
+                    _edata = _event.get('data') or _event.get('output') or _event.get('text')
+                else:
+                    _etype = getattr(_event, 'type', None) or getattr(_event, 'event', None)
+                    _edata = getattr(_event, 'data', None)
+
+                # Extract text payloads
+                if _etype == "output" or _edata is not None:
+                    if isinstance(_edata, (list, tuple)):
+                        for _piece in _edata:
+                            if isinstance(_piece, (str, bytes)):
+                                yield (_piece.decode('utf-8', errors='ignore') if isinstance(_piece, bytes) else _piece)
+                    elif isinstance(_edata, (str, bytes)):
+                        yield (_edata.decode('utf-8', errors='ignore') if isinstance(_edata, bytes) else _edata)
+                    elif isinstance(_edata, dict):
+                        # Common nested keys
+                        for _k in ("text", "output", "delta"):
+                            if _k in _edata and isinstance(_edata[_k], (str, bytes)):
+                                _v = _edata[_k]
+                                yield (_v.decode('utf-8', errors='ignore') if isinstance(_v, bytes) else _v)
+                                break
+                    elif _etype in {"completed", "done", "end"}:
+                        break
+                    else:
+                        # Fallback to string form (restore old working behavior)
+                        try:
+                            _s = str(_event)
+                            if _s:
+                                yield _s
+                        except Exception:
+                            pass
+                elif _etype in {"error", "logs", "warning"}:
+                    try:
+                        _lg.warning(f"Replicate stream {_etype}: {_edata}")
+                    except Exception:
+                        pass
+                elif _etype in {"completed", "done", "end"}:
+                    break
+                else:
+                    # Unknown/eventless object; fallback to string form
+                    try:
+                        _s = str(_event)
+                        if _s:
+                            yield _s
+                    except Exception:
+                        pass
         except Exception as _e:
             _lg.error(f"Streaming error for {_replicate_model}: {_e}")
-            yield f"Error: {_e}"
+            # Surface a minimal safe error token
+            yield ""
 
     def _stream_from_prediction(self, _prediction):
         """Stream from a prediction using the stream URL"""
@@ -303,7 +367,7 @@ class _RC:
             yield f"Error: {_e}"
             
     def _complete_chat(self, _model_name, _prompt, _system="", **_kwargs):
-        """Complete chat using Replicate's run method"""
+        """Complete chat using Replicate's run method and coalesce into a single string."""
         _replicate_model = self._get_replicate_model(_model_name)
         _params = self._sanitize_params(**_kwargs)
 
@@ -315,12 +379,42 @@ class _RC:
             "top_p": _params['top_p']
         }
 
+        if 'stop' in _kwargs and _kwargs['stop'] is not None:
+            _input["stop"] = _kwargs['stop']
+
         try:
             _result = self._client.run(_replicate_model, input=_input)
-            return "".join(_result) if isinstance(_result, list) else str(_result)
+
+            # If it's a list of strings or chunks, join
+            if isinstance(_result, list):
+                _joined = "".join([x.decode("utf-8", errors="ignore") if isinstance(x, bytes) else str(x) for x in _result])
+                return _joined
+
+            # Some models return generators/iterables; accumulate
+            try:
+                from collections.abc import Iterator, Iterable
+                if isinstance(_result, Iterator) or (
+                    isinstance(_result, Iterable) and not isinstance(_result, (str, bytes))
+                ):
+                    _buf = []
+                    for _piece in _result:
+                        if isinstance(_piece, (str, bytes)):
+                            _buf.append(_piece.decode("utf-8", errors="ignore") if isinstance(_piece, bytes) else _piece)
+                        else:
+                            _buf.append(str(_piece))
+                    _text = "".join(_buf)
+                    if _text:
+                        return _text
+            except Exception:
+                pass
+
+            # FileOutput or scalar: cast to string; if empty, safe fallback
+            _text = str(_result) if _result is not None else ""
+            return _text
         except Exception as _e:
             _lg.error(f"Completion error for {_replicate_model}: {_e}")
-            return f"Error: {_e}"
+            # Return empty to avoid leaking internals into user-visible content
+            return ""
 
 # Global variables
 _client = None
@@ -492,7 +586,8 @@ async def _generate_stream_response(_request: _CCR, _prompt: str, _system: str,
                 'temperature': _request.temperature,
                 'top_p': _request.top_p,
                 'presence_penalty': _request.presence_penalty,
-                'frequency_penalty': _request.frequency_penalty
+                'frequency_penalty': _request.frequency_penalty,
+                'stop': _request.stop
             }
 
             # Use Replicate's direct streaming method with model parameter
@@ -617,11 +712,10 @@ async def _create_chat_completion(_request: _CCR):
             _lg.info(f"[{_request_id}] Starting streaming response")
             return _SR(
                 _generate_stream_response(_request, _prompt, _system, _request_id),
-                media_type="text/plain",
+                media_type="text/event-stream",
                 headers={
                     "Cache-Control": "no-cache",
-                    "Connection": "keep-alive",
-                    "Content-Type": "text/event-stream"
+                    "Connection": "keep-alive"
                 }
             )
         else: