|
|
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": |
|
|
|
|
|
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": |
|
|
|
|
|
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 = "" |
|
|
reasoning = "" |
|
|
function_calls_to_handle = [] |
|
|
|
|
|
for chunk in stream: |
|
|
delta = chunk.choices[0].delta |
|
|
|
|
|
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: |
|
|
reasoning += delta.reasoning_content |
|
|
elif delta.content: |
|
|
output += delta.content |
|
|
|
|
|
yield f"*{reasoning}*\n{output}" |
|
|
|
|
|
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) |
