import json
import math
from writerai import Writer
# Initialize the Writer client. If you don't pass the `apiKey` parameter,
# the client looks for the `WRITER_API_KEY` environment variable.
client = Writer()
def calculate_mean(numbers: list) -> float:
if not numbers:
raise ValueError("Cannot calculate mean of an empty list")
return sum(numbers) / len(numbers)
def calculate_standard_deviation(numbers: list) -> float:
if len(numbers) < 2:
raise ValueError("Cannot calculate standard deviation with fewer than 2 numbers")
mean = calculate_mean(numbers)
variance = sum((x - mean) ** 2 for x in numbers) / len(numbers)
return math.sqrt(variance)
tools = [
{
"type": "function",
"function": {
"name": "calculate_mean",
"description": "Calculate the mean (average) of a list of numbers. Use this when asked for the average or mean of numbers.",
"parameters": {
"type": "object",
"properties": {
"numbers": {
"type": "array",
"items": {"type": "number"},
"description": "List of numbers to calculate the mean for"
}
},
"required": ["numbers"]
}
}
},
{
"type": "function",
"function": {
"name": "calculate_standard_deviation",
"description": "Calculate the standard deviation of a list of numbers. Use this when asked for standard deviation, variance, or statistical spread of numbers.",
"parameters": {
"type": "object",
"properties": {
"numbers": {
"type": "array",
"items": {"type": "number"},
"description": "List of numbers to calculate the standard deviation for"
}
},
"required": ["numbers"]
}
}
}
]
messages = [{"role": "user", "content": "What are the mean and standard deviation of [10, 20, 30, 40, 50]?"}]
# Step 1: Initial request with tools
response = client.chat.chat(
model="palmyra-x5",
messages=messages,
tools=tools,
tool_choice="auto",
stream=True
)
# Step 2: Process streaming response to collect multiple tool calls
streaming_content = ""
function_calls = []
for chunk in response:
choice = chunk.choices[0]
if choice.delta:
# Collect tool calls as they stream in
if choice.delta.tool_calls:
for tool_call in choice.delta.tool_calls:
if tool_call.id:
# Start a new function call
function_calls.append({
"name": "",
"arguments": "",
"call_id": tool_call.id
})
if tool_call.function:
# Append to the most recent function call
if function_calls:
function_calls[-1]["name"] += tool_call.function.name or ""
function_calls[-1]["arguments"] += tool_call.function.arguments or ""
# Collect regular content (for cases where no tools are called)
elif choice.delta.content:
streaming_content += choice.delta.content
# Check if streaming is complete
if choice.finish_reason:
if choice.finish_reason == "stop":
# No tools were called, just regular response
print(f"Response: {streaming_content}")
messages.append({"role": "assistant", "content": streaming_content})
break
elif choice.finish_reason == "tool_calls":
print(f"Tool calls collected: {len(function_calls)}")
break
# Step 3: Reconstruct and append assistant message with multiple tool calls
tool_calls_for_message = []
for func_call in function_calls:
tool_calls_for_message.append({
"id": func_call["call_id"],
"type": "function",
"function": {
"name": func_call["name"],
"arguments": func_call["arguments"]
}
})
assistant_message = {
"role": "assistant",
"content": None,
"tool_calls": tool_calls_for_message
}
messages.append(assistant_message)
# Step 4: Execute all functions and append results
for function_call in function_calls:
function_name = function_call["name"]
print(f"Processing tool call: {function_name}")
if function_name == "calculate_mean":
try:
arguments_dict = json.loads(function_call["arguments"])
function_response = calculate_mean(arguments_dict["numbers"])
except Exception as e:
function_response = f"Error: {str(e)}"
elif function_name == "calculate_standard_deviation":
try:
arguments_dict = json.loads(function_call["arguments"])
function_response = calculate_standard_deviation(arguments_dict["numbers"])
except Exception as e:
function_response = f"Error: {str(e)}"
else:
function_response = f"Unknown function: {function_name}"
print(f"Function response: {function_response}")
# Append tool result to conversation history
messages.append({
"role": "tool",
"content": str(function_response),
"tool_call_id": function_call["call_id"],
"name": function_name,
})
# Step 5: Get the final response
final_response = client.chat.chat(
model="palmyra-x5",
messages=messages,
stream=True
)
final_content = ""
for chunk in final_response:
choice = chunk.choices[0]
if choice.delta and choice.delta.content:
final_content += choice.delta.content
print(choice.delta.content, end="", flush=True)
print(f"\nFinal response: {final_content}")
# Step 6: Append final response to conversation history
messages.append({
"role": "assistant",
"content": final_content
})