754e273049
Added an endpoint for getting the actual stored responses, and used it to test and improve the python package.
41 lines
1.6 KiB
Python
41 lines
1.6 KiB
Python
from typing import Any, Optional
|
|
|
|
|
|
def merge_openai_chunks(base: Optional[Any], chunk: Any) -> Any:
|
|
if base is None:
|
|
return merge_openai_chunks({**chunk, "choices": []}, chunk)
|
|
|
|
choices = base["choices"].copy()
|
|
for choice in chunk["choices"]:
|
|
base_choice = next((c for c in choices if c["index"] == choice["index"]), None)
|
|
|
|
if base_choice:
|
|
base_choice["finish_reason"] = (
|
|
choice.get("finish_reason") or base_choice["finish_reason"]
|
|
)
|
|
base_choice["message"] = base_choice.get("message") or {"role": "assistant"}
|
|
|
|
if choice.get("delta") and choice["delta"].get("content"):
|
|
base_choice["message"]["content"] = (
|
|
base_choice["message"].get("content") or ""
|
|
) + (choice["delta"].get("content") or "")
|
|
if choice.get("delta") and choice["delta"].get("function_call"):
|
|
fn_call = base_choice["message"].get("function_call") or {}
|
|
fn_call["name"] = (fn_call.get("name") or "") + (
|
|
choice["delta"]["function_call"].get("name") or ""
|
|
)
|
|
fn_call["arguments"] = (fn_call.get("arguments") or "") + (
|
|
choice["delta"]["function_call"].get("arguments") or ""
|
|
)
|
|
else:
|
|
# Here, we'll have to handle the omitted property "delta" manually
|
|
new_choice = {k: v for k, v in choice.items() if k != "delta"}
|
|
choices.append(
|
|
{**new_choice, "message": {"role": "assistant", **choice["delta"]}}
|
|
)
|
|
|
|
return {
|
|
**base,
|
|
"choices": choices,
|
|
}
|