INNER CODE UNIT · Python

create_completion_stream

hyperonym/basaran · basaran/__main__.py:176

def create_completion_stream(options, template):
    """Return text completion results in event stream."""

    # Serialize data for event stream.
    def serialize(data):
        data = json.dumps(data, ensure_ascii=False, separators=(",", ":"))
        return f"data: {data}\n\n"

    def stream():
        buffers = {}
        times = {}
        for choice in stream_model(**options):
            index = choice["index"]
            now = time.time_ns()
            if index not in buffers:
                buffers[index] = []
            if index not in times:
                times[index] = now

View source record →