agent.py

# Tencent is pleased to support the open source community by making tRPC-Agent-Python available.
#
# Copyright (C) 2026 Tencent. All rights reserved.
#
# tRPC-Agent-Python is licensed under Apache-2.0.
""" Agent module"""

from trpc_agent_sdk.agents import LlmAgent
from trpc_agent_sdk.code_executors import BaseCodeExecutor
from trpc_agent_sdk.code_executors import ContainerCodeExecutor
from trpc_agent_sdk.code_executors import UnsafeLocalCodeExecutor
from trpc_agent_sdk.log import logger
from trpc_agent_sdk.models import LLMModel
from trpc_agent_sdk.models import OpenAIModel

from .config import get_model_config
from .prompts import INSTRUCTION


def _create_model() -> LLMModel:
    """ Create a model"""
    api_key, url, model_name = get_model_config()
    model = OpenAIModel(model_name=model_name, api_key=api_key, base_url=url)
    return model


def _create_code_executor(code_executor_type: str = "unsafe_local") -> BaseCodeExecutor:
    """Create a code executor.

    Args:
        code_executor_type: Type of code executor to use. Options:
            - "unsafe_local": Use UnsafeLocalCodeExecutor (default, no Docker required)
            - "container": Use ContainerCodeExecutor (requires Docker)
            - None: Auto-detect from environment variable CODE_EXECUTOR_TYPE,
                    or default to "unsafe_local"

    Returns:
        BaseCodeExecutor instance.

    Raises:
        RuntimeError: If container type is requested but Docker is not available.
            The error message will include detailed instructions on how to fix the issue.
    """
    # Get executor type from environment variable if not specified
    if code_executor_type == "unsafe_local":
        return UnsafeLocalCodeExecutor(timeout=10)
    elif code_executor_type == "container":
        # ContainerCodeExecutor will raise a clear error if Docker is not available
        # The error message includes detailed instructions on how to fix the issue
        executor = ContainerCodeExecutor(image="python:3-slim", error_retry_attempts=1)
        logger.info("ContainerCodeExecutor initialized successfully")
        return executor
    else:
        raise ValueError(f"Invalid code executor type: {code_executor_type}. "
                         "Valid options are: 'unsafe_local', 'container'")


def create_agent() -> LlmAgent:
    """Create an agent with code execution capabilities.

    The agent can:
    - Execute Python code blocks generated by the LLM
    - Use tools like get_weather_report
    - Perform calculations and data processing through code execution

    Note: UnsafeLocalCodeExecutor executes code in the current process context.
    For production use, consider using ContainerCodeExecutor for better security.
    """
    executor = _create_code_executor()
    agent = LlmAgent(
        name="code_assistant",
        description="代码执行助手",
        model=_create_model(),  # You can change this to your preferred model
        instruction=INSTRUCTION,
        code_executor=executor,  # Enables code execution functionality
    )
    return agent


root_agent = create_agent()