Add pydanticAI example of tools

Author: pamelafox

examples/pydanticai_tools.py

@@ -0,0 +1,79 @@
+import asyncio
+import logging
+import os
+import random
+from datetime import datetime
+
+import azure.identity
+from dotenv import load_dotenv
+from openai import AsyncAzureOpenAI, AsyncOpenAI
+from pydantic_ai import Agent
+from pydantic_ai.models.openai import OpenAIModel
+from pydantic_ai.providers.openai import OpenAIProvider
+from rich.logging import RichHandler
+
+# Setup logging with rich
+logging.basicConfig(level=logging.WARNING, format="%(message)s", datefmt="[%X]", handlers=[RichHandler()])
+logger = logging.getLogger("weekend_planner")
+
+
+# Setup the OpenAI client to use either Azure OpenAI or GitHub Models
+load_dotenv(override=True)
+API_HOST = os.getenv("API_HOST", "github")
+
+if API_HOST == "github":
+    client = AsyncOpenAI(api_key=os.environ["GITHUB_TOKEN"], base_url="https://models.inference.ai.azure.com")
+    model = OpenAIModel(os.getenv("GITHUB_MODEL", "gpt-4o"), provider=OpenAIProvider(openai_client=client))
+elif API_HOST == "azure":
+    token_provider = azure.identity.get_bearer_token_provider(azure.identity.DefaultAzureCredential(), "https://cognitiveservices.azure.com/.default")
+    client = AsyncAzureOpenAI(
+        api_version=os.environ["AZURE_OPENAI_VERSION"],
+        azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
+        azure_ad_token_provider=token_provider,
+    )
+    model = OpenAIModel(os.environ["AZURE_OPENAI_CHAT_DEPLOYMENT"], provider=OpenAIProvider(openai_client=client))
+
+
+def get_weather(city: str) -> str:
+    logger.info(f"Getting weather for {city}")
+    if random.random() < 0.05:
+        return {
+            "city": city,
+            "temperature": 72,
+            "description": "Sunny",
+        }
+    else:
+        return {
+            "city": city,
+            "temperature": 60,
+            "description": "Rainy",
+        }
+
+
+def get_activities(city: str, date: str) -> list:
+    logger.info(f"Getting activities for {city} on {date}")
+    return [
+        {"name": "Hiking", "location": city},
+        {"name": "Beach", "location": city},
+        {"name": "Museum", "location": city},
+    ]
+
+def get_current_date() -> str:
+    logger.info("Getting current date")
+    return datetime.now().strftime("%Y-%m-%d")
+
+
+agent = Agent(
+    model,
+    system_prompt="You help users plan their weekends and choose the best activities for the given weather. If an activity would be unpleasant in the weather, don't suggest it. Include the date of the weekend in your response.",
+    tools=[get_weather, get_activities, get_current_date],
+)
+
+async def main():
+    result = await agent.run("what can I do for funzies this weekend in Seattle?")
+    print(result.output)
+
+
+if __name__ == "__main__":
+    logger.setLevel(logging.INFO)
+    asyncio.run(main())