FilesExpand file tree

main

/

function_calling_errors.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

157 lines (137 loc) · 5.57 KB

main

/

function_calling_errors.py

Copy path

Top

File metadata and controls

• Code
• Blame

157 lines (137 loc) · 5.57 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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

153

154

155

156

157

import json

import os

from collections.abc import Callable

from typing import Any

import azure.identity

import openai

from dotenv import load_dotenv

# Setup the OpenAI client to use either Azure, OpenAI.com, or Ollama API

load_dotenv(override=True)

API_HOST = os.getenv("API_HOST", "azure")

if API_HOST == "azure":

token_provider = azure.identity.get_bearer_token_provider(

azure.identity.DefaultAzureCredential(), "https://cognitiveservices.azure.com/.default"

)

client = openai.OpenAI(

base_url=f"{os.environ['AZURE_OPENAI_ENDPOINT'].rstrip('/')}/openai/v1/",

api_key=token_provider,

)

MODEL_NAME = os.environ["AZURE_OPENAI_CHAT_DEPLOYMENT"]

elif API_HOST == "ollama":

client = openai.OpenAI(base_url=os.environ["OLLAMA_ENDPOINT"], api_key="nokeyneeded")

MODEL_NAME = os.environ["OLLAMA_MODEL"]

else:

client = openai.OpenAI(api_key=os.environ["OPENAI_KEY"])

MODEL_NAME = os.environ["OPENAI_MODEL"]

# ---------------------------------------------------------------------------

# Tool implementation(s)

# ---------------------------------------------------------------------------

def search_database(search_query: str, price_filter: dict | None = None) -> dict[str, str]:

"""Search database for relevant products based on user query"""

if not search_query:

raise ValueError("search_query is required")

if price_filter:

if "comparison_operator" not in price_filter or "value" not in price_filter:

raise ValueError("Both comparison_operator and value are required in price_filter")

if price_filter["comparison_operator"] not in {">", "<", ">=", "<=", "="}:

raise ValueError("Invalid comparison_operator in price_filter")

if not isinstance(price_filter["value"], int | float):

raise ValueError("Value in price_filter must be a number")

return [{"id": "123", "name": "Example Product", "price": 19.99}]

tool_mapping: dict[str, Callable[..., Any]] = {

"search_database": search_database,

}

tools = [

{

"type": "function",

"name": "search_database",

"description": "Search database for relevant products based on user query",

"parameters": {

"type": "object",

"properties": {

"search_query": {

"type": "string",

"description": "Query string to use for full text search, e.g. 'red shoes'",

},

"price_filter": {

"type": "object",

"description": "Filter search results based on price of the product",

"properties": {

"comparison_operator": {

"type": "string",

"description": "Operator to compare the column value, either '>', '<', '>=', '<=', '='", # noqa

},

"value": {

"type": "number",

"description": "Value to compare against, e.g. 30",

},

},

},

},

"required": ["search_query"],

},

}

]

messages: list[dict[str, Any]] = [

{"role": "system", "content": "You are a product search assistant."},

{"role": "user", "content": "Find me a red shirt under $20."},

]

print(f"Model: {MODEL_NAME} on Host: {API_HOST}\n")

# First model response (may include tool call)

response = client.responses.create(

model=MODEL_NAME,

input=messages,

tools=tools,

tool_choice="auto",

store=False,

)

tool_calls = [item for item in response.output if item.type == "function_call"]

# If no tool calls were requested, just print the answer.

if not tool_calls:

print("Assistant:")

print(response.output_text)

else:

# Add the function call items from the response output

messages.extend(response.output)

# Process each requested tool sequentially (though usually one here)

for tool_call in tool_calls:

fn_name = tool_call.name

raw_args = tool_call.arguments or "{}"

print(f"Tool request: {fn_name}({raw_args})")

target = tool_mapping.get(fn_name)

if not target:

tool_result: Any = f"ERROR: No implementation registered for tool '{fn_name}'"

else:

# Parse arguments safely

try:

parsed_args = json.loads(raw_args) if raw_args.strip() else {}

except json.JSONDecodeError:

parsed_args = {}

tool_result = "Warning: Malformed JSON arguments received; proceeding with empty args"

else:

try:

tool_result = target(**parsed_args)

except Exception as e: # safeguard tool execution

tool_result = f"Tool execution error in {fn_name}: {e}"

# Serialize tool output (dict or str) as JSON string for the model

try:

tool_content = json.dumps(tool_result)

except Exception:

# Fallback to string conversion if something isn't JSON serializable

tool_content = json.dumps({"result": str(tool_result)})

messages.append(

{

"type": "function_call_output",

"call_id": tool_call.call_id,

"output": tool_content,

}

)

# Follow-up model response after supplying tool outputs

followup = client.responses.create(

model=MODEL_NAME,

input=messages,

tools=tools,

store=False,

)

print("Assistant (final):")

print(followup.output_text)