Building AI Agents with Function Calling: Tools, Structured Output & Agent Loops

Master function calling for AI agents — from basic tool use to complex multi-step agents with planning, memory, and error recovery.

Function calling transforms LLMs from text generators into agents that act. This guide covers production patterns for tool use, structured output, agent loops, and orchestration.

# 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 ) Enter fullscreen mode Exit fullscreen mode Structured Output with JSON Schema # 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 )) Enter fullscreen mode Exit fullscreen mode Part 2: Building a Tool-Enabled Agent Tool Registry # 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 ) Enter fullscreen mode Exit fullscreen mode Agent Core Loop # 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 ", content = json. dumps ( result ), tool_call_id = tool_call [ " id " ] )) return state async def _get_response ( self, state: AgentState ) -> Any: messages = [ { " role ": m. role, " content ": m. content, " tool_calls ": m. tool_calls } for m in state. messages ] response = self. client. chat. completions. create ( model = self. model, messages = messages, tools = self. tools. get_openai_tools (), tool_choice = " auto ", temperature = self. temperature, parallel_tool_calls = True ) return response. choices [ 0 ]. message async def _execute_tool ( self, tool_call: dict ) -> dict: name = tool_call. function. name args = json. loads ( tool_call. function. arguments ) try: result = await self. tools. execute ( name, args ) return { " success ": True, " result ": result } except Exception as e: return { " success ": False, " error ": str ( e )} # Usage SYSTEM_PROMPT = """ You are a capable AI agent with access to tools. Plan your approach, use tools efficiently, and provide clear results. Always explain your reasoning before taking actions.

