mirror of
https://github.com/open-webui/pipelines
synced 2025-05-12 16:40:45 +00:00
139 lines
4.4 KiB
Python
139 lines
4.4 KiB
Python
import uuid
|
|
import time
|
|
|
|
from typing import List
|
|
from schemas import OpenAIChatMessage
|
|
|
|
import inspect
|
|
from typing import get_type_hints, Literal
|
|
|
|
|
|
def stream_message_template(model: str, message: str):
|
|
return {
|
|
"id": f"{model}-{str(uuid.uuid4())}",
|
|
"object": "chat.completion.chunk",
|
|
"created": int(time.time()),
|
|
"model": model,
|
|
"choices": [
|
|
{
|
|
"index": 0,
|
|
"delta": {"content": message},
|
|
"logprobs": None,
|
|
"finish_reason": None,
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
def get_last_user_message(messages: List[dict]) -> str:
|
|
for message in reversed(messages):
|
|
if message["role"] == "user":
|
|
if isinstance(message["content"], list):
|
|
for item in message["content"]:
|
|
if item["type"] == "text":
|
|
return item["text"]
|
|
return message["content"]
|
|
return None
|
|
|
|
|
|
def get_last_assistant_message(messages: List[dict]) -> str:
|
|
for message in reversed(messages):
|
|
if message["role"] == "assistant":
|
|
if isinstance(message["content"], list):
|
|
for item in message["content"]:
|
|
if item["type"] == "text":
|
|
return item["text"]
|
|
return message["content"]
|
|
return None
|
|
|
|
|
|
def add_or_update_system_message(content: str, messages: List[dict]):
|
|
"""
|
|
Adds a new system message at the beginning of the messages list
|
|
or updates the existing system message at the beginning.
|
|
|
|
:param msg: The message to be added or appended.
|
|
:param messages: The list of message dictionaries.
|
|
:return: The updated list of message dictionaries.
|
|
"""
|
|
|
|
if messages and messages[0].get("role") == "system":
|
|
messages[0]["content"] += f"{content}\n{messages[0]['content']}"
|
|
else:
|
|
# Insert at the beginning
|
|
messages.insert(0, {"role": "system", "content": content})
|
|
|
|
return messages
|
|
|
|
|
|
def doc_to_dict(docstring):
|
|
lines = docstring.split("\n")
|
|
description = lines[1].strip()
|
|
param_dict = {}
|
|
|
|
for line in lines:
|
|
if ":param" in line:
|
|
line = line.replace(":param", "").strip()
|
|
param, desc = line.split(":", 1)
|
|
param_dict[param.strip()] = desc.strip()
|
|
ret_dict = {"description": description, "params": param_dict}
|
|
return ret_dict
|
|
|
|
|
|
def get_tools_specs(tools) -> List[dict]:
|
|
function_list = [
|
|
{"name": func, "function": getattr(tools, func)}
|
|
for func in dir(tools)
|
|
if callable(getattr(tools, func)) and not func.startswith("__")
|
|
]
|
|
|
|
specs = []
|
|
|
|
for function_item in function_list:
|
|
function_name = function_item["name"]
|
|
function = function_item["function"]
|
|
|
|
function_doc = doc_to_dict(function.__doc__ or function_name)
|
|
specs.append(
|
|
{
|
|
"name": function_name,
|
|
# TODO: multi-line desc?
|
|
"description": function_doc.get("description", function_name),
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
param_name: {
|
|
"type": param_annotation.__name__.lower(),
|
|
**(
|
|
{
|
|
"enum": (
|
|
param_annotation.__args__
|
|
if hasattr(param_annotation, "__args__")
|
|
else None
|
|
)
|
|
}
|
|
if hasattr(param_annotation, "__args__")
|
|
else {}
|
|
),
|
|
"description": function_doc.get("params", {}).get(
|
|
param_name, param_name
|
|
),
|
|
}
|
|
for param_name, param_annotation in get_type_hints(
|
|
function
|
|
).items()
|
|
if param_name != "return"
|
|
},
|
|
"required": [
|
|
name
|
|
for name, param in inspect.signature(
|
|
function
|
|
).parameters.items()
|
|
if param.default is param.empty
|
|
],
|
|
},
|
|
}
|
|
)
|
|
|
|
return specs
|