diff --git a/python/packages/autogen-ext/src/autogen_ext/models/openai/_responses_client.py b/python/packages/autogen-ext/src/autogen_ext/models/openai/_responses_client.py index 6e42375fc..c66a808dd 100644 --- a/python/packages/autogen-ext/src/autogen_ext/models/openai/_responses_client.py +++ b/python/packages/autogen-ext/src/autogen_ext/models/openai/_responses_client.py @@ -123,14 +123,18 @@ from typing_extensions import Unpack from .._utils.normalize_stop_reason import normalize_stop_reason from . import _model_info -from ._openai_client import azure_openai_client_from_config as _azure_openai_client_from_config # noqa: F401 +from ._openai_client import ( + azure_openai_client_from_config as _azure_openai_client_from_config, # noqa: F401 # pyright: ignore[reportUnusedImport] +) from ._openai_client import ( convert_tools, normalize_name, ) # Backward-compatible private aliases for tests that patch private symbols -from ._openai_client import openai_client_from_config as _openai_client_from_config # noqa: F401 +from ._openai_client import ( + openai_client_from_config as _openai_client_from_config, # noqa: F401 # pyright: ignore[reportUnusedImport] +) from .config import ( AzureOpenAIClientConfiguration, OpenAIClientConfiguration, diff --git a/python/packages/autogen-ext/src/autogen_ext/tools/graphrag/__init__.py b/python/packages/autogen-ext/src/autogen_ext/tools/graphrag/__init__.py index 01e13b678..d58a9ae7d 100644 --- a/python/packages/autogen-ext/src/autogen_ext/tools/graphrag/__init__.py +++ b/python/packages/autogen-ext/src/autogen_ext/tools/graphrag/__init__.py @@ -1,5 +1,7 @@ # Compatibility shim for OpenAI SDK type location changes used by transitive deps (e.g., fnllm) try: + from typing import Any, cast + from openai.types.chat import ( chat_completion_message_function_tool_call as _func_mod, ) @@ -10,12 +12,16 @@ try: chat_completion_message_tool_call_param as _tool_param_mod, ) + _func_mod_any = cast(Any, _func_mod) + _tool_mod_any = cast(Any, _tool_mod) + _tool_param_mod_any = cast(Any, _tool_param_mod) + # Ensure Function exists on the tool_call module - if not hasattr(_tool_mod, "Function") and hasattr(_func_mod, "Function"): - setattr(_tool_mod, "Function", _func_mod.Function) + if not hasattr(_tool_mod_any, "Function") and hasattr(_func_mod_any, "Function"): + _tool_mod_any.Function = _func_mod_any.Function # pyright: ignore[reportAttributeAccessIssue] # Ensure Function exists on the tool_call_param module (some libs import from here) - if not hasattr(_tool_param_mod, "Function") and hasattr(_func_mod, "Function"): - setattr(_tool_param_mod, "Function", _func_mod.Function) + if not hasattr(_tool_param_mod_any, "Function") and hasattr(_func_mod_any, "Function"): + _tool_param_mod_any.Function = _func_mod_any.Function # pyright: ignore[reportAttributeAccessIssue] except Exception: # Best-effort shim; safe to ignore if modules are unavailable pass diff --git a/python/samples/gpt5_examples/gpt5_agent_integration.py b/python/samples/gpt5_examples/gpt5_agent_integration.py index d7cdba78f..2f7a7e55c 100644 --- a/python/samples/gpt5_examples/gpt5_agent_integration.py +++ b/python/samples/gpt5_examples/gpt5_agent_integration.py @@ -16,27 +16,40 @@ This showcases enterprise-grade patterns for GPT-5 integration. import asyncio import os -from typing import Any, Dict, List +from typing import Any, Dict, Literal, Optional -from autogen_agentchat.agents import AssistantAgent -from autogen_agentchat.teams import SelectorGroupChat from autogen_core import CancellationToken from autogen_core.models import UserMessage from autogen_core.tools import BaseCustomTool, CustomToolFormat from autogen_ext.models.openai import OpenAIChatCompletionClient, OpenAIResponsesAPIClient +from pydantic import BaseModel +import json -class DataAnalysisTool(BaseCustomTool[str]): +class TextResult(BaseModel): + text: str + + +def _coerce_content_to_text(content: object) -> str: + if isinstance(content, str): + return content + try: + return json.dumps(content, ensure_ascii=False, default=str) + except Exception: + return str(content) + + +class DataAnalysisTool(BaseCustomTool[TextResult]): """GPT-5 custom tool for data analysis with freeform input.""" def __init__(self): super().__init__( - return_type=str, + return_type=TextResult, name="data_analysis", description="Analyze data and generate insights. Input should be data description or analysis request.", ) - async def run(self, input_text: str, cancellation_token: CancellationToken) -> str: + async def run(self, input_text: str, cancellation_token: CancellationToken) -> TextResult: """Simulate data analysis.""" # In production, this would connect to data analysis tools analysis_types = { @@ -52,29 +65,33 @@ class DataAnalysisTool(BaseCustomTool[str]): analysis_type = key break - return f"Data Analysis Results:\n{analysis_types[analysis_type]}\n\nDetailed analysis: {input_text}" + return TextResult(text=f"Data Analysis Results:\n{analysis_types[analysis_type]}\n\nDetailed analysis: {input_text}") -class ResearchTool(BaseCustomTool[str]): +class ResearchTool(BaseCustomTool[TextResult]): """GPT-5 custom tool for research tasks.""" def __init__(self): super().__init__( - return_type=str, + return_type=TextResult, name="research", description="Conduct research and gather information on specified topics.", ) - async def run(self, input_text: str, cancellation_token: CancellationToken) -> str: + async def run(self, input_text: str, cancellation_token: CancellationToken) -> TextResult: """Simulate research functionality.""" - return f"šŸ” Research Results for: {input_text}\n" \ - f"ā€¢ Found 15 relevant academic papers\n" \ - f"ā€¢ Identified 3 key trends\n" \ - f"ā€¢ Generated comprehensive summary with citations\n" \ - f"ā€¢ Confidence level: High" + return TextResult( + text=( + f"šŸ” Research Results for: {input_text}\n" + f"ā€¢ Found 15 relevant academic papers\n" + f"ā€¢ Identified 3 key trends\n" + f"ā€¢ Generated comprehensive summary with citations\n" + f"ā€¢ Confidence level: High" + ) + ) -class CodeReviewTool(BaseCustomTool[str]): +class CodeReviewTool(BaseCustomTool[TextResult]): """GPT-5 custom tool with grammar constraints for code review.""" def __init__(self): @@ -105,33 +122,40 @@ class CodeReviewTool(BaseCustomTool[str]): ) super().__init__( - return_type=str, + return_type=TextResult, name="code_review", description="Review code with structured input. Format: REVIEW LANG:python CODE:your_code TYPE:security", format=code_review_grammar, ) - async def run(self, input_text: str, cancellation_token: CancellationToken) -> str: + async def run(self, input_text: str, cancellation_token: CancellationToken) -> TextResult: """Perform structured code review.""" - return f"šŸ“ Code Review Complete:\n" \ - f"Input: {input_text}\n" \ - f"āœ… No security vulnerabilities found\n" \ - f"āš” Performance suggestions: Use list comprehension\n" \ - f"šŸŽØ Style: Follows PEP 8 guidelines\n" \ - f"šŸ› No bugs detected\n" \ - f"Overall: Production ready" + return TextResult( + text=( + f"šŸ“ Code Review Complete:\n" + f"Input: {input_text}\n" + f"āœ… No security vulnerabilities found\n" + f"āš” Performance suggestions: Use list comprehension\n" + f"šŸŽØ Style: Follows PEP 8 guidelines\n" + f"šŸ› No bugs detected\n" + f"Overall: Production ready" + ) + ) + + +ReasoningEffort = Literal["minimal", "low", "medium", "high"] class GPT5ReasoningAgent: """Assistant agent optimized for GPT-5 reasoning tasks.""" - def __init__(self, name: str, reasoning_effort: str = "high"): + def __init__(self, name: str, reasoning_effort: ReasoningEffort = "high"): self.name = name self.client = OpenAIChatCompletionClient( model="gpt-5", api_key=os.getenv("OPENAI_API_KEY", "your-api-key-here") ) - self.reasoning_effort = reasoning_effort + self.reasoning_effort: ReasoningEffort = reasoning_effort # Configure for reasoning tasks self.system_message = """ @@ -156,7 +180,7 @@ class GPT5ReasoningAgent: preambles=True ) - return response.content + return _coerce_content_to_text(response.content) class GPT5CodeAgent: @@ -195,7 +219,7 @@ class GPT5CodeAgent: preambles=True # Explain code choices ) - return response.content + return _coerce_content_to_text(response.content) class GPT5AnalysisAgent: @@ -235,7 +259,7 @@ class GPT5AnalysisAgent: preambles=True ) - return response.content + return _coerce_content_to_text(response.content) class GPT5ConversationManager: @@ -246,10 +270,10 @@ class GPT5ConversationManager: model="gpt-5", api_key=os.getenv("OPENAI_API_KEY", "your-api-key-here") ) - self.conversation_history = [] - self.last_response_id = None + self.conversation_history: list[dict[str, Any]] = [] + self.last_response_id: Optional[str] = None - async def continue_conversation(self, user_input: str, reasoning_effort: str = "medium") -> Dict[str, Any]: + async def continue_conversation(self, user_input: str, reasoning_effort: ReasoningEffort = "medium") -> Dict[str, Any]: """Continue conversation with CoT preservation.""" response = await self.client.create( input=user_input, @@ -262,7 +286,7 @@ class GPT5ConversationManager: # Update conversation state self.conversation_history.append({ "user_input": user_input, - "response": response.content, + "response": _coerce_content_to_text(response.content), "reasoning": response.thought, "response_id": getattr(response, 'response_id', None) }) @@ -270,7 +294,7 @@ class GPT5ConversationManager: self.last_response_id = getattr(response, 'response_id', None) return { - "content": response.content, + "content": _coerce_content_to_text(response.content), "reasoning": response.thought, "usage": response.usage, "turn_number": len(self.conversation_history) @@ -479,7 +503,7 @@ async def demonstrate_tool_specialization(): preambles=True # Explain tool restrictions ) - print(f"Agent Response: {response.content}") + print(f"Agent Response: {_coerce_content_to_text(response.content)}") if response.thought: print(f"Tool Usage Explanation: {response.thought}") diff --git a/python/samples/gpt5_examples/gpt5_basic_usage.py b/python/samples/gpt5_examples/gpt5_basic_usage.py index 6c39a7e4f..76348549d 100644 --- a/python/samples/gpt5_examples/gpt5_basic_usage.py +++ b/python/samples/gpt5_examples/gpt5_basic_usage.py @@ -17,45 +17,76 @@ Run this script to see GPT-5 features in action. import asyncio import os -from typing import List +from typing import Literal from autogen_core import CancellationToken from autogen_core.models import UserMessage from autogen_core.tools import BaseCustomTool, CustomToolFormat from autogen_ext.models.openai import OpenAIChatCompletionClient, OpenAIResponsesAPIClient +from pydantic import BaseModel +import json -class CodeExecutorTool(BaseCustomTool[str]): +class TextResult(BaseModel): + text: str + + +def _coerce_content_to_text(content: object) -> str: + if isinstance(content, str): + return content + try: + return json.dumps(content, ensure_ascii=False, default=str) + except Exception: + return str(content) + + +ReasoningEffort = Literal["minimal", "low", "medium", "high"] + + +class CodeExecutorTool(BaseCustomTool[TextResult]): """GPT-5 custom tool for executing Python code with freeform text input.""" def __init__(self): super().__init__( - return_type=str, + return_type=TextResult, name="code_exec", description="Executes Python code and returns the output. Input should be valid Python code.", ) - async def run(self, input_text: str, cancellation_token: CancellationToken) -> str: + async def run(self, input_text: str, cancellation_token: CancellationToken) -> TextResult: """Execute Python code safely (in a real implementation, use proper sandboxing).""" try: # In production, use proper sandboxing like RestrictedPython or containers # This is a simplified example import io - import sys from contextlib import redirect_stdout output = io.StringIO() with redirect_stdout(output): - exec(input_text, {"__builtins__": {"print": print, "len": len, "str": str, "int": int, "float": float}}) + exec( + input_text, + { + "__builtins__": { + "print": print, + "len": len, + "str": str, + "int": int, + "float": float, + } + }, + ) result = output.getvalue() - return f"Code executed successfully:\n{result}" if result else "Code executed successfully (no output)" + text = ( + f"Code executed successfully:\n{result}" if result else "Code executed successfully (no output)" + ) + return TextResult(text=text) - except Exception as e: - return f"Error executing code: {str(e)}" + except Exception as e: # noqa: BLE001 + return TextResult(text=f"Error executing code: {e}") -class SQLQueryTool(BaseCustomTool[str]): +class SQLQueryTool(BaseCustomTool[TextResult]): """GPT-5 custom tool with grammar constraints for SQL queries.""" def __init__(self): @@ -63,7 +94,7 @@ class SQLQueryTool(BaseCustomTool[str]): sql_grammar = CustomToolFormat( type="grammar", syntax="lark", - definition=""" + definition=r""" start: select_statement select_statement: "SELECT" column_list "FROM" table_name where_clause? @@ -89,43 +120,46 @@ class SQLQueryTool(BaseCustomTool[str]): %import common.WS %ignore WS - """ + """, ) super().__init__( - return_type=str, + return_type=TextResult, name="sql_query", description="Execute SQL SELECT queries with grammar validation. Only SELECT statements are allowed.", format=sql_grammar, ) - async def run(self, input_text: str, cancellation_token: CancellationToken) -> str: + async def run(self, input_text: str, cancellation_token: CancellationToken) -> TextResult: """Simulate SQL query execution.""" # In a real implementation, this would connect to a database # This is a mock response for demonstration - return f"SQL Query Results:\nExecuted: {input_text}\nResult: [Mock data returned - 3 rows affected]" + return TextResult( + text=( + f"SQL Query Results:\nExecuted: {input_text}\nResult: [Mock data returned - 3 rows affected]" + ) + ) -class CalculatorTool(BaseCustomTool[str]): +class CalculatorTool(BaseCustomTool[TextResult]): """Simple calculator tool for safe mathematical operations.""" def __init__(self): super().__init__( - return_type=str, + return_type=TextResult, name="calculator", - description="Perform basic mathematical calculations safely. Input should be a mathematical expression.", + description=( + "Perform basic mathematical calculations safely. Input should be a mathematical expression." + ), ) - async def run(self, input_text: str, cancellation_token: CancellationToken) -> str: + async def run(self, input_text: str, cancellation_token: CancellationToken) -> TextResult: """Safely evaluate mathematical expressions.""" try: - # Simple safe evaluation for basic math - import re import ast import operator - # Only allow safe mathematical operations - allowed_ops = { + allowed_ops: dict[type[ast.AST], object] = { ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, @@ -135,33 +169,32 @@ class CalculatorTool(BaseCustomTool[str]): ast.USub: operator.neg, } - def safe_eval(node): + def safe_eval(node: ast.AST) -> float | int: if isinstance(node, ast.Expression): - return safe_eval(node.body) - elif isinstance(node, ast.Num): - return node.n - elif isinstance(node, ast.Constant): - return node.value - elif isinstance(node, ast.BinOp): + return safe_eval(node.body) # type: ignore[arg-type] + if isinstance(node, ast.Constant): + if isinstance(node.value, (int, float)): + return node.value + raise ValueError("Only numeric constants are allowed") + if isinstance(node, ast.BinOp): left = safe_eval(node.left) right = safe_eval(node.right) op = allowed_ops.get(type(node.op)) if op: - return op(left, right) - elif isinstance(node, ast.UnaryOp): + return op(left, right) # type: ignore[call-arg] + if isinstance(node, ast.UnaryOp): operand = safe_eval(node.operand) op = allowed_ops.get(type(node.op)) if op: - return op(operand) - + return op(operand) # type: ignore[call-arg] raise ValueError(f"Unsupported operation: {type(node)}") - tree = ast.parse(input_text, mode='eval') + tree = ast.parse(input_text, mode="eval") result = safe_eval(tree) - return f"Calculation result: {result}" + return TextResult(text=f"Calculation result: {result}") - except Exception as e: - return f"Error in calculation: {str(e)}" + except Exception as e: # noqa: BLE001 + return TextResult(text=f"Error in calculation: {e}") async def demonstrate_gpt5_basic_usage(): @@ -173,7 +206,7 @@ async def demonstrate_gpt5_basic_usage(): # Initialize GPT-5 client client = OpenAIChatCompletionClient( model="gpt-5", - api_key=os.getenv("OPENAI_API_KEY", "your-api-key-here") + api_key=os.getenv("OPENAI_API_KEY", "your-api-key-here"), ) # Example 1: Basic reasoning with different effort levels @@ -184,14 +217,14 @@ async def demonstrate_gpt5_basic_usage(): response = await client.create( messages=[UserMessage( content="Explain the concept of quantum entanglement and its implications for quantum computing", - source="user" + source="user", )], reasoning_effort="high", verbosity="medium", - preambles=True + preambles=True, ) - print(f"High reasoning response: {response.content}") + print(f"High reasoning response: {_coerce_content_to_text(response.content)}") if response.thought: print(f"Reasoning process: {response.thought}") @@ -199,13 +232,13 @@ async def demonstrate_gpt5_basic_usage(): response = await client.create( messages=[UserMessage( content="What's 2 + 2?", - source="user" + source="user", )], reasoning_effort="minimal", - verbosity="low" + verbosity="low", ) - print(f"Minimal reasoning response: {response.content}") + print(f"Minimal reasoning response: {_coerce_content_to_text(response.content)}") await client.close() @@ -218,13 +251,12 @@ async def demonstrate_gpt5_custom_tools(): client = OpenAIChatCompletionClient( model="gpt-5", - api_key=os.getenv("OPENAI_API_KEY", "your-api-key-here") + api_key=os.getenv("OPENAI_API_KEY", "your-api-key-here"), ) # Initialize custom tools code_tool = CodeExecutorTool() sql_tool = SQLQueryTool() - calc_tool = CalculatorTool() print("\n2. Custom Tool with Freeform Input:") print("-" * 40) @@ -233,15 +265,15 @@ async def demonstrate_gpt5_custom_tools(): response = await client.create( messages=[UserMessage( content="Calculate the factorial of 8 using Python code", - source="user" + source="user", )], tools=[code_tool], reasoning_effort="medium", verbosity="low", - preambles=True # Explain why tools are used + preambles=True, # Explain why tools are used ) - print(f"Tool response: {response.content}") + print(f"Tool response: {_coerce_content_to_text(response.content)}") if response.thought: print(f"Tool explanation: {response.thought}") @@ -252,14 +284,14 @@ async def demonstrate_gpt5_custom_tools(): response = await client.create( messages=[UserMessage( content="Query all users from the users table where age is greater than 25", - source="user" + source="user", )], tools=[sql_tool], reasoning_effort="low", - preambles=True + preambles=True, ) - print(f"SQL response: {response.content}") + print(f"SQL response: {_coerce_content_to_text(response.content)}") await client.close() @@ -272,7 +304,7 @@ async def demonstrate_allowed_tools(): client = OpenAIChatCompletionClient( model="gpt-5", - api_key=os.getenv("OPENAI_API_KEY", "your-api-key-here") + api_key=os.getenv("OPENAI_API_KEY", "your-api-key-here"), ) # Create multiple tools @@ -289,16 +321,16 @@ async def demonstrate_allowed_tools(): response = await client.create( messages=[UserMessage( content="I need help with calculations, database queries, and code execution", - source="user" + source="user", )], tools=all_tools, allowed_tools=safe_tools, # Restrict to only calculator tool_choice="auto", reasoning_effort="medium", - preambles=True + preambles=True, ) - print(f"Restricted response: {response.content}") + print(f"Restricted response: {_coerce_content_to_text(response.content)}") if response.thought: print(f"Tool restriction explanation: {response.thought}") @@ -314,7 +346,7 @@ async def demonstrate_responses_api(): # Use the Responses API for better performance in multi-turn conversations client = OpenAIResponsesAPIClient( model="gpt-5", - api_key=os.getenv("OPENAI_API_KEY", "your-api-key-here") + api_key=os.getenv("OPENAI_API_KEY", "your-api-key-here"), ) print("\n5. Multi-Turn Conversation with CoT Preservation:") @@ -326,10 +358,10 @@ async def demonstrate_responses_api(): input="Design a distributed system architecture for a real-time chat application that can handle millions of users", reasoning_effort="high", verbosity="medium", - preambles=True + preambles=True, ) - print(f"Response 1: {response1.content}") + print(f"Response 1: {_coerce_content_to_text(response1.content)}") if response1.thought: print(f"Reasoning 1: {response1.thought[:200]}...") @@ -339,10 +371,10 @@ async def demonstrate_responses_api(): input="How would you handle data consistency in this distributed system?", previous_response_id=getattr(response1, 'response_id', None), # Preserve CoT context reasoning_effort="medium", # Can use lower effort due to context - verbosity="medium" + verbosity="medium", ) - print(f"Response 2: {response2.content}") + print(f"Response 2: {_coerce_content_to_text(response2.content)}") # Turn 3: Implementation request with tools print("\nTurn 3: Implementation with custom tools") @@ -353,10 +385,10 @@ async def demonstrate_responses_api(): previous_response_id=getattr(response2, 'response_id', None), tools=[code_tool], reasoning_effort="low", # Minimal reasoning needed due to established context - preambles=True + preambles=True, ) - print(f"Response 3: {response3.content}") + print(f"Response 3: {_coerce_content_to_text(response3.content)}") if response3.thought: print(f"Implementation explanation: {response3.thought}") @@ -375,19 +407,19 @@ async def demonstrate_model_variants(): # GPT-5 (full model) gpt5_client = OpenAIChatCompletionClient( model="gpt-5", - api_key=os.getenv("OPENAI_API_KEY", "your-api-key-here") + api_key=os.getenv("OPENAI_API_KEY", "your-api-key-here"), ) # GPT-5 Mini (cost-optimized) gpt5_mini_client = OpenAIChatCompletionClient( model="gpt-5-mini", - api_key=os.getenv("OPENAI_API_KEY", "your-api-key-here") + api_key=os.getenv("OPENAI_API_KEY", "your-api-key-here"), ) # GPT-5 Nano (high-throughput) gpt5_nano_client = OpenAIChatCompletionClient( model="gpt-5-nano", - api_key=os.getenv("OPENAI_API_KEY", "your-api-key-here") + api_key=os.getenv("OPENAI_API_KEY", "your-api-key-here"), ) question = "Briefly explain machine learning" @@ -397,27 +429,27 @@ async def demonstrate_model_variants(): response = await gpt5_client.create( messages=[UserMessage(content=question, source="user")], reasoning_effort="medium", - verbosity="medium" + verbosity="medium", ) - print(f" {response.content[:100]}...") + print(f" {_coerce_content_to_text(response.content)[:100]}...") print(f" Token usage: {response.usage.prompt_tokens + response.usage.completion_tokens}") print("\nGPT-5 Mini (cost-optimized):") response = await gpt5_mini_client.create( messages=[UserMessage(content=question, source="user")], reasoning_effort="medium", - verbosity="medium" + verbosity="medium", ) - print(f" {response.content[:100]}...") + print(f" {_coerce_content_to_text(response.content)[:100]}...") print(f" Token usage: {response.usage.prompt_tokens + response.usage.completion_tokens}") print("\nGPT-5 Nano (high-throughput):") response = await gpt5_nano_client.create( messages=[UserMessage(content=question, source="user")], reasoning_effort="minimal", - verbosity="low" + verbosity="low", ) - print(f" {response.content[:100]}...") + print(f" {_coerce_content_to_text(response.content)[:100]}...") print(f" Token usage: {response.usage.prompt_tokens + response.usage.completion_tokens}") await gpt5_client.close() @@ -451,7 +483,7 @@ async def main(): print("ā€¢ Responses API optimizes multi-turn conversations with CoT preservation") print("ā€¢ Different model variants (gpt-5, gpt-5-mini, gpt-5-nano) balance performance and cost") - except Exception as e: + except Exception as e: # noqa: BLE001 print(f"\nāŒ Error running examples: {e}") print("Make sure you have:") print("1. Set OPENAI_API_KEY environment variable")