""" agent = ToolAgent ( system_prompt = SYSTEM_PROMPT, tools = registry, max_iterations = 15 ) result = await agent. run ( " Find the latest Python 3.12 features and create a summary file " ) Enter fullscreen mode Exit fullscreen mode Part 3: Advanced Agent Patterns ReAct Pattern (Reasoning + Acting) # agent/react.py from dataclasses import dataclass from typing import List import re REACT_PROMPT = """ Answer the following questions as best you can. You have access to the following tools: {tools} Use the following format: Question: the input question you must answer Thought: you should always think about what to do Action: the action to take, should be one of [{tool_names}] Action Input: the input to the action Observation: the result of the action... ( this Thought / Action / Action Input / Observation can repeat N times ) Thought: I now know the final answer Final Answer: the final answer to the original input question Begin! Question: {input} Thought: """ class ReActAgent: def __init__ ( self, llm, tools: ToolRegistry, max_steps: int = 10 ): self. llm = llm self. tools = tools self. max_steps = max_steps async def run ( self, question: str ) -> str: tool_descriptions = " \n ". join ( f " { spec. name }: { spec. description } " for spec in self. tools. tools. values () ) tool_names = ", ". join ( self. tools. tools. keys ()) prompt = REACT_PROMPT. format ( tools = tool_descriptions, tool_names = tool_names, input = question ) for step in range ( self. max_steps ): response = await self. llm. ainvoke ( prompt ) # Parse action action_match = re. search ( r " Action:\s*(\w+) ", response ) input_match = re. search ( r " Action Input:\s*(.+) ", response ) final_match = re. search ( r " Final Answer:\s*(.+) ", response, re.

DOTALL ) if final_match: return final_match. group ( 1 ). strip () if action_match and input_match: tool_name = action_match. group ( 1 ) tool_input = json. loads ( input_match. group ( 1 )) result = await self. tools. execute ( tool_name, tool_input ) prompt += f " \n { response } \n Observation: { json. dumps ( result ) } \n Thought: " else: prompt += f " \n { response } \n Thought: " return " Max steps reached without final answer " Enter fullscreen mode Exit fullscreen mode Plan-and-Execute Agent # agent/plan_execute.py from dataclasses import dataclass from typing import List, Optional from pydantic import BaseModel, Field class PlanStep ( BaseModel ): step_number: int description: str tool: str args: dict expected_output: str depends_on: List [ int ] = Field ( default_factory = list ) class ExecutionPlan ( BaseModel ): goal: str steps: List [ PlanStep ] success_criteria: str class PlanAndExecuteAgent: def __init__ ( self, planner_llm, executor_llm, tools: ToolRegistry ): self. planner = planner_llm self. executor = executor_llm self. tools = tools async def run ( self, goal: str ) -> dict: # Phase 1: Plan plan = await self. _create_plan ( goal ) # Phase 2: Execute results = {} for step in plan. steps: # Check dependencies for dep in step. depends_on: if dep not in results: raise ValueError ( f " Dependency step { dep } not completed " ) # Execute step result = await self. _execute_step ( step, results ) results [ step. step_number ] = result # Check if we should continue if not result. get ( " success " ): return { " success ": False, " plan ": plan, " results ": results, " failed_step ": step. step_number } # Phase 3: Verify verification = await self. _verify ( plan, results ) return { " success ": verification [ " success " ], " plan ": plan, " results ": results, " verification ": verification } async def _create_plan ( self, goal: str ) -> ExecutionPlan: planning_prompt = f """ Create a detailed execution plan for: { goal } Available tools: { list ( self. tools. tools. keys ()) } Return a plan with numbered steps, each specifying: - tool to use - arguments - expected output - dependencies on other steps Output as JSON matching ExecutionPlan schema. """ return await self. planner. with_structured_output ( ExecutionPlan ). ainvoke ( planning_prompt ) async def _execute_step ( self, step: PlanStep, previous_results: dict ) -> dict: # Substitute variables from previous results args = self. _substitute_variables ( step. args, previous_results ) try: result = await self. tools. execute ( step. tool, args ) return { " success ": True, " output ": result } except Exception as e: return { " success ": False, " error ": str ( e )} def _substitute_variables ( self, args: dict, results: dict ) -> dict: # Replace {{step_N.output.field}} with actual values import re def replace ( match ): ref = match. group ( 1 ) step_num, * path = ref. split ( ". " ) step_num = int ( step_num ) if step_num in results: value = results [ step_num ][ " output " ] for p in path: value = value. get ( p, "" ) return str ( value ) return match. group ( 0 ) substituted = {} for k, v in args. items (): if isinstance ( v, str ): substituted [ k ] = re. sub ( r " \{\{step_(\d+(?:\.\w+)*)\}\} ", replace, v ) else: substituted [ k ] = v return substituted Enter fullscreen mode Exit fullscreen mode Self-Correcting Agent # agent/self_correcting.py from dataclasses import dataclass from typing import Optional import asyncio @dataclass class Attempt: iteration: int plan: str actions: list result: dict error: Optional [ str ] = None reflection: str = "" class SelfCorrectingAgent: def __init__ ( self, llm, tools: ToolRegistry, max_attempts: int = 3, reflection_llm = None ): self. llm = llm self. tools = tools self. max_attempts = max_attempts self. reflection_llm = reflection_llm or llm async def run ( self, goal: str ) -> dict: attempts = [] for attempt_num in range ( 1, self. max_attempts + 1 ): # Create plan (incorporating reflections from previous attempts) plan = await self. _create_plan ( goal, attempts ) # Execute result = await self. _execute_plan ( plan ) attempt = Attempt ( iteration = attempt_num, plan = plan, actions = result. get ( " actions ", []), result = result, error = result. get ( " error " ) ) attempts. append ( attempt ) if result. get ( " success " ): return { " success ": True, " attempts ": attempts, " final_result ": result } # Reflect on failure if attempt_num < self. max_attempts: reflection = await self. _reflect ( goal, attempt ) attempt. reflection = reflection return { " success ": False, " attempts ": attempts } async def _create_plan ( self, goal: str, attempts: list ) -> str: reflection_context = "" if attempts: reflection_context = " \n\n Previous attempts and reflections: \n " for a in attempts: reflection_context += f " Attempt { a. iteration }: { a. reflection } \n " prompt = f """ Goal: { goal } { reflection_context } Create a plan to achieve the goal.

Learn from previous failures. Available tools: { list ( self. tools. tools. keys ()) } """ return await self. llm. ainvoke ( prompt ) async def _execute_plan ( self, plan: str ) -> dict: # Execute using tool agent agent = ToolAgent ( llm = self. llm, tools = self. tools ) state = await agent. run ( plan ) # Check success last_msg = state. messages [ - 1 ] return { " success ": " error " not in last_msg. content. lower () if last_msg. content else False, " actions ": [ m. tool_calls for m in state. messages if m. tool_calls ], " final_output ": last_msg. content } async def _reflect ( self, goal: str, attempt: Attempt ) -> str: prompt = f """ Goal: { goal } Plan: { attempt. plan } Result: { attempt. result } Error: { attempt. error } Analyze why this failed and provide specific guidance for the next attempt. Focus on: wrong tool choice, incorrect arguments, missing steps, logic errors.

