File size: 5,478 Bytes
9726fac 1e2d981 9726fac 2332ba1 9726fac e5f6777 1e2d981 2332ba1 1729f2d 2332ba1 1729f2d 2332ba1 1e2d981 2332ba1 6a34e6a 2332ba1 1e2d981 9702f0e e5f6777 9726fac 1e2d981 9726fac 9702f0e 9726fac e5f6777 2332ba1 e5f6777 9726fac 1d9d6ab 9a28b27 9726fac 2332ba1 9726fac 1c58dec e5f6777 2332ba1 1729f2d 1c58dec 9726fac 1e2d981 2332ba1 9726fac 8a8d916 e5f6777 1e2d981 d24f851 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 |
import subprocess
import gradio as gr
from openai import OpenAI
import json
subprocess.Popen("bash /home/user/app/start.sh", shell=True)
client = OpenAI(base_url="http://0.0.0.0:8000/v1", api_key="sk-local", timeout=600)
def handle_function_call(function_name, arguments):
"""Handle function calls from the model"""
if function_name == "browser_search":
# Implement your browser search logic here
query = arguments.get("query", "")
max_results = arguments.get("max_results", 5)
return f"Search results for '{query}' (max {max_results} results): [Implementation needed]"
elif function_name == "code_interpreter":
# Implement your code interpreter logic here
code = arguments.get("code", "")
if not code:
return "No code provided to execute."
return f"Code interpreter results for '{code}': [Implementation needed]"
return f"Unknown function: {function_name}"
def respond(
message,
history: list[tuple[str, str]] = [],
system_message=None,
max_tokens=None,
temperature=0.7,
):
messages = []
if system_message:
messages = [{"role": "system", "content": system_message}]
for user, assistant in history:
if user:
messages.append({"role": "user", "content": user})
if assistant:
messages.append({"role": "assistant", "content": assistant})
messages.append({"role": "user", "content": message})
try:
stream = client.chat.completions.create(
model="Deepseek-R1-0528-Qwen3-8B",
messages=messages,
max_tokens=max_tokens,
temperature=temperature,
stream=True,
tools=[
{
"type": "function",
"function": {
"name": "browser_search",
"description": (
"Search the web for a given query and return the most relevant results."
),
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query string.",
},
"max_results": {
"type": "integer",
"description": (
"Maximum number of search results to return. "
"If omitted the service will use its default."
),
"default": 5,
},
},
"required": ["query"],
},
},
},
{
"type": "function",
"function": {
"name": "code_interpreter",
"description": (
"Execute Python code and return the results. "
"Can generate plots, perform calculations, and data analysis."
),
"parameters": {
"type": "object",
"properties": {
"code": {
"type": "string",
"description": "The Python code to execute.",
},
},
"required": ["code"],
},
},
},
],
)
print("messages", messages)
output = ""
function_calls_to_handle = []
for chunk in stream:
delta = chunk.choices[0].delta
# Handle function calls
if hasattr(delta, "tool_calls") and delta.tool_calls:
for tool_call in delta.tool_calls:
if tool_call.function:
function_calls_to_handle.append(
{
"name": tool_call.function.name,
"arguments": json.loads(tool_call.function.arguments),
}
)
if hasattr(delta, "reasoning_content") and delta.reasoning_content:
# output += delta.reasoning_content
output = f"""*{output}{delta.reasoning_content}*\n"""
elif delta.content:
output += delta.content
yield output
# Handle any function calls that were made
if function_calls_to_handle:
for func_call in function_calls_to_handle:
func_result = handle_function_call(
func_call["name"], func_call["arguments"]
)
output += (
f"\n\n**Function Result ({func_call['name']}):**\n{func_result}"
)
yield output
except Exception as e:
print(f"[Error] {e}")
yield "⚠️ Llama.cpp server error"
demo = gr.ChatInterface(respond)
if __name__ == "__main__":
demo.launch(show_api=False)
|