Structured Output & Tool Integration

# Use OpenAI structured outputs with a JSON schema constraint
# The model is forced to return valid JSON matching the schema
from openai import OpenAI

client = OpenAI()

# Send chat completion request to the API
response = client.chat.completions.create(
 model="gpt-4o",
 response_format={"type": "json_object"},
 messages=[
 {"role": "system", "content": "Extract contact info. Return JSON with keys: name, email, sentiment."},
 {"role": "user", "content": "Hi, I'm Sarah Chen (sarah@example.com) and I love your product!"}
 ]
)

import json
# Extract the generated message from the API response
data = json.loads(response.choices[0].message.content)
# Pretty-print for readability
print(json.dumps(data, indent=2))

# Use Instructor to extract structured Pydantic objects from LLM responses
# Instructor patches the OpenAI client to return validated models
import instructor
from pydantic import BaseModel, Field
from openai import OpenAI

client = instructor.from_openai(OpenAI())

class StrictRating(BaseModel):
 score: int = Field(ge=1, le=5, description="Rating from 1 to 5 only")
 explanation: str

# Instructor retries automatically if the model returns score=6 or score=0
rating = client.chat.completions.create(
 model="gpt-4o",
 response_model=StrictRating,
 max_retries=3, # Retry up to 3 times on validation failure
 messages=[
 {"role": "user", "content": "Rate this product on a scale of 1-10: Amazing!"}
 ]
)
# Even though the prompt says "1-10", Pydantic enforces 1-5 via retries
print(f"Score: {rating.score} (constrained to 1-5)")
print(f"Explanation: {rating.explanation}")

# Set up structured output extraction from the LLM
# Parse the response into a typed data structure
from openai import OpenAI
import json

client = OpenAI()

# Define the available tools
tools = [
 {
 "type": "function",
 "function": {
 "name": "get_weather",
 "description": "Get current weather for a city",
 "parameters": {
 "type": "object",
 "properties": {
 "city": {"type": "string", "description": "City name"},
 "units": {
 "type": "string",
 "enum": ["celsius", "fahrenheit"],
 "description": "Temperature unit"
 }
 },
 "required": ["city"]
 }
 }
 }
]

# Simulated weather function
def get_weather(city: str, units: str = "celsius") -> str:
 data = {"Paris": "22C, sunny", "London": "15C, cloudy", "Tokyo": "28C, humid"}
 return data.get(city, f"No data for {city}")

# Step 1: Send user message with tools
response = client.chat.completions.create(
 model="gpt-4o",
 messages=[{"role": "user", "content": "What's the weather in Paris?"}],
 tools=tools
)

# Step 2: Check if model wants to call a tool
message = response.choices[0].message
if message.tool_calls:
 tool_call = message.tool_calls[0]
 args = json.loads(tool_call.function.arguments)
 print(f"Model wants to call: {tool_call.function.name}({args})")

 # Step 3: Execute the function
 result = get_weather(**args)

 # Step 4: Send result back to model
 final_response = client.chat.completions.create(
 model="gpt-4o",
 messages=[
 {"role": "user", "content": "What's the weather in Paris?"},
 message, # The assistant's tool_call message
 {"role": "tool", "tool_call_id": tool_call.id, "content": result}
 ],
 tools=tools
 )
 print(f"Final answer: {final_response.choices[0].message.content}")

# Define tools for function calling and handle tool_calls in the response
# The model decides which function to invoke based on the user query
import anthropic
import json

client = anthropic.Anthropic()

tools = [
 {
 "name": "get_weather",
 "description": "Get current weather for a city",
 "input_schema": {
 "type": "object",
 "properties": {
 "city": {"type": "string", "description": "City name"},
 "units": {"type": "string", "enum": ["celsius", "fahrenheit"]}
 },
 "required": ["city"]
 }
 }
]

def get_weather(city, units="celsius"):
 data = {"Paris": "22C, sunny", "London": "15C, cloudy"}
 return data.get(city, f"No data for {city}")

