# function_calling/basics.py from openai import OpenAI from pydantic import BaseModel, Field from typing import List, Optional import json client = OpenAI() # Define function schema using Pydantic class GetWeather(BaseModel): """Get current weather for a location.""" location: str = Field(description="City and state, e.g. 'San Francisco, CA'") unit: str = Field(default="celsius", enum=["celsius", "fahrenheit"]) class SearchWeb(BaseModel): """Search the web for information.""" query: str = Field(description="Search query") max_results: int = Field(default=5, ge=1, le=10) recency_days: Optional[int] = Field(default=None, ge=1, le=365) # Convert to OpenAI format def pydantic_to_openai(model: type[BaseModel]) -> dict: schema = model.model_json_schema() return { "type": "function", "function": { "name": schema["title"], "description": schema["description"], "parameters": { "type": "object", "properties": schema["properties"], "required": schema.get("required", []) } } } tools = [pydantic_to_openai(GetWeather), pydantic_to_openai(SearchWeb)] # Execute function calls def execute_function(name: str, arguments: dict) -> dict: if name == "GetWeather": # Call actual weather API  return {"temperature": 22, "condition": "Sunny", "unit": arguments["unit"]} elif name == "SearchWeb": # Call search API  return {"results": [{"title": "Result 1", "url": "..."}]} return {"error": f"Unknown function: {name}"} # Chat completion with tools def chat_with_tools(messages: list, tools: list) -> dict: response = client.chat.completions.create( model="gpt-4o", messages=messages, tools=tools, tool_choice="auto", parallel_tool_calls=True ) return response.choices[0].message # Usage messages = [ {"role": "user", "content": "What's the weather in Tokyo and search for 'AI agents'"} ] response_message = chat_with_tools(messages, tools) if response_message.tool_calls: for tool_call in response_message.tool_calls: args = json.loads(tool_call.function.arguments) result = execute_function(tool_call.function.name, args) messages.append(response_message.model_dump()) messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(result) }) # Get final response  final = client.chat.completions.create( model="gpt-4o", messages=messages, tools=tools ) print(final.choices[0].message.content) 

# function_calling/structured_output.py from pydantic import BaseModel, Field, validator from typing import Literal, List, Optional from openai import OpenAI import instructor client = instructor.from_openai(OpenAI()) class AgentAction(BaseModel): """Single action in an agent plan.""" tool: str = Field(description="Tool name to call") args: dict = Field(description="Arguments for the tool") reasoning: str = Field(description="Why this action") class AgentPlan(BaseModel): """Complete plan for solving a task.""" goal: str = Field(description="High-level goal") steps: List[AgentAction] = Field(min_items=1) estimated_duration_seconds: int = Field(ge=1) requires_user_input: bool = False class AgentResult(BaseModel): """Result of agent execution.""" success: bool output: str artifacts: List[str] = Field(default_factory=list) error: Optional[str] = None tokens_used: int = 0 # Use instructor for guaranteed structured output plan = client.chat.completions.create( model="gpt-4o", response_model=AgentPlan, messages=[ {"role": "system", "content": "You are a planning agent. Create detailed plans."}, {"role": "user", "content": "Research the latest LangGraph features and write a summary"} ] ) print(plan.model_dump_json(indent=2)) 

# agent/tool_registry.py from abc import ABC, abstractmethod from dataclasses import dataclass from typing import Any, Callable, Dict, List, Optional from pydantic import BaseModel import inspect @dataclass class ToolSpec: name: str description: str parameters: dict # JSON Schema  function: Callable return_type: type class ToolRegistry: def __init__(self): self.tools: Dict[str, ToolSpec] = {} def register(self, spec: ToolSpec): self.tools[spec.name] = spec def register_function( self, name: str, description: str, func: Callable, param_model: type[BaseModel] = None ): if param_model: schema = param_model.model_json_schema() parameters = { "type": "object", "properties": schema.get("properties", {}), "required": schema.get("required", []) } else: # Infer from function signature  sig = inspect.signature(func) parameters = {"type": "object", "properties": {}, "required": []} for param_name, param in sig.parameters.items(): if param_name in ("self", "cls"): continue parameters["properties"][param_name] = { "type": self._python_type_to_json(param.annotation) } if param.default == inspect.Parameter.empty: parameters["required"].append(param_name) self.register(ToolSpec( name=name, description=description, parameters=parameters, function=func, return_type=inspect.signature(func).return_annotation )) def _python_type_to_json(self, annotation) -> str: type_map = { str: "string", int: "integer", float: "number", bool: "boolean", list: "array", dict: "object" } return type_map.get(annotation, "string") def get_openai_tools(self) -> List[dict]: return [ { "type": "function", "function": { "name": spec.name, "description": spec.description, "parameters": spec.parameters } } for spec in self.tools.values() ] async def execute(self, name: str, arguments: dict) -> Any: if name not in self.tools: raise ValueError(f"Unknown tool: {name}") spec = self.tools[name] return await spec.function(**arguments) # Example tools async def web_search(query: str, max_results: int = 5) -> dict: # Implementation  return {"results": []} async def code_execute(code: str, language: str = "python") -> dict: # Implementation with sandbox  return {"output": "", "error": None} async def file_read(path: str) -> dict: # Implementation  return {"content": ""} async def file_write(path: str, content: str) -> dict: # Implementation  return {"success": True} registry = ToolRegistry() registry.register_function("web_search", "Search the web", web_search) registry.register_function("code_execute", "Execute code", code_execute) registry.register_function("file_read", "Read a file", file_read) registry.register_function("file_write", "Write a file", file_write) 

# agent/core.py from dataclasses import dataclass, field from typing import List, Optional, Dict, Any from openai import OpenAI import json import asyncio @dataclass class Message: role: str content: str tool_calls: List[dict] = field(default_factory=list) tool_call_id: str = "" @dataclass class AgentState: messages: List[Message] tools: ToolRegistry max_iterations: int = 10 current_iteration: int = 0 context: Dict[str, Any] = field(default_factory=dict) class ToolAgent: def __init__( self, model: str = "gpt-4o", system_prompt: str = "", tools: ToolRegistry = None, max_iterations: int = 10, temperature: float = 0 ): self.client = OpenAI() self.model = model self.system_prompt = system_prompt self.tools = tools or ToolRegistry() self.max_iterations = max_iterations self.temperature = temperature async def run(self, user_input: str, context: dict = None) -> AgentState: state = AgentState( messages=[ Message(role="system", content=self.system_prompt), Message(role="user", content=user_input) ], tools=self.tools, max_iterations=self.max_iterations, context=context or {} ) while state.current_iteration < state.max_iterations: state.current_iteration += 1 # Get model response  response = await self._get_response(state) if not response.tool_calls: # No tool calls = done  state.messages.append(Message( role="assistant", content=response.content )) break # Execute tool calls  state.messages.append(Message( role="assistant", content=response.content or "", tool_calls=response.tool_calls )) for tool_call in response.tool_calls: result = await self._execute_tool(tool_call) state.messages.append(Message( role="tool", Source：https://dev.to/ipsphil_digital/building-ai-agents-with-function-calling-tools-structured-output-agent-loops-26nj
My favorites