""" return await self. reflection_llm. ainvoke ( prompt ) Enter fullscreen mode Exit fullscreen mode Part 4: Multi-Tool Workflows Parallel Tool Execution # agent/parallel.py import asyncio from dataclasses import dataclass from typing import List @dataclass class ParallelAction: tool: str args: dict dependency: str = None # Tool name this depends on class ParallelAgent: def __init__ ( self, llm, tools: ToolRegistry ): self. llm = llm self. tools = tools async def execute_parallel ( self, actions: List [ ParallelAction ]) -> dict: """ Execute independent actions in parallel. """ # Group by dependencies independent = [ a for a in actions if not a. dependency ] dependent = [ a for a in actions if a. dependency ] # Execute independent first results = {} if independent: tasks = [ self. tools. execute ( a. tool, a. args ) for a in independent ] outputs = await asyncio. gather ( * tasks, return_exceptions = True ) for action, output in zip ( independent, outputs ): results [ action. tool ] = output if not isinstance ( output, Exception ) else { " error ": str ( output )} # Execute dependent for action in dependent: dep_result = results. get ( action. dependency ) if dep_result and " error " not in dep_result: # Substitute dependency result args = self. _inject_dependency ( action. args, action. dependency, dep_result ) output = await self. tools. execute ( action. tool, args ) results [ action. tool ] = output else: results [ action. tool ] = { " error ": f " Dependency { action. dependency } failed " } return results Enter fullscreen mode Exit fullscreen mode Tool Chaining with Validation # agent/chaining.py from pydantic import BaseModel, field_validator from typing import Any, Callable class ToolChainStep ( BaseModel ): tool: str args: dict validator: Optional [ Callable [[ Any ], bool ]] = None on_failure: str = " stop " # "stop", "retry", "fallback" fallback: Optional [ " ToolChainStep " ] = None class ToolChain: def __init__ ( self, tools: ToolRegistry ): self. tools = tools self. steps: List [ ToolChainStep ] = [] def add_step ( self, step: ToolChainStep ): self. steps. append ( step ) return self async def execute ( self, initial_input: dict ) -> dict: context = { " input ": initial_input } for i, step in enumerate ( self. steps ): # Prepare args with context args = self. _prepare_args ( step. args, context ) # Execute with retries max_retries = 2 if step. on_failure == " retry " else 0 for attempt in range ( max_retries + 1 ): try: result = await self. tools. execute ( step. tool, args ) # Validate if step. validator and not step. validator ( result ): raise ValueError ( f " Validation failed for step { i } " ) context [ f " step_ { i } " ] = result break except Exception as e: if attempt == max_retries: if step. on_failure == " fallback " and step. fallback: result = await self. _execute_fallback ( step. fallback, context ) context [ f " step_ { i } " ] = result break else: return { " success ": False, " error ": str ( e ), " step ": i, " context ": context } return { " success ": True, " context ": context } Enter fullscreen mode Exit fullscreen mode Part 5: Memory & State Management Conversation Memory # agent/memory.py from dataclasses import dataclass, field from typing import List, Dict, Any from datetime import datetime import json @dataclass class MemoryEntry: timestamp: str type: str # "user", "assistant", "tool", "observation" content: str metadata: Dict [ str, Any ] = field ( default_factory = dict ) class ConversationMemory: def __init__ ( self, max_entries: int = 100 ): self. entries: List [ MemoryEntry ] = [] self. max_entries = max_entries self. summary: str = "" def add ( self, entry: MemoryEntry ): self. entries. append ( entry ) if len ( self. entries ) > self. max_entries: self. _summarize_old () def _summarize_old ( self ): # Keep recent entries, summarize older ones keep = self. entries [ - 20:] old = self. entries [: - 20 ] if old: # Generate summary of old entries old_text = " \n ". join ( f " { e. type }: { e. content } " for e in old ) self. summary = f " Previous conversation summary: { old_text [: 2000 ] }... " self. entries = keep def get_context ( self, max_tokens: int = 4000 ) -> str: parts = [] if self. summary: parts. append ( self. summary ) for entry in self. entries: parts. append ( f " { entry. type }: { entry. content } " ) context = " \n ". join ( parts ) # Rough token estimation if len ( context ) > max_tokens * 4: # Truncate older entries return context [ - max_tokens * 4:] return context def get_relevant ( self, query: str, k: int = 5 ) -> List [ MemoryEntry ]: # Simple keyword matching (replace with embeddings for production) query_words = set ( query. lower (). split ()) scored = [] for entry in self. entries: content_words = set ( entry. content. lower (). split ()) overlap = len ( query_words & content_words ) if overlap > 0: scored. append (( overlap, entry )) scored. sort ( reverse = True ) return [ e for _, e in scored [: k ]] Enter fullscreen mode Exit fullscreen mode Persistent State (SQLite) # agent/state.py import sqlite3 import json from dataclasses import dataclass from typing import Any, Optional from datetime import datetime @dataclass class AgentState: session_id: str agent_type: str state_data: dict created_at: str updated_at: str class StateManager: def __init__ ( self, db_path: str = " agent_state.db " ): self. conn = sqlite3. connect ( db_path ) self. _init_db () def _init_db ( self ): self. conn. execute ( """ CREATE TABLE IF NOT EXISTS agent_states ( session_id TEXT PRIMARY KEY, agent_type TEXT, state_data TEXT, created_at TEXT, updated_at TEXT ) """ ) self. conn. execute ( """ CREATE TABLE IF NOT EXISTS agent_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT, timestamp TEXT, event_type TEXT, data TEXT, FOREIGN KEY (session_id) REFERENCES agent_states(session_id) ) """ ) self. conn. commit () def save_state ( self, state: AgentState ): self. conn. execute ( """ INSERT OR REPLACE INTO agent_states (session_id, agent_type, state_data, created_at, updated_at) VALUES (?,?,?,?,?) """, ( state. session_id, state. agent_type, json. dumps ( state. state_data ), state. created_at, state. updated_at )) self. conn. commit () def load_state ( self, session_id: str ) -> Optional [ AgentState ]: cursor = self. conn. execute ( " SELECT * FROM agent_states WHERE session_id =?

", ( session_id,) ) row = cursor. fetchone () if row: return AgentState ( session_id = row [ 0 ], agent_type = row [ 1 ], state_data = json. loads ( row [ 2 ]), created_at = row [ 3 ], updated_at = row [ 4 ] ) return None def log_event ( self, session_id: str, event_type: str, data: dict ): self. conn. execute ( " INSERT INTO agent_logs (session_id, timestamp, event_type, data) VALUES (?,?,?,?) ", ( session_id, datetime. utcnow (). isoformat (), event_type, json. dumps ( data )) ) self. conn. commit () def get_logs ( self, session_id: str ) -> list: cursor = self. conn. execute ( " SELECT * FROM agent_logs WHERE session_id =? ORDER BY timestamp ", ( session_id,) ) return cursor. fetchall () Enter fullscreen mode Exit fullscreen mode Part 6: Production Deployment FastAPI Agent Server # server/agent_server.py from fastapi import FastAPI, HTTPException, BackgroundTasks from pydantic import BaseModel from typing import Optional, Dict, Any import uuid app = FastAPI ( title = " AI Agent API " ) class RunRequest ( BaseModel ): goal: str agent_type: str = " tool_agent " context: Optional [ Dict [ str, Any ]] = None max_iterations: int = 10 session_id: Optional [ str ] = None class RunResponse ( BaseModel ): session_id: str success: bool result: Optional [ str ] = None error: Optional [ str ] = None steps: int tokens_used: int @app.post ( " /agent/run ", response_model = RunResponse ) async def run_agent ( request: RunRequest, background_tasks: BackgroundTasks ): session_id = request. session_id or str ( uuid. uuid4 ()) # Load or create agent agent = get_agent ( request. agent_type ) # Load previous state if exists state = state_manager. load_state ( session_id ) if request. session_id else None try: result = await agent. run ( request. goal, context = request. context ) # Save state background_tasks. add_task ( save_state, session_id, result ) return RunResponse ( session_id = session_id, success = True, result = result. messages [ - 1 ]. content if result. messages else None, steps = result. current_iteration, tokens_used = calculate_tokens ( result. messages ) ) except Exception as e: return RunResponse ( session_id = session_id, success = False, error = str ( e ), steps = 0, tokens_used = 0 ) @app.get ( " /agent/state/{session_id} " ) async def get_state ( session_id: str ): state = state_manager. load_state ( session_id ) if not state: raise HTTPException ( 404, " Session not found " ) return state. state_data @app.get ( " /agent/logs/{session_id} " ) async def get_logs ( session_id: str ): return state_manager. get_logs ( session_id ) Enter fullscreen mode Exit fullscreen mode Docker Deployment # Dockerfile FROM python:3.11-slim WORKDIR /app # Install dependencies RUN apt-get update && apt-get install -y \ curl \ && rm -rf /var/lib/apt/lists/ * COPY requirements.txt. RUN pip install --no-cache-dir -r requirements.txt COPY.. # Non-root user RUN useradd -m -u 1000 agent && chown -R agent:agent /app USER agent HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \ CMD curl -f http://localhost:8000/health || exit 1 CMD ["uvicorn", "server.agent_server:app", "--host", "0.0.0.0", "--port", "8000"] Enter fullscreen mode Exit fullscreen mode Part 7: Testing & Evaluation Agent Evaluation Framework # evaluation/agent_eval.py from dataclasses import dataclass from typing import List, Callable import asyncio @dataclass class AgentTestCase: name: str goal: str expected_outcome: str expected_tools: List [ str ] = None max_steps: int = 10 setup: Callable = None teardown: Callable = None class AgentEvaluator: def __init__ ( self, agent_factory: Callable ): self. agent_factory = agent_factory async def evaluate ( self, test_cases: List [ AgentTestCase ]) -> dict: results = [] for tc in test_cases: if tc. setup: await tc. setup () agent = self. agent_factory () result = await agent. run ( tc. goal ) # Evaluate eval_result = await self. _evaluate_single ( tc, result ) results. append ( eval_result ) if tc. teardown: await tc. teardown () return self. _aggregate ( results ) async def _evaluate_single ( self, tc: AgentTestCase, result ) -> dict: # Check success success = result. get ( " success ", False ) # Check tool usage tools_used = set () for msg in result. get ( " messages ", []): if msg. get ( " tool_calls " ): for tc_call in msg [ " tool_calls " ]: tools_used. add ( tc_call [ " function " ][ " name " ]) expected_tools = set ( tc. expected_tools or []) tool_precision = len ( tools_used & expected_tools ) / max ( 1, len ( tools_used )) tool_recall = len ( tools_used & expected_tools ) / max ( 1, len ( expected_tools )) # LLM judge for outcome quality quality = await self. _llm_judge ( tc. goal, tc. expected_outcome, result ) return { " name ": tc. name, " success ": success, " steps ": result. get ( " steps ", 0 ), " tool_precision ": tool_precision, " tool_recall ": tool_recall, " quality_score ": quality } async def _llm_judge ( self, goal: str, expected: str, actual ) -> float: # Use LLM to evaluate quality prompt = f """ Rate the agent ' s result on a scale of 0-1.

