### Install Pydantic and Verify Version Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Pydantic/BaseModel.md Installs the Pydantic library and verifies the installed version using pip and a simple Python script. Ensure you have pydantic-settings installed for environment file support. ```bash pip install pydantic>=2 # For settings via env files: pip install pydantic-settings ``` ```python import pydantic print(pydantic.VERSION) ``` -------------------------------- ### Recommended Prompt Prefix for Reliable Agent Handoffs in Python Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/OpenAISDK/02-handsoff.md Provides an example of how to integrate the `RECOMMENDED_PROMPT_PREFIX` into an agent's instructions to ensure reliable handoff behavior. By including this prefix, the LLM is guided to correctly interpret and execute handoff instructions, improving the robustness of the system. ```python from agents import Agent from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX billing_agent = Agent( name="Billing agent", instructions=f"{RECOMMENDED_PROMPT_PREFIX}\ You are a billing support agent. Handle invoices, refunds, and subscriptions." ) ``` -------------------------------- ### Default Agent Initialization (Python) Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/OpenAISDK/01-defaults.md A concise example showing how to instantiate an Agent object with all its default settings applied, requiring only a name. ```python agent = Agent(name="Default") # gets every default above ``` -------------------------------- ### Basic Agent Handoff Setup in Python Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/OpenAISDK/02-handsoff.md Demonstrates the basic setup for agent handoffs using the Agents SDK. It shows how to create specialized agents (Billing, Refund) and a Triage agent that can hand off tasks to them. The handoff is configured by including the target agents in the `handoffs` list of the source agent. ```python from agents import Agent, handoff billing_agent = Agent(name="Billing agent") refund_agent = Agent(name="Refund agent") # Triage agent can hand off to billing or refund triage_agent = Agent( name="Triage agent", handoffs=[billing_agent, handoff(refund_agent)] ) ``` -------------------------------- ### Voice Integration Installation (Bash) Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/OpenAISDK/01-defaults.md Provides the command-line instruction to install the necessary dependencies for enabling voice capabilities within the OpenAI Agents SDK. ```bash pip install 'openai-agents[voice]' ``` -------------------------------- ### Python: Structure Nested Models and Collections Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Pydantic/BaseModel.md Illustrates how to define and use nested Pydantic models and collections (like lists). This example shows an `Order` model containing a list of `Item` models, demonstrating how Pydantic handles nested data structures. ```python from pydantic import BaseModel, Field from typing import List class Item(BaseModel): sku: str qty: int = Field(gt=0) class Order(BaseModel): order_id: str items: List[Item] Order(order_id='A1', items=[{'sku': 'X', 'qty': 2}]) # dicts → Item instances ``` -------------------------------- ### Install OpenAI Agents SDK and Set API Key Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Course/Orchestration/OrchestrationDiff.md Installs the necessary OpenAI Agents SDK package using pip and sets the OpenAI API key as an environment variable. This is a prerequisite for using the SDK. ```bash pip install openai-agents export OPENAI_API_KEY=sk-... ``` -------------------------------- ### Pydantic Field Constraints with Field and Annotated Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Pydantic/BaseModel.md Demonstrates how to apply constraints to model fields using `Field` and `Annotated`. This example defines constraints for string length, numerical range (greater than 0, greater than or equal to 0, less than or equal to 1), and specifies a default value for the `discount` field. ```python from pydantic import BaseModel, Field from typing import Annotated PosInt = Annotated[int, Field(gt=0)] SizedStr = Annotated[str, Field(min_length=3, max_length=50)] class Product(BaseModel): name: SizedStr price: Annotated[float, Field(ge=0)] discount: Annotated[float, Field(ge=0, le=1)] = 0.0 ``` -------------------------------- ### Voice Configuration Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/OpenAISDK/01-defaults.md Options for configuring text-to-speech and speech-to-text capabilities, requiring an optional installation. ```APIDOC ## Voice Configuration ### Description Configuration options for voice features, including speech synthesis and recognition. ### Method Setting voice-related attributes during `Agent` instantiation or via configuration. ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body - **model** (string) - Optional - The voice model to use. Defaults to `"gpt-4o-audio-preview"`. - **voice** (string) - Optional - The specific voice to use for synthesis. Defaults to `"alloy"`. - **format** (string) - Optional - The audio format for the output. Defaults to `"wav"`. ### Installation ```bash pip install 'openai-agents[voice]' ``` ### Request Example ```python from agents import Agent # Example with custom voice model, voice, and format voice_agent = Agent(name="VoiceBot", model="gpt-4o-audio", voice="echo", format="mp3") ``` ### Response #### Success Response (200) Agent configured with voice capabilities. #### Response Example ```json { "message": "Voice configuration applied" } ``` ``` -------------------------------- ### Add a Tool (Python Function) to an Agent Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Course/Orchestration/OrchestrationDiff.md Shows how to equip an agent with a tool, which is a real Python function. The example defines a `get_weather` function and adds it to a `WeatherBot` agent, which can then answer weather-related queries. ```python from agents import function_tool @function_tool def get_weather(city: str) -> str: return f"{city} is 22 °C and sunny." weather_bot = Agent( name="WeatherBot", instructions="Answer weather questions.", tools=[get_weather] ) print(Runner.run_sync(weather_bot, "Weather in Paris?")) # Paris is 22 °C and sunny. ``` -------------------------------- ### Python: Validate Cross-Field Logic with Model Validator Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Pydantic/BaseModel.md Shows how to implement cross-field validation logic within a Pydantic model using `@model_validator`. This example ensures that a discount is only applied when the price is positive. ```python from pydantic import BaseModel, model_validator class Discount(BaseModel): price: float discount: float # 0..1 @model_validator(mode='after') def check_discount(self): if self.discount > 0 and self.price <= 0: raise ValueError('discount requires positive price') return self ``` -------------------------------- ### Python Code Execution Example Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/slides/pythonInfo/runProcess.md This snippet demonstrates a simple Python code that iterates and performs addition. It represents the human-readable source code that an interpreter processes. ```python x = 0 for i in range(10**6): x += i ``` -------------------------------- ### Python Asyncio.gather Example Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Course/asyncio/.gather.md Demonstrates how to use asyncio.gather to run multiple asynchronous tasks concurrently. Tasks are defined with a name and delay, and asyncio.gather waits for all of them to complete. The output shows the order of completion, highlighting concurrent execution. ```python import asyncio async def task(name, delay): await asyncio.sleep(delay) print(f"{name} finished after {delay}s") async def main(): await asyncio.gather( task("A", 2), task("B", 3), task("C", 1), ) asyncio.run(main()) ``` -------------------------------- ### Pydantic BaseModel Basic Usage Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Pydantic/BaseModel.md Demonstrates the fundamental usage of Pydantic's BaseModel for defining data structures. It shows how to create a model with type hints, instantiate it with data, and access its attributes. Pydantic automatically coerces compatible types, such as strings to integers. ```python from pydantic import BaseModel class User(BaseModel): id: int email: str is_active: bool = True u = User(id='1', email='a@b.com') # coerces '1' -> 1 print(u) # User id=1 email='a@b.com' is_active=True print(u.id + 1) # 2 ``` -------------------------------- ### Python: Configure Aliases and Populate by Name Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Pydantic/BaseModel.md Shows how to use field aliases and the `populate_by_name` configuration in Pydantic. This allows models to accept data using either the alias or the model field name, enhancing flexibility in data input. ```python from pydantic import BaseModel, Field, ConfigDict class Customer(BaseModel): model_config = ConfigDict(populate_by_name=True) full_name: str = Field(alias='fullName') c = Customer(fullName='Ada Lovelace') # by alias c2 = Customer(full_name='Ada Lovelace') # by name (due to populate_by_name) ``` -------------------------------- ### Python: Create Generic Models with TypeVar Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Pydantic/BaseModel.md Shows how to define generic Pydantic models using `typing.TypeVar`. This allows for creating models that can hold values of any specified type, promoting code reusability. ```python from typing import Generic, TypeVar from pydantic import BaseModel T = TypeVar('T') class Box(BaseModel, Generic[T]): item: T int_box = Box[int](item=1) str_box = Box[str](item='x') ``` -------------------------------- ### Python: Validate Types with TypeAdapter Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Pydantic/BaseModel.md Demonstrates `TypeAdapter` for validating Python types directly without defining a Pydantic `BaseModel`. This is efficient for validating simple types or collections in performance-critical code paths. ```python from pydantic import TypeAdapter from typing import Annotated from pydantic import Field Tags = Annotated[list[str], Field(min_items=1, max_items=10)] ta = TypeAdapter(Tags) print(ta.validate_python(['a', 'b'])) # validated list[str] ``` -------------------------------- ### Get Typed JSON Output from Agents using Pydantic Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Course/Orchestration/OrchestrationDiff.md Demonstrates how to configure an agent to return structured data in the form of a Pydantic model. This allows for robust and type-safe extraction of information, like addresses, without needing regex. ```python from agents import Agent, Runner from pydantic import BaseModel class Address(BaseModel): street: str city: str extractor = Agent( name="Extractor", instructions="Extract the address.", output_type=Address ) addr = Runner.run_sync(extractor, "I live at 42 Maple St, Boston").final_output print(addr.city) # Boston ``` -------------------------------- ### Python: Handle Dates, Times, and Timezones Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Pydantic/BaseModel.md Shows how Pydantic automatically parses ISO-formatted date and time strings into `datetime` objects. It emphasizes the importance of timezone-aware datetimes for accuracy, recommending UTC storage and local time presentation. ```python from datetime import datetime, timezone from pydantic import BaseModel class Event(BaseModel): at: datetime # ISO strings parse nicely Event(at='2025-09-07T10:00:00Z') # Make datetimes timezone‑aware for correctness now_utc = datetime.now(timezone.utc) ``` -------------------------------- ### Python Asyncio Example: Concurrent Task Execution Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Course/asyncio/base.md Demonstrates how to define and run asynchronous tasks concurrently using Python's asyncio library. It showcases the use of 'async def' for coroutines, 'await asyncio.sleep()' to yield control, and 'asyncio.gather' to run multiple tasks. ```python import asyncio async def task(name, delay): print(f"{name} started") await asyncio.sleep(delay) # release control print(f"{name} finished") async def main(): await asyncio.gather( task("A", 2), task("B", 3), ) asyncio.run(main()) ``` -------------------------------- ### Filtering Conversation History for Agent Handoffs in Python Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/OpenAISDK/02-handsoff.md Demonstrates how to use input filters to control the conversation history passed during an agent handoff. The example uses `handoff_filters.remove_all_tools` to exclude all previous tool calls from the history sent to the target agent. This is useful for ensuring a clean context for the next agent. ```python from agents import Agent, handoff from agents.extensions import handoff_filters faq_agent = Agent(name="FAQ agent") handoff_obj = handoff( agent=faq_agent, input_filter=handoff_filters.remove_all_tools ) ``` -------------------------------- ### Python: Serialize Models to Dict and JSON Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Pydantic/BaseModel.md Demonstrates Pydantic's serialization methods: `model_dump()` to convert a model instance to a dictionary, and `model_dump_json()` to serialize it into a JSON string. Options for excluding `None` values and using aliases are also shown. ```python from pydantic import BaseModel class Profile(BaseModel): name: str age: int | None = None p = Profile(name='Lee') print(p.model_dump()) # dict print(p.model_dump(exclude_none=True)) # omit nulls print(p.model_dump_json(by_alias=True)) # JSON string ``` -------------------------------- ### Python: Use Enums, Literals, and Unions Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Pydantic/BaseModel.md Explains how to use Python's `Enum`, `Literal`, and `Union` types within Pydantic models for robust data definition. It also demonstrates discriminated unions using `Annotated` for type-safe handling of different model types based on a discriminator field. ```python from enum import Enum from typing import Literal, Union, Annotated from pydantic import BaseModel, Field class Role(str, Enum): admin = 'admin' user = 'user' class Cat(BaseModel): kind: Literal['cat'] lives: int class Dog(BaseModel): kind: Literal['dog'] good_boy: bool Pet = Annotated[Union[Cat, Dog], Field(discriminator='kind')] class ZooEntry(BaseModel): pet: Pet ``` -------------------------------- ### Agent Initialization with Defaults (Python) Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/OpenAISDK/01-defaults.md Demonstrates the creation of an Agent instance using all factory default settings and a simple synchronous run to test its basic functionality. ```python from agents import Agent, Runner agent = Agent(name="Echo") print(Runner.run_sync(agent, "ping")).final_output # -> "pong" ``` -------------------------------- ### Python: Normalize Email with Field Validator Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Pydantic/BaseModel.md Illustrates using a `@field_validator` in Pydantic to normalize a specific field. This example normalizes an email address by stripping whitespace and converting to lowercase, and also validates its format. ```python from pydantic import BaseModel, field_validator class Signup(BaseModel): email: str @field_validator('email') @classmethod def normalize_email(cls, v: str) -> str: v = v.strip().lower() if '@' not in v: raise ValueError('invalid email') return v ``` -------------------------------- ### Python: Basic Agent Initialization and Execution Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/OpenAISDK/00-intro.md Demonstrates the fundamental usage of the OpenAI Agents SDK to initialize an agent with instructions and execute a task synchronously. It requires the `openai-agents` library and the `OPENAI_API_KEY` environment variable to be set. ```python from agents import Agent, Runner agent = Agent(name="Assistant", instructions="You are a helpful assistant") result = Runner.run_sync(agent, "Write a haiku about recursion in programming.") print(result.final_output) ``` -------------------------------- ### Create and Run a Basic Agent with System Prompt Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Course/Orchestration/OrchestrationDiff.md Demonstrates how to create a simple agent with a system prompt defining its persona and task. It then runs the agent synchronously to solve a math problem and prints the result. ```python from agents import Agent, Runner math_bot = Agent( name="MathBot", instructions="You are a friendly math tutor. Always show the steps.", model="gpt-4o-mini" ) # Run it result = Runner.run_sync(math_bot, "solve 3x+5=17") print(result.final_output) # x = 4, here’s how… ``` -------------------------------- ### Implement Input and Output Guardrails in Python Source: https://context7.com/muhammadariyan/open-ai-sdk-journey/llms.txt Illustrates setting up input and output guardrails for AI agents to enforce content filtering and safety checks, using Pydantic models for structured validation. ```python from agents import ( Agent, Runner, AsyncOpenAI, OpenAIChatCompletionsModel, input_guardrail, output_guardrail, GuardrailFunctionOutput, InputGuardrailTripwireTriggered, OutputGuardrailTripwireTriggered, RunContextWrapper, TResponseInputItem ) from pydantic import BaseModel # Guardrail output type class IsAboutDrugs(BaseModel): is_about_drugs: bool ``` -------------------------------- ### Memory Demonstration with SQLiteSession (Python) Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/OpenAISDK/01-defaults.md Shows how to implement persistent memory for an agent using SQLiteSession. It demonstrates saving information and recalling it in subsequent interactions. ```python from agents import SQLiteSession, Runner, Agent agent = Agent(name="Memo") sess = SQLiteSession("alice") Runner.run_sync(agent, "My dog is Max", session=sess) print(Runner.run_sync(agent, "What's my dog's name?", session=sess).final_output) # -> "Your dog's name is Max." ``` -------------------------------- ### Dynamic Agent Instructions with Context in Python Source: https://context7.com/muhammadariyan/open-ai-sdk-journey/llms.txt Creates dynamic agent instructions that adapt based on the provided user context, enabling personalized agent behavior and tool usage suggestions. ```python from agents import Agent, Runner, function_tool from agents.agent import RunContextWrapper class UserContext: def __init__(self, name: str, age: int, preferences: list, context: list): self.name = name self.age = age self.preferences = preferences self.context = context # Context-aware tools @function_tool async def fetch_user_age(wrapper: RunContextWrapper[UserContext]) -> str: """Fetch the age of the user.""" return f"you are {wrapper.context.name} who is {wrapper.context.age} year(s) old." @function_tool async def change_user_age(wrapper: RunContextWrapper[UserContext], newAge: str) -> str: """Change the age of the user.""" wrapper.context.age = newAge return f"So, Now your age is {wrapper.context.age} year(s) old." # Dynamic instructions based on context async def DynamicInstruction( context: RunContextWrapper[UserContext], agent: Agent[UserContext] ): return ( f"you are jenny. The user name is {context.context.name}. " f"Help them with {context.context}. Use fetch_user_age tool for age queries." ) # Conditional handoff based on preferences def LikesSummary(context: RunContextWrapper[UserContext], agent_base: Agent) -> bool: return 'Likes Summary' in context.context.preferences # Summarizer agent Summarizer = Agent( name='Summarize', handoff_description='You are a Summarizer', instructions='Summarize the input and nothing more.', output_type={"summary": str} ) # Dynamic agent with context-aware tools DynamicAgent = Agent[UserContext]( name='Dynamic Agent', instructions=DynamicInstruction, tools=[ Summarizer.as_tool( tool_name='Summarizer', tool_description='summarize the input and answer' ), fetch_user_age, change_user_age ] ) # Recipe wizard sub-agent RecipeWizard = Agent( name='Recipe Wizard', handoff_description='Returns a recipe', instructions='You return a recipe based on the request' ) # Chef agent using RecipeWizard as tool ChefAgent = Agent( name='chef', instructions='You are a chef. Use Making_Recipe tool first to get recipe.', tools=[ RecipeWizard.as_tool( tool_name='Making_Recipe', tool_description='Get recipe for cooking' ) ] ) # Run with user context # result = await Runner.run( # starting_agent=DynamicAgent, # input="What is my age?", # run_config=config, # context=UserContext( # name='ARY PAGLU', # age=17, # preferences=['Likes Summary'], # context=['Loves to cook and bake'] # ) # ) # print(result.final_output) ``` -------------------------------- ### Testing Pydantic Models for Validation Errors Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Pydantic/BaseModel.md Tests that a Pydantic model raises a `ValidationError` when provided with invalid data. This ensures that guardrails and business rules are enforced correctly. ```python import pytest from pydantic import ValidationError from myapp.models import Order def test_order_requires_qty_positive(): with pytest.raises(ValidationError): Order(order_id='X', items=[{'sku': 'A', 'qty': 0}]) ``` -------------------------------- ### Create Reusable Function Tools in Python Source: https://context7.com/muhammadariyan/open-ai-sdk-journey/llms.txt Demonstrates the creation of custom, reusable function tools ('punch', 'kick') that can be integrated into AI agents for specific actions. ```python from agents import Agent, Runner, function_tool # Define custom function tools @function_tool async def punch(person: str): """Function to punch.""" return f"ye le mukka kha! sun beta {person} mera mukaa khatarnak hai!" @function_tool async def kick(person: str): """Function to kick.""" return f"ye le laat kha! sun beta {person} mera laat khatarnak hai!" ``` -------------------------------- ### Clone Agents with Modified Settings in Python Source: https://context7.com/muhammadariyan/open-ai-sdk-journey/llms.txt Shows how to clone an existing agent ('champaklal') to create a new agent ('tapu') with modified instructions and model settings, enabling variations of an agent. ```python from agents import Agent, Runner, function_tool, ModelSettings from agents.agent import StopAtTools from pydantic import BaseModel # Define custom function tools @function_tool async def punch(person: str): """Function to punch.""" return f"ye le mukka kha! sun beta {person} mera mukaa khatarnak hai!" @function_tool async def kick(person: str): """Function to kick.""" return f"ye le laat kha! sun beta {person} mera laat khatarnak hai!" # Custom output type class CustomOutput(BaseModel): name: str quote: str joke: str # Model settings ModelSettingBase = ModelSettings( temperature=0.7, tool_choice='required', parallel_tool_calls=True ) ModelSettingFinal = ModelSettings(temperature=0.3) # Base agent with tools champaklal = Agent( name="Champaklal", handoff_description="You are Champaklal from TMKOC", instructions=( "You are Champaklal from TMKOC. " "If someone says punch or mukka, use the punch tool. " "If someone says kick or laat, use the kick tool." ), tools=[punch, kick], tool_use_behavior=StopAtTools(stop_at_tool_names=['punch']) ) # Clone agent with modified settings tapu = champaklal.clone( name="Tapu", instructions=[ "You are Tapu from TMKOC. If talked about champaklal, " "then handsoff to champaklal agent" ], model_settings=ModelSettingBase.resolve(ModelSettingFinal) ) # Parent agent with handoffs jhetalal = Agent( name="Jhetalal", instructions=( "You are Jhetalal from TMKOC. " "If talked about champaklal, handoff to champaklal agent" ), output_type=CustomOutput, handoffs=[champaklal, tapu] ) # Execute with multiple agents # result = await Runner.run( # champaklal, # 'use punch tool to tapu and kick tool to jhetia', # run_config=config # ) # print(result.final_output) ``` -------------------------------- ### Custom Field and Model Serialization in Pydantic Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Pydantic/BaseModel.md Customizes how specific fields or entire models are serialized. `field_serializer` formats individual fields, while `model_serializer` defines custom output for the whole model. ```python from pydantic import BaseModel, field_serializer, model_serializer from datetime import datetime class Audit(BaseModel): actor: str at: datetime @field_serializer('at') def ser_at(self, v: datetime) -> str: return v.isoformat(timespec='seconds') class Money(BaseModel): currency: str amount: int # store cents internally @model_serializer def ser_model(self): return {self.currency: self.amount / 100} ``` -------------------------------- ### ORM Mode Validation with Pydantic Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Pydantic/BaseModel.md Validates objects like ORM instances by reading attributes instead of mapping dictionary keys. Uses `ConfigDict(from_attributes=True)` to enable ORM mode. ```python from pydantic import BaseModel, ConfigDict class UserDTO(BaseModel): model_config = ConfigDict(from_attributes=True) id: int username: str class ORMUser: def __init__(self, id, username): self.id = id self.username = username u = ORMUser(1, 'neo') UserDTO.model_validate(u) # reads attributes instead of mapping keys ``` -------------------------------- ### Customizing Agent Handoffs in Python Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/OpenAISDK/02-handsoff.md Illustrates how to customize agent handoffs with specific configurations. This includes overriding the default tool name and description, and defining a callback function (`on_handoff`) that executes when the handoff is triggered. The `handoff` function accepts parameters for customization. ```python from agents import Agent, handoff, RunContextWrapper def on_handoff(ctx: RunContextWrapper[None]): print("Handoff triggered!") target_agent = Agent(name="Target agent") handoff_obj = handoff( agent=target_agent, on_handoff=on_handoff, tool_name_override="custom_tool", tool_description_override="Custom description", ) ``` -------------------------------- ### Tracing & Observability Configuration Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/OpenAISDK/01-defaults.md Settings for enabling and configuring tracing and observability for agent runs. ```APIDOC ## Tracing & Observability Configuration ### Description Configuration options for tracing agent execution for observability. ### Method Setting environment variables or parameters during `Runner` execution. ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body - **provider** (string) - Optional - The tracing provider. Defaults to `"console"`. - **sample_rate** (float) - Optional - The rate at which traces are sampled. Defaults to `1.0` (100%). - **environment variable `TRACE_ENDPOINT`** - Optional - The endpoint for the tracing service. - **Exporters** - Optional - Additional exporters can be installed via pip (e.g., `logfire`). ### Request Example ```python # Using environment variable for endpoint export TRACE_ENDPOINT="https://your-tracing-service.com" # Using sample rate parameter (if available in Runner config or via environment) # Example assuming a direct runner config option: # Runner.run(agent, "Message", sample_rate=0.1) # Install exporter pip install logfire ``` ### Response #### Success Response (200) Agent execution can be traced and observed. #### Response Example ```json { "message": "Tracing configured successfully" } ``` ``` -------------------------------- ### Implement Input and Output Guardrails for Agents in Python Source: https://context7.com/muhammadariyan/open-ai-sdk-journey/llms.txt This snippet demonstrates how to create and apply input and output guardrail agents to an existing agent. It includes defining guardrail agents, applying them as decorators to check functions, and handling potential guardrail violations during agent execution. The guardrails are designed to detect and block specific types of input or output, such as drug-related content. ```python from agents import Agent, Runner, AgentHooks, function_tool from agents.guardrails import input_guardrail, output_guardrail, GuardrailFunctionOutput, IsAboutDrugs, InputGuardrailTripwireTriggered, OutputGuardrailTripwireTriggered from runners import RunContextWrapper from typing import TResponseInputItem, TypeVar # Assume config and Runner are defined elsewhere # Example placeholder for config: config = {} # Input guardrail agent inputGuardrail = Agent( name='InputGuardrail for Drugs', instructions='Detect if the user is asking about drugs', output_type=IsAboutDrugs ) # Input guardrail decorator @input_guardrail async def check_for_drugs( ctx: RunContextWrapper[IsAboutDrugs], agent: Agent, input: str | list[TResponseInputItem] ) -> None: runInputGuardrail = await Runner.run( inputGuardrail, input, context=ctx.context, run_config=config ) return GuardrailFunctionOutput( output_info=runInputGuardrail.final_output, tripwire_triggered=runInputGuardrail.final_output.is_about_drugs ) # Output guardrail agent outputGuardrail = Agent( name='OutputGuardrail for Drugs', instructions='Check if there is any information about drugs.', output_type=IsAboutDrugs ) # Output guardrail decorator @output_guardrail async def check_output_for_drugs( ctx: RunContextWrapper[IsAboutDrugs], agent: Agent, input: str | list[TResponseInputItem] ) -> None: runOutputGuardrail = await Runner.run( outputGuardrail, input, context=ctx.context, run_config=config ) return GuardrailFunctionOutput( output_info=runOutputGuardrail.final_output, tripwire_triggered=runOutputGuardrail.final_output.is_about_drugs ) # Agent with guardrails AgentPaglu = Agent( name='Paglu', instructions='You are a helpful assistant.', input_guardrails=[check_for_drugs], output_guardrails=[check_output_for_drugs] ) # Handle guardrail violations async def run_agent_with_guardrails(): try: result = await Runner.run( starting_agent=AgentPaglu, run_config=config, input='Tell me about marijuana' ) print(result.final_output) except InputGuardrailTripwireTriggered: print('Input Guardrail Tripwire Triggered - Blocked harmful input') except OutputGuardrailTripwireTriggered: print('Output Guardrail Tripwire Triggered - Blocked harmful output') # To run this example, you would need to set up an async context and call run_agent_with_guardrails() # import asyncio # asyncio.run(run_agent_with_guardrails()) ``` -------------------------------- ### Python: Handle Pydantic Validation Errors Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Pydantic/BaseModel.md Demonstrates how to catch and process Pydantic's `ValidationError` using the `.errors()` method, which returns structured error information. This is useful for providing detailed feedback to clients. ```python from pydantic import BaseModel, ValidationError, Field class NegativeOnly(BaseModel): value: int = Field(..., lt=0) try: NegativeOnly(value=5) except ValidationError as e: print(e.errors()) # [{'type': 'less_than', 'loc': ('value',), ...}] ``` -------------------------------- ### Python: Use RootModel for Single Value Validation Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Pydantic/BaseModel.md Introduces `RootModel` for validating and wrapping a single value, rather than a dictionary-like structure. This is useful for types like strings with specific patterns or lists with constraints. ```python from pydantic import BaseModel, RootModel, Field from typing import Annotated SlugStr = Annotated[str, Field(pattern=r'^[a-z0-9-]+$')] class Slug(RootModel[SlugStr]): pass s = Slug('hello-world') # OK # Slug('Hello') # ❌ pattern fails ``` -------------------------------- ### Python: Define Computed Field for Derived Values Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Pydantic/BaseModel.md Demonstrates the creation of computed fields in Pydantic using the `@computed_field` decorator. These fields are read-only and their values are derived from other model attributes, included automatically in model dumps. ```python from pydantic import BaseModel, computed_field class Cart(BaseModel): quantity: int unit_price: float @computed_field @property def total(self) -> float: return round(self.quantity * self.unit_price, 2) c = Cart(quantity=3, unit_price=19.99) assert c.total == 59.97 ``` -------------------------------- ### Sessions (Memory) Configuration Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/OpenAISDK/01-defaults.md Information on configuring session persistence for agent memory using different backends like SQLite. ```APIDOC ## Sessions (Memory) Configuration ### Description Configuring memory persistence for agents using the `session` parameter in the `Runner`. ### Method Passing a `Session` object to `Runner.run()` or `Runner.run_sync()`. ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body - **session** (Session object) - Optional - A session object for managing conversation history. Defaults to `None` (no memory). - **SQLiteSession**: Stores memory in an SQLite database. - **user_id** (string) - Required - Identifier for the user session. - **db_path** (string) - Optional - Path to the SQLite database file. Defaults to `"sessions.db"`. ### Request Example ```python from agents import Runner, Agent, SQLiteSession agent = Agent(name="MemoAgent") # Use SQLiteSession with default db path session_default_db = SQLiteSession("user_123") Runner.run_sync(agent, "First message", session=session_default_db) # Use SQLiteSession with custom db path session_custom_db = SQLiteSession("user_456", db_path="chat_history.db") Runner.run_sync(agent, "Second message", session=session_custom_db) ``` ### Response #### Success Response (200) Conversation history is persisted and retrieved correctly. #### Response Example ```json { "final_output": "Agent's response incorporating memory" } ``` ``` -------------------------------- ### Implement Agent Lifecycle Hooks in Python Source: https://context7.com/muhammadariyan/open-ai-sdk-journey/llms.txt This code defines custom agent hooks by extending the `AgentHooks` class. It provides implementations for `on_start`, `on_end`, `on_Tool_start`, and `on_Tool_end` to log events during an agent's execution. An agent is then configured with these custom hooks and a sample tool, demonstrating how to integrate lifecycle monitoring into agent workflows. ```python from agents import Agent, Runner, AgentHooks, function_tool from runners import RunContextWrapper from typing import TResponseInputItem # Assume config is defined elsewhere # Example placeholder for config: config = {} # Custom agent hooks class SastaAgentHooks(AgentHooks): async def on_start(self, context, agent) -> None: """Called before the agent is invoked.""" print(f"[HOOK] Agent {agent.name} is starting.") async def on_end(self, context, agent, output) -> None: """Called when the agent produces a final output.""" print(f"[HOOK] Agent {agent.name} has finished with output: {output}") async def on_handoff(self, context, agent, source) -> None: """Called when the agent is being handed off to.""" print(f"[HOOK] Agent {source.name} is handing off to agent {agent.name}.") async def on_tool_start(self, context, agent, tool) -> None: """Called before a tool is invoked.""" print(f"[HOOK] Agent {agent.name} is starting tool {tool.name}.") async def on_tool_end(self, context, agent, tool, result: str) -> None: """Called after a tool is invoked.""" print(f"[HOOK] Agent {agent.name} finished tool {tool.name} with result: {result}") @function_tool def get_name(context: dict) -> str: return f"Your name is {context.get('name', 'unknown')}." # Agent with hooks AgentPaglu = Agent( name='Paglu', instructions='You are a helpful assistant.', hooks=SastaAgentHooks(), tools=[get_name] ) async def run_agent_with_hooks(): result = await Runner.run( starting_agent=AgentPaglu, run_config=config, input='What is my name?' ) print(result.final_output) # To run this example, you would need to set up an async context and call run_agent_with_hooks() # import asyncio # asyncio.run(run_agent_with_hooks()) ``` -------------------------------- ### Guardrails Configuration Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/OpenAISDK/01-defaults.md Details on configuring Guardrails for input and output validation and filtering. ```APIDOC ## Guardrails Configuration ### Description Configuration for `Guardrails` to enforce rules on agent input or output. ### Method Setting the `guardrails` parameter during `Agent` instantiation or using `Guardrails` directly. ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body - **input** (object) - Optional - Guardrail for input processing. Example: `BlockProfanity()`. - **output** (object) - Optional - Guardrail for output processing. Example: `RequireJsonSchema(...)`. ### Request Example ```python from agents import Agent, Guardrails from agents.guardrails import BlockProfanity, RequireJsonSchema # Input guardrail guarded_agent_input = Agent(name="SafeAgent", guardrails=Guardrails(input=BlockProfanity())) # Output guardrail guarded_agent_output = Agent(name="JsonAgent", guardrails=Guardrails(output=RequireJsonSchema({...}))) ``` ### Response #### Success Response (200) Agent configured with specified guardrails. #### Response Example ```json { "message": "Agent configured with guardrails" } ``` ``` -------------------------------- ### Chain Agents with Handoffs for Complex Workflows Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Course/Orchestration/OrchestrationDiff.md Illustrates how to create a multi-agent system where agents can hand off tasks to each other. A `Triage` agent directs math-related questions to `MathBot` and weather questions to `WeatherBot`. ```python from agents import Agent, Runner # Assuming math_bot and weather_bot are defined as in previous examples # math_bot = Agent(...) # weather_bot = Agent(...) triage = Agent( name="Triage", instructions="If the user asks about math, hand off to MathBot. If weather, hand off to WeatherBot.", handoffs=[math_bot, weather_bot] ) print(Runner.run_sync(triage, "What’s 9*6?")) # → MathBot answers ``` -------------------------------- ### Agent Configuration Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/OpenAISDK/01-defaults.md Defines the configuration options for an Agent, including name, instructions, model, and various parameters for controlling generation behavior and tool usage. ```APIDOC ## Agent Configuration ### Description Configuration options for creating an `Agent` instance. ### Method Instantiation of `Agent` class. ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body - **name** (string) - Required - The name of the agent. - **instructions** (string) - Optional - Instructions for the agent's behavior. Defaults to an empty string. - **model** (string) - Optional - The language model to use. Defaults to `"gpt-4o"`. - **temperature** (float) - Optional - Controls randomness. Defaults to `1.0`. - **top_p** (float) - Optional - Controls diversity. Defaults to `1.0`. - **max_tokens** (integer) - Optional - Maximum tokens for the response. Defaults to `None` (unlimited). - **tools** (list) - Optional - A list of tools the agent can use. Defaults to `[]`. - **handoffs** (list) - Optional - A list of other agents the current agent can hand off to. Defaults to `[]`. - **guardrails** (Guardrails object) - Optional - Guardrails to apply to agent interactions. Defaults to `None`. - **output_type** (schema object) - Optional - Expected output schema. Defaults to `None` (plain text). - **context** (dict) - Optional - Contextual data for the agent. Defaults to `{}`. ### Request Example ```python from agents import Agent agent = Agent(name="MathBot", instructions="Solve every problem step by step.", model="gpt-4o-mini", temperature=0.3, max_tokens=512, tools=[get_weather], handoffs=[spanish_agent], guardrails=Guardrails(input=BlockProfanity()), output_type=WeatherSchema, context={"user_id": "42"}) ``` ### Response #### Success Response (200) Agent instance configured with provided parameters. #### Response Example ```json { "message": "Agent configured successfully" } ``` ``` -------------------------------- ### Pydantic Strict Types for Disabling Coercion Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Pydantic/BaseModel.md Illustrates the use of strict types in Pydantic to prevent data coercion. By using `StrictInt` and `StrictBool`, the model will only accept values of the exact specified type, raising a `ValidationError` for incompatible types like strings being passed to an integer field. ```python from pydantic import BaseModel, StrictInt, StrictBool class S(BaseModel): a: StrictInt b: StrictBool S(a=1, b=True) # OK S(a='1', b='true') # ❌ ValidationError ``` -------------------------------- ### Pydantic Field Defaults and Optional Fields Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/Pydantic/BaseModel.md Shows how to define fields with default values and optional fields in Pydantic BaseModel. `default_factory` is used to generate a default value dynamically (e.g., a UUID for an ID), while `Optional[str] = None` makes the `subtitle` field allow `None` and be omittable. ```python from pydantic import BaseModel, Field from typing import Optional from uuid import uuid4 class Doc(BaseModel): id: str = Field(default_factory=lambda: str(uuid4())) title: str subtitle: Optional[str] = None # Optional = may be omitted ``` -------------------------------- ### Agent Handoff with Structured Input in Python Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/OpenAISDK/02-handsoff.md Shows how to implement agent handoffs that require structured input data. A Pydantic model (`EscalationData`) defines the expected input structure, and the `on_handoff` callback function is designed to receive and process this structured data. The `input_type` parameter in the `handoff` function specifies the data model. ```python from pydantic import BaseModel from agents import Agent, handoff, RunContextWrapper class EscalationData(BaseModel): reason: str async def on_handoff(ctx: RunContextWrapper[None], input_data: EscalationData): print(f"Handoff called with reason: {input_data.reason}") escalation_agent = Agent(name="Escalation agent") handoff_obj = handoff( agent=escalation_agent, on_handoff=on_handoff, input_type=EscalationData, ) ``` -------------------------------- ### Handoff to Spanish Agent (Python) Source: https://github.com/muhammadariyan/open-ai-sdk-journey/blob/main/OpenAISDK/01-defaults.md Illustrates the handoff mechanism in the Agents SDK, where an agent ('Triage') routes a user query to a specific language agent ('ES' for Spanish) based on the input language. ```python from agents import Agent, Runner triage = Agent( name="Triage", instructions="Route to the correct language agent.", handoffs=[Agent(name="ES", instructions="Only Spanish."), Agent(name="EN", instructions="Only English.")] ) print(Runner.run_sync(triage, "Hola")).final_output # Spanish reply ```