29 KiB
29 KiB
In [ ]:
from openai import OpenAI
import json
from sglang.utils import wait_for_server, print_highlight, terminate_process
from sglang.test.test_utils import is_in_ci
if is_in_ci():
from patch import launch_server_cmd
else:
from sglang.utils import launch_server_cmd
import nest_asyncio
nest_asyncio.apply()
server_process, port = launch_server_cmd(
"python3 -m sglang.launch_server --model-path Qwen/Qwen2.5-7B-Instruct --tool-call-parser qwen25 --host 0.0.0.0" # qwen25
)
wait_for_server(f"http://localhost:{port}")In [ ]:
# Define tools
tools = [
{
"type": "function",
"function": {
"name": "get_current_weather",
"description": "Get the current weather in a given location",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "The city to find the weather for, e.g. 'San Francisco'",
},
"state": {
"type": "string",
"description": "the two-letter abbreviation for the state that the city is"
" in, e.g. 'CA' which would mean 'California'",
},
"unit": {
"type": "string",
"description": "The unit to fetch the temperature in",
"enum": ["celsius", "fahrenheit"],
},
},
"required": ["city", "state", "unit"],
},
},
}
]In [ ]:
def get_messages():
return [
{
"role": "user",
"content": "What's the weather like in Boston today? Output a reasoning before act, then use the tools to help you.",
}
]
messages = get_messages()In [ ]:
# Initialize OpenAI-like client
client = OpenAI(api_key="None", base_url=f"http://0.0.0.0:{port}/v1")
model_name = client.models.list().data[0].idIn [ ]:
# Non-streaming mode test
response_non_stream = client.chat.completions.create(
model=model_name,
messages=messages,
temperature=0,
top_p=0.95,
max_tokens=1024,
stream=False, # Non-streaming
tools=tools,
)
print_highlight("Non-stream response:")
print(response_non_stream)
print_highlight("==== content ====")
print(response_non_stream.choices[0].message.content)
print_highlight("==== tool_calls ====")
print(response_non_stream.choices[0].message.tool_calls)In [ ]:
name_non_stream = response_non_stream.choices[0].message.tool_calls[0].function.name
arguments_non_stream = (
response_non_stream.choices[0].message.tool_calls[0].function.arguments
)
print_highlight(f"Final streamed function call name: {name_non_stream}")
print_highlight(f"Final streamed function call arguments: {arguments_non_stream}")In [ ]:
# Streaming mode test
print_highlight("Streaming response:")
response_stream = client.chat.completions.create(
model=model_name,
messages=messages,
temperature=0,
top_p=0.95,
max_tokens=1024,
stream=True, # Enable streaming
tools=tools,
)
texts = ""
tool_calls = []
name = ""
arguments = ""
for chunk in response_stream:
if chunk.choices[0].delta.content:
texts += chunk.choices[0].delta.content
if chunk.choices[0].delta.tool_calls:
tool_calls.append(chunk.choices[0].delta.tool_calls[0])
print_highlight("==== Text ====")
print(texts)
print_highlight("==== Tool Call ====")
for tool_call in tool_calls:
print(tool_call)In [ ]:
# Parse and combine function call arguments
arguments = []
for tool_call in tool_calls:
if tool_call.function.name:
print_highlight(f"Streamed function call name: {tool_call.function.name}")
if tool_call.function.arguments:
arguments.append(tool_call.function.arguments)
# Combine all fragments into a single JSON string
full_arguments = "".join(arguments)
print_highlight(f"streamed function call arguments: {full_arguments}")In [ ]:
# This is a demonstration, define real function according to your usage.
def get_current_weather(city: str, state: str, unit: "str"):
return (
f"The weather in {city}, {state} is 85 degrees {unit}. It is "
"partly cloudly, with highs in the 90's."
)
available_tools = {"get_current_weather": get_current_weather}In [ ]:
messages.append(response_non_stream.choices[0].message)
# Call the corresponding tool function
tool_call = messages[-1].tool_calls[0]
tool_name = tool_call.function.name
tool_to_call = available_tools[tool_name]
result = tool_to_call(**(json.loads(tool_call.function.arguments)))
print_highlight(f"Function call result: {result}")
# messages.append({"role": "tool", "content": result, "name": tool_name})
messages.append(
{
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(result),
"name": tool_name,
}
)
print_highlight(f"Updated message history: {messages}")In [ ]:
final_response = client.chat.completions.create(
model=model_name,
messages=messages,
temperature=0,
top_p=0.95,
stream=False,
tools=tools,
)
print_highlight("Non-stream response:")
print(final_response)
print_highlight("==== Text ====")
print(final_response.choices[0].message.content)In [ ]:
from openai import OpenAI
import json
from sglang.utils import wait_for_server, print_highlight, terminate_process
from sglang.test.test_utils import is_in_ci
if is_in_ci():
from patch import launch_server_cmd
else:
from sglang.utils import launch_server_cmd
import nest_asyncio
nest_asyncio.apply()
# Start a new server session for tool choice examples
server_process_tool_choice, port_tool_choice = launch_server_cmd(
"python3 -m sglang.launch_server --model-path Qwen/Qwen2.5-7B-Instruct --tool-call-parser qwen25 --host 0.0.0.0"
)
wait_for_server(f"http://localhost:{port_tool_choice}")
# Initialize client for tool choice examples
client_tool_choice = OpenAI(
api_key="None", base_url=f"http://0.0.0.0:{port_tool_choice}/v1"
)
model_name_tool_choice = client_tool_choice.models.list().data[0].id
# Example with tool_choice="required" - forces the model to call a tool
messages_required = [
{"role": "user", "content": "Hello, what is the capital of France?"}
]
# Define tools
tools = [
{
"type": "function",
"function": {
"name": "get_current_weather",
"description": "Get the current weather in a given location",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "The city to find the weather for, e.g. 'San Francisco'",
},
"unit": {
"type": "string",
"description": "The unit to fetch the temperature in",
"enum": ["celsius", "fahrenheit"],
},
},
"required": ["city", "unit"],
},
},
}
]
response_required = client_tool_choice.chat.completions.create(
model=model_name_tool_choice,
messages=messages_required,
temperature=0,
max_tokens=1024,
tools=tools,
tool_choice="required", # Force the model to call a tool
)
print_highlight("Response with tool_choice='required':")
print("Content:", response_required.choices[0].message.content)
print("Tool calls:", response_required.choices[0].message.tool_calls)In [ ]:
# Example with specific function choice - forces the model to call a specific function
messages_specific = [
{"role": "user", "content": "What are the most attactive places in France?"}
]
response_specific = client_tool_choice.chat.completions.create(
model=model_name_tool_choice,
messages=messages_specific,
temperature=0,
max_tokens=1024,
tools=tools,
tool_choice={
"type": "function",
"function": {"name": "get_current_weather"},
}, # Force the model to call the specific get_current_weather function
)
print_highlight("Response with specific function choice:")
print("Content:", response_specific.choices[0].message.content)
print("Tool calls:", response_specific.choices[0].message.tool_calls)
if response_specific.choices[0].message.tool_calls:
tool_call = response_specific.choices[0].message.tool_calls[0]
print(f"Called function: {tool_call.function.name}")
print(f"Arguments: {tool_call.function.arguments}")In [ ]:
from transformers import AutoTokenizer
import requests
# generate an answer
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2.5-7B-Instruct")
messages = get_messages()
input = tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=True,
tools=tools,
)
gen_url = f"http://localhost:{port}/generate"
gen_data = {
"text": input,
"sampling_params": {
"skip_special_tokens": False,
"max_new_tokens": 1024,
"temperature": 0,
"top_p": 0.95,
},
}
gen_response = requests.post(gen_url, json=gen_data).json()["text"]
print_highlight("==== Response ====")
print(gen_response)
# parse the response
parse_url = f"http://localhost:{port}/parse_function_call"
function_call_input = {
"text": gen_response,
"tool_call_parser": "qwen25",
"tools": tools,
}
function_call_response = requests.post(parse_url, json=function_call_input)
function_call_response_json = function_call_response.json()
print_highlight("==== Text ====")
print(function_call_response_json["normal_text"])
print_highlight("==== Calls ====")
print("function name: ", function_call_response_json["calls"][0]["name"])
print("function arguments: ", function_call_response_json["calls"][0]["parameters"])In [ ]:
terminate_process(server_process)In [ ]:
import sglang as sgl
from sglang.srt.function_call.function_call_parser import FunctionCallParser
from sglang.srt.managers.io_struct import Tool, Function
llm = sgl.Engine(model_path="Qwen/Qwen2.5-7B-Instruct")
tokenizer = llm.tokenizer_manager.tokenizer
input_ids = tokenizer.apply_chat_template(
messages, tokenize=True, add_generation_prompt=True, tools=tools
)
sampling_params = {
"max_new_tokens": 1024,
"temperature": 0,
"top_p": 0.95,
"skip_special_tokens": False,
}
# 1) Offline generation
result = llm.generate(input_ids=input_ids, sampling_params=sampling_params)
generated_text = result["text"] # Assume there is only one prompt
print("=== Offline Engine Output Text ===")
print(generated_text)
# 2) Parse using FunctionCallParser
def convert_dict_to_tool(tool_dict: dict) -> Tool:
function_dict = tool_dict.get("function", {})
return Tool(
type=tool_dict.get("type", "function"),
function=Function(
name=function_dict.get("name"),
description=function_dict.get("description"),
parameters=function_dict.get("parameters"),
),
)
tools = [convert_dict_to_tool(raw_tool) for raw_tool in tools]
parser = FunctionCallParser(tools=tools, tool_call_parser="qwen25")
normal_text, calls = parser.parse_non_stream(generated_text)
print("=== Parsing Result ===")
print("Normal text portion:", normal_text)
print("Function call portion:")
for call in calls:
# call: ToolCallItem
print(f" - tool name: {call.name}")
print(f" parameters: {call.parameters}")
# 3) If needed, perform additional logic on the parsed functions, such as automatically calling the corresponding function to obtain a return value, etc.In [ ]:
llm.shutdown()In [ ]:
import openai
server_process, port = launch_server_cmd(
" python3 -m sglang.launch_server --model-path meta-llama/Llama-3.2-1B-Instruct --tool-call-parser pythonic --tp 1" # llama-3.2-1b-instruct
)
wait_for_server(f"http://localhost:{port}")
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get the current weather for a given location.",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The name of the city or location.",
}
},
"required": ["location"],
},
},
},
{
"type": "function",
"function": {
"name": "get_tourist_attractions",
"description": "Get a list of top tourist attractions for a given city.",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "The name of the city to find attractions for.",
}
},
"required": ["city"],
},
},
},
]
def get_messages():
return [
{
"role": "system",
"content": (
"You are a travel assistant. "
"When asked to call functions, ALWAYS respond ONLY with a python list of function calls, "
"using this format: [func_name1(param1=value1, param2=value2), func_name2(param=value)]. "
"Do NOT use JSON, do NOT use variables, do NOT use any other format. "
"Here is an example:\n"
'[get_weather(location="Paris"), get_tourist_attractions(city="Paris")]'
),
},
{
"role": "user",
"content": (
"I'm planning a trip to Tokyo next week. What's the weather like and what are some top tourist attractions? "
"Propose parallel tool calls at once, using the python list of function calls format as shown above."
),
},
]
messages = get_messages()
client = openai.Client(base_url=f"http://localhost:{port}/v1", api_key="xxxxxx")
model_name = client.models.list().data[0].id
response_non_stream = client.chat.completions.create(
model=model_name,
messages=messages,
temperature=0,
top_p=0.9,
stream=False, # Non-streaming
tools=tools,
)
print_highlight("Non-stream response:")
print(response_non_stream)
response_stream = client.chat.completions.create(
model=model_name,
messages=messages,
temperature=0,
top_p=0.9,
stream=True,
tools=tools,
)
texts = ""
tool_calls = []
name = ""
arguments = ""
for chunk in response_stream:
if chunk.choices[0].delta.content:
texts += chunk.choices[0].delta.content
if chunk.choices[0].delta.tool_calls:
tool_calls.append(chunk.choices[0].delta.tool_calls[0])
print_highlight("Streaming Response:")
print_highlight("==== Text ====")
print(texts)
print_highlight("==== Tool Call ====")
for tool_call in tool_calls:
print(tool_call)
terminate_process(server_process)