Goal: { goal } Expected: { expected } Actual: { actual } Consider: completeness, accuracy, efficiency, clarity. Return only a number between 0 and 1. """ response = await self. llm. ainvoke ( prompt ) try: return float ( response. strip ()) except: return 0.5 Enter fullscreen mode Exit fullscreen mode Decision Framework FUNCTION CALLING PATTERN SELECTION: Simple Q&A with tools? ├─→ Yes → Basic function calling (single turn) └─→ No ├─→ Multi-step with known sequence? │ ├─→ Yes → Plan-and-Execute │ └─→ No │ ├─→ Need to adapt based on results? │ │ ├─→ Yes → ReAct or Self-Correcting │ │ └─→ No → Tool Chain │ │ │ ├─→ Multiple independent tools? │ │ └─→ Yes → Parallel execution │ │ │ └─→ Complex, evolving task → Hybrid (Plan + ReAct + Self-Correct) Enter fullscreen mode Exit fullscreen mode Production Checklist Function Definitions [ ] All tools have clear descriptions and examples [ ] Parameter schemas are strict (Pydantic validated) [ ] Tool responses are structured, not free text [ ] Error responses include actionable information [ ] Idempotent tools marked for safe retry Agent Reliability [ ] Max iteration limits prevent infinite loops [ ] Timeout handling for each tool call [ ] Circuit breakers for external APIs [ ] Graceful degradation when tools fail [ ] State persistence for long-running agents Observability [ ] Tool call logging (input, output, latency, errors) [ ] Token usage tracking per step [ ] Cost tracking per agent run [ ] Success/failure rates by tool [ ] Latency percentiles (p50, p95, p99) Security [ ] Input validation on all tool parameters [ ] Sandbox for code execution tools [ ] Rate limiting on external API calls [ ] Authentication for sensitive tools [ ] Audit logging for all actions Testing [ ] Unit tests for each tool [ ] Integration tests for agent flows [ ] Golden test cases for common tasks [ ] Chaos testing (tool failures, timeouts) [ ] Regression tests in CI/CD Resources OpenAI Function Calling Instructor Library LangGraph Agents ReAct Paper Agent Patterns Function Calling Best Practices Function calling is the bridge between LLM reasoning and real-world action.

Build robust tool interfaces, implement proper error handling, and always evaluate your agents on realistic tasks before deploying.