エージェントコンポーネント¶

agent/agent/
├── __init__.py           # Package initialization
├── myagent.py            # Main agent implementation, including prompts
├── config.py             # Configuration management
├── register.py           # DRAgent / NAT registration (framework-specific)
├── workflow.yaml         # Declarative workflow config for DRAgent (framework-specific)
└── model-metadata.yaml   # Agent metadata configuration 

---
name: agent_name
type: inference
targetType: agenticworkflow
runtimeParameterDefinitions:
  - fieldName: LLM_DEPLOYMENT_ID
    defaultValue: SET_VIA_PULUMI_OR_MANUALLY
    type: string
  - fieldName: LLM_DEFAULT_MODEL
    defaultValue: SET_VIA_PULUMI_OR_MANUALLY
    type: string
  - fieldName: LLM_DEFAULT_MODEL_FRIENDLY_NAME
    defaultValue: SET_VIA_PULUMI_OR_MANUALLY
    type: string
  - fieldName: USE_DATAROBOT_LLM_GATEWAY
    defaultValue: SET_VIA_PULUMI_OR_MANUALLY
    type: string
  - fieldName: MCP_DEPLOYMENT_ID
    defaultValue: SET_VIA_PULUMI_OR_MANUALLY
    type: string
  - fieldName: EXTERNAL_MCP_URL
    defaultValue: SET_VIA_PULUMI_OR_MANUALLY
    type: string
  - fieldName: SESSION_SECRET_KEY
    defaultValue: SET_VIA_PULUMI_OR_MANUALLY
    type: string 

def load_model(code_dir: str) -> tuple[ThreadPoolExecutor, asyncio.AbstractEventLoop]:
    """The agent is instantiated in this function and returned.

    Args:
        code_dir: Path to the agentic workflow directory

    Returns:
        tuple[ThreadPoolExecutor, asyncio.AbstractEventLoop]: Thread pool executor and event loop for async operations
    """
    thread_pool_executor = ThreadPoolExecutor(1)
    event_loop = asyncio.new_event_loop()
    thread_pool_executor.submit(asyncio.set_event_loop, event_loop).result()
    return (thread_pool_executor, event_loop) 

def chat(
    completion_create_params: CompletionCreateParams
    | CompletionCreateParamsNonStreaming
    | CompletionCreateParamsStreaming,
    load_model_result: tuple[ThreadPoolExecutor, asyncio.AbstractEventLoop],
    **kwargs: Any,
) -> Union[CustomModelChatResponse, Iterator[CustomModelStreamingResponse]]:
    """Main entry point for agent execution via chat endpoint.

    Args:
        completion_create_params: OpenAI-compatible completion parameters
        load_model_result: Result from load_model() function
        **kwargs: Additional keyword arguments (e.g., headers)

    Returns:
        Union[CustomModelChatResponse, Iterator[CustomModelStreamingResponse]]: Formatted response with agent output
    """ 

class MyAgent:
    """Agent implementation following DataRobot patterns."""

    def __init__(
        self,
        api_key: Optional[str] = None,
        api_base: Optional[str] = None,
        model: Optional[str] = None,
        verbose: Optional[Union[bool, str]] = True,
        timeout: Optional[int] = 90,
        **kwargs: Any,
    ):
        """Initialize agent with credentials and configuration."""
        self.api_key = api_key or os.environ.get("DATAROBOT_API_TOKEN")
        self.api_base = api_base or os.environ.get("DATAROBOT_ENDPOINT")
        self.model = model
        self.timeout = timeout
        # ... other initialization

    @property
    def llm(self) -> LLM:  # Framework-specific type
        """Primary LLM configuration."""
        if os.environ.get("LLM_DEPLOYMENT_ID"):
            return self.llm_with_datarobot_deployment
        else:
            return self.llm_with_datarobot_llm_gateway

    def invoke(self, completion_create_params: CompletionCreateParams) -> Union[
        Generator[tuple[str, Any | None, dict[str, int]], None, None],
        tuple[str, Any | None, dict[str, int]],
    ]:
        """Main execution method - REQUIRED."""
        # Extract inputs
        inputs = create_inputs_from_completion_params(completion_create_params)

        # Execute agent workflow

        # Return results
        return response_text, pipeline_interactions, usage_metrics 

class MyAgent:
    def __init__(self, api_key: Optional[str] = None, 
                 api_base: Optional[str] = None,
                 model: Optional[str] = None,
                 verbose: Optional[Union[bool, str]] = True,
                 timeout: Optional[int] = 90,
                 **kwargs: Any):
        """Initialize agent with DataRobot credentials and configuration.""" 

    def invoke(self, completion_create_params: CompletionCreateParams) -> Union[
        Generator[tuple[str, Any | None, dict[str, int]], None, None],
        tuple[str, Any | None, dict[str, int]],
    ]:
        """Main execution method - REQUIRED for DataRobot integration.

        Args:
            completion_create_params: Input parameters from DataRobot

        Returns:
            Union of generator (for streaming) or tuple (for non-streaming):
            - response_text: str - The agent's response
            - pipeline_interactions: Any | None - Event tracking data
            - usage_metrics: dict[str, int] - Token usage statistics
        """ 

@property
def llm(self) -> LLM:  # Framework-specific type
    """Primary LLM instance for agent operations.

    Returns:
        Framework-specific LLM type:
        - CrewAI: LLM
        - LlamaIndex: DataRobotLiteLLM
    """
    api_base = urlparse(self.api_base)
    if os.environ.get("LLM_DEPLOYMENT_ID"):
        # Handle deployment-specific URL construction
        # ... implementation details ...
        return LLM(model="openai/gpt-4o-mini", api_base=deployment_url, ...)
    else:
        # Handle LLM gateway URL construction
        # ... implementation details ...
        return LLM(model="datarobot/azure/gpt-5-mini-2025-08-07", api_base=api_base.geturl(), ...) 

@property
def llm(self) -> Any:
    """Primary LLM instance for agent operations.

    Returns:
        Any: Minimal implementation for custom frameworks
    """
    if os.environ.get("LLM_DEPLOYMENT_ID"):
        return self.llm_with_datarobot_deployment
    else:
        return self.llm_with_datarobot_llm_gateway 

llms:
    datarobot_llm:
    _type: datarobot-llm-gateway
    model_name: azure/gpt-4o-mini  # Define the model name you want to use
    temperature: 0.0 

functions:
    planner:
    _type: chat_completion
    llm_name: datarobot_llm  # Reference the LLM defined above
    system_prompt: |
        You are a content planner...