# Step 1: Send request with tools
response = client.messages.create(
 model="claude-sonnet-4-20250514",
 max_tokens=300,
 tools=tools,
 messages=[{"role": "user", "content": "What's the weather in Paris?"}]
)

# Step 2: Find the tool_use block
for block in response.content:
 if block.type == "tool_use":
 print(f"Tool call: {block.name}({block.input})")
 result = get_weather(**block.input)

 # Step 3: Send tool result back
 final = client.messages.create(
 model="claude-sonnet-4-20250514",
 max_tokens=300,
 tools=tools,
 messages=[
 {"role": "user", "content": "What's the weather in Paris?"},
 {"role": "assistant", "content": response.content},
 {"role": "user", "content": [
 {"type": "tool_result", "tool_use_id": block.id, "content": result}
 ]}
 ]
 )
 print(f"Final: {final.content[0].text}")

# Use Google Gemini function calling for structured output
# Gemini uses FunctionDeclaration for tool definitions
from google import genai
from google.genai import types

client = genai.Client()

# Define tools using function_declarations
weather_tool = types.Tool(
 function_declarations=[
 types.FunctionDeclaration(
 name="get_weather",
 description="Get current weather for a city",
 parameters=types.Schema(
 type="OBJECT",
 properties={
 "city": types.Schema(type="STRING", description="City name"),
 },
 required=["city"],
 ),
 )
 ]
)

def get_weather(city: str) -> str:
 data = {"Paris": "22C, sunny", "Tokyo": "28C, humid"}
 return data.get(city, f"No data for {city}")

# Step 1: Send request with tools
response = client.models.generate_content(
 model="gemini-2.5-flash",
 contents="What's the weather in Paris?",
 config=types.GenerateContentConfig(tools=[weather_tool]),
)

# Step 2: Extract the function call from the response
part = response.candidates[0].content.parts[0]
print(f"Function call: {part.function_call.name}({dict(part.function_call.args)})")
result = get_weather(**dict(part.function_call.args))

# Step 3: Send function response back
from google.genai.types import Content, Part

final = client.models.generate_content(
 model="gemini-2.5-flash",
 contents=[
 Content(parts=[Part(text="What's the weather in Paris?")], role="user"),
 response.candidates[0].content,
 Content(parts=[Part(function_response=types.FunctionResponse(
 name="get_weather", response={"result": result}
 ))], role="user"),
 ],
 config=types.GenerateContentConfig(tools=[weather_tool]),
)
print(f"Final: {final.text}")

# Run multiple LLM calls concurrently using asyncio
# Async batching reduces wall-clock time for parallel requests
import asyncio
from openai import OpenAI
import json

client = OpenAI()

tools = [
 {
 "type": "function",
 "function": {
 "name": "get_weather",
 "description": "Get current weather for a city",
 "parameters": {
 "type": "object",
 "properties": {
 "city": {"type": "string"}
 },
 "required": ["city"]
 }
 }
 }
]

def get_weather(city):
 data = {"Paris": "22C, sunny", "Tokyo": "28C, humid", "London": "15C, cloudy"}
 return data.get(city, "Unknown")

# Send chat completion request to the API
response = client.chat.completions.create(
 model="gpt-4o",
 messages=[{"role": "user", "content": "Weather in Paris and Tokyo?"}],
 tools=tools
)

# Extract the generated message from the API response
message = response.choices[0].message
print(f"Number of tool calls: {len(message.tool_calls)}")

# Execute all tool calls and collect results
tool_messages = []
for tc in message.tool_calls:
 args = json.loads(tc.function.arguments)
 result = get_weather(**args)
 print(f" {tc.function.name}({args}) = {result}")
 tool_messages.append({
 "role": "tool",
 "tool_call_id": tc.id,
 "content": result
 })

# Send all results back at once
final = client.chat.completions.create(
 model="gpt-4o",
 messages=[
 {"role": "user", "content": "Weather in Paris and Tokyo?"},
 message,
 *tool_messages
 ],
 tools=tools
)
print(f"\nFinal: {final.choices[0].message.content}")