Aidputils para Agentes y Herramientas Código de Ejemplo

Agente sin herramientas

Puede utilizar el código de ejemplo proporcionado para probar un agente de IA de Oracle AI Data Platform que no incluya herramientas, como prompt, SQL o RAG.

# Generated code for SIMPLE_AGENT operator muse_agent_node
from aidputils.agents.toolkit.tool_helper import create_langgraph_tool
from aidputils.agents.toolkit.agent_helper import init_oci_llm, pre_tool_setup, post_tool_setup, pre_invoke_setup
from aidputils.agents.toolkit.configs import AIDPToolConf, OCIAIConf, ModelArgs
from langgraph.prebuilt import create_react_agent
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
import logging

logger = logging.getLogger('SingleAgentNoTool')
class_name = 'SingleAgentNoTool'
checkpointer = globals().get("checkpointer", None)

########## Guardrails Configuration ################
guardrails_config = {
    "name" : "Default Guardrails",
    "description" : "Default empty guardrails configuration",
    "policies" : [ ]
  }
########## End Guardrails Configuration ############

########## Start Generated code for Agent Flow ################
########## Generated code for OCI Gen AI LLM
model_args = {
    "temperature" : 0.8,
    "max_tokens" : 500,
    "frequency_penalty" : 0,
    "presence_penalty" : 0,
    "top_p" : 1.0,
    "top_k" : 0
  }

llm_conf = OCIAIConf(model_provider='cohere',
                     compartment_id='<your-compartment-ocid>',
                     model_args=model_args,
                     endpoint='https://inference.generativeai.<oci-region>.oci.oraclecloud.com',
                     model_id='<your-model-id>')

## Agent class definition
class SingleAgentNoTool:
  def __init__(self) -> None:
    self.agent = None
  """
  Setup for LangGraph agent. This includes returns react_agent or compiled langgraph object.
  """
  def setup(self) -> None:
    logger.info(llm_conf)
    # TODO: Handle other kinds of llms, for example openAI or gemini
    oci_llm = init_oci_llm(llm_conf)
    system_prompt = """
        You are an AI Agent
        """
    try:
      if checkpointer:
        self.agent = create_react_agent(model=oci_llm, tools=[], prompt=system_prompt, debug=True, checkpointer= checkpointer)
      else:
        self.agent = create_react_agent(model=oci_llm, tools=[], prompt=system_prompt, debug=True)
    except Exception as e:
      # Fallback compile without checkpointer if wiring fails
      self.agent = create_react_agent(model=oci_llm, tools=[], prompt=system_prompt, debug=True)
      logger.warning(f"Checkpointer could not be initialized {e}")
    logger.info(f"Setup for agent completed {self.agent}")

  async def invoke(self, user_query: str, **kwargs):
    token = pre_tool_setup(**kwargs)
    config = pre_invoke_setup(**kwargs)
    user_message = HumanMessage(content=user_query)
    message = {"messages": [dict(user_message)]}
    try:
      return await self.agent.ainvoke(input=message, config = config)
    except Exception as e:
      logger.error(f"Exception while calling invoke {e}")
    finally:
      post_tool_setup(token)

##########End Generated code for Agent Flow################

Prueba de herramienta SQL

Este código de ejemplo muestra cómo puede utilizar las ayudas para probar la herramienta SQL.

from aidputils.agents.tools import utils
from aidputils.agents.auth.util import auth_utils

tool_conf = {'catalogKey': 'aidp_tools_dev',
             'schemaKey': 'aidpuser',
             'query': 'select * from employees where SALARY>={{SALARY_RANGE}}'}
runtime_params = {"SALARY_RANGE": 60000}
context_vars = {'datalake_id': 'YOUR_DATALAKE_ID'}

try:
    tool_result = utils.call_tool_by_class('SQLTool', tool_conf, runtime_params, **context_vars)
    print(tool_result)
except Exception as e:
    print(f"SQLTool execution failed: {e}")

Prueba de herramienta de petición de datos (LLM)

Este código de ejemplo muestra cómo puede utilizar las ayudas para probar la herramienta de petición de datos.

from aidputils.agents.tools import utils
from aidputils.agents.auth.util import auth_utils

tool_conf = {
    'prompt_template': 'What is the capital of {country}',
    'llm': {
        'model_id': 'cohere.command-r-08-2024',
        'model_provider': 'cohere',
        'model_args': {
            'temperature': 1,
            'max_tokens': 600,
            'frequency_penalty': 0,
            'presence_penalty': 0,
            'top_k': 0,
            'top_p': 0.75
        },
        'compartment_id': '<your-compartment-ocid>',
        'auth_type': 'REMOTE',
        'endpoint': 'https://inference.generativeai.<oci-region>.oci.oraclecloud.com',
        'auth_profile': 'DEFAULT'
    }
}
runtime_params = {'country': 'India'}
context_vars = {'datalake_id': 'YOUR_DATALAKE_ID'}

try:
    tool_result = utils.call_tool_by_class('PromptTool', tool_conf, runtime_params, **context_vars)
    print(tool_result)
except Exception as e:
    print(f"PromptTool execution failed: {e}")

Herramienta de código personalizado - Hello World

Este código de ejemplo muestra cómo puede utilizar las ayudas para probar la herramienta de código personalizado.

El ejemplo de Hello World es la herramienta de código personalizado más sencilla posible. Define una sola clase de herramienta que acepta un parámetro name y devuelve un saludo. Úsalo como punto de partida para tu propia herramienta.

tool_implementation.py

from aidputils.agents.tools.custom_tools.base import CustomToolBase
 

@BaseTool.register
 class HelloTool(CustomToolBase):
     """A simple greeting tool."""
 
    @classmethod
     def _execute_tool(cls, conf, runtime_params, **context_vars):
         name = runtime_params.get("name", "World")
         return {"greeting": f"Hello, {name}!"}

tool_config.json

{
   "displayName": "Hello Tool",
   "description": "A simple hello world tool",
   "tools": [
     {
       "toolClassName": "HelloTool",
       "displayName": "Hello Tool",
       "description": "Returns a hello world greeting",
       "version": "1.0.0",
       "schema": [
         {
           "name": "name",
           "type": "string",
           "description": "Name to greet"
         }
       ],
       "conf": {}
     }
   ]
 }

requirements.txt

# no deps

Empaquete los tres archivos en la raíz de un archivo ZIP y cargue el ZIP mediante el separador Paquete. Una vez cargado, cambie al separador Parámetros, rellene la descripción si desea sustituir el valor por defecto y cambie al separador Prueba para llamar a la herramienta. Con name="Alice", la herramienta devuelve:

{"greeting": "Hello, Alice!"}

Herramienta de código personalizado - Developer Toolkit

Este código de ejemplo muestra cómo puede utilizar las ayudas para probar la herramienta de código personalizado.

El ejemplo de Developer Toolkit muestra un paquete de varias herramientas y el uso de módulos auxiliares en un directorio utils/. El paquete registra tres herramientas, un corredor de comandos bash, una herramienta de operaciones de archivos y un corredor de código Python, y utiliza funciones de ayuda compartidas para el truncamiento de salida y la desinfección de rutas.

Note:

El kit de herramientas para desarrolladores es un ejemplo ilustrativo. La ejecución de comandos de Bash y la ejecución de código de Python tienen implicaciones de seguridad significativas. En producción, restrinja los recursos informáticos de IA, el entorno de prueba de las operaciones y aplique listas de permitidos estrictas para los comandos y los patrones de código que ejecutará la herramienta.

Diseño de paquete

advanced_tool.zip
 ├── tool_implementation.py
 ├── tool_config.json
 ├── requirements.txt          # stdlib only
 └── utils/
     ├── __init__.py
     └── text_utils.py         # truncate_output, sanitize_path

tool_implementation.py

import subprocess
 import os

from aidputils.agents.tools.custom_tools.base import CustomToolBase
 from .utils.text_utils import truncate_output, sanitize_path
 

def _get_cfg(conf, key, default):
     """Read a config value from either the outer dict or the
     nested user conf. Coerces numeric settings to int to avoid
     type mismatches when values are rendered as strings by the
     template substitution layer."""
     inner = conf.get("conf") if isinstance(conf, dict) else None
     if isinstance(inner, dict) and key in inner:
         value = inner[key]
     elif isinstance(conf, dict) and key in conf:
         value = conf[key]
     else:
         value = default
     if isinstance(default, int) and not isinstance(value, bool):
         try:
             return int(value)
         except (TypeError, ValueError):
             return default
     return value
 

@BaseTool.register
 class BashTool(CustomToolBase):
     """Execute bash commands and return output."""
 
    @classmethod
     def _execute_tool(cls, conf, runtime_params, **context_vars):
         command = runtime_params.get("command", "")
         timeout = _get_cfg(conf, "timeout", 30)
         max_lines = _get_cfg(conf, "max_output_lines", 200)
         try:
             result = subprocess.run(
                 ["bash", "-c", command],
                 capture_output=True, text=True, timeout=timeout
             )
         except subprocess.TimeoutExpired:
             # Surface the timeout as a tool failure rather than
             # returning {"error": ...}, which would be treated as
             # a successful response.
             raise RuntimeError(f"Command timed out after {timeout}s")
         output = result.stdout or ""
         if result.stderr:
             output += "\n[stderr]\n" + result.stderr
         return {"output": truncate_output(output, max_lines)}
 

@BaseTool.register
 class FileTool(CustomToolBase):
     """Read, write, or list files in the workspace."""
 
    @classmethod
     def _execute_tool(cls, conf, runtime_params, **context_vars):
         operation = runtime_params.get("operation", "")
         path = runtime_params.get("path", "")
         content = runtime_params.get("content", "")
         base_dir = _get_cfg(conf, "base_dir", "/workspace")
         max_size = _get_cfg(conf, "max_file_size_kb", 1024) * 1024
 
        safe_path = sanitize_path(base_dir, path)
         if safe_path is None:
             raise ValueError("Invalid path: path traversal detected")
 
        if operation == "read":
             with open(safe_path, "r") as f:
                 return {"output": f.read()}
         if operation == "write":
             parent = os.path.dirname(safe_path)
             if parent:
                 os.makedirs(parent, exist_ok=True)
             with open(safe_path, "w") as f:
                 f.write(content)
             return {"output": f"Written {len(content)} chars to {path}"}
         if operation == "list":
             target = safe_path if os.path.isdir(safe_path) else os.path.dirname(safe_path)
             return {"output": "\n".join(sorted(os.listdir(target)))}
         raise ValueError(f"Unknown operation: {operation}. Use read/write/list")
 

@BaseTool.register
 class PythonTool(CustomToolBase):
     """Execute Python code in an isolated subprocess."""
 
    @classmethod
     def _execute_tool(cls, conf, runtime_params, **context_vars):
         code = runtime_params.get("code", "")
         timeout = _get_cfg(conf, "timeout", 60)
         max_lines = _get_cfg(conf, "max_output_lines", 500)
         try:
             result = subprocess.run(
                 ["python3", "-c", code],
                 capture_output=True, text=True, timeout=timeout
             )
         except subprocess.TimeoutExpired:
             raise RuntimeError(f"Execution timed out after {timeout}s")
         output = result.stdout or ""
         if result.stderr:
             output += "\n[stderr]\n" + result.stderr
         return {"output": truncate_output(output, max_lines)}

tool_config.json

{
   "displayName": "Developer Toolkit",
   "description": "A collection of tools for bash commands, file operations, and Python execution",
   "tools": [
     {
       "toolClassName": "BashTool",
       "displayName": "Bash Tool",
       "description": "Executes a bash command and returns stdout/stderr output",
       "version": "1.0.0",
       "schema": [
         {
           "name": "command",
           "type": "string",
           "description": "The bash command to execute"
         }
       ],
       "conf": {
         "timeout": 30,
         "max_output_lines": 200
       }
     },
     {
       "toolClassName": "FileTool",
       "displayName": "File Tool",
       "description": "Read, write, or list files in the workspace",
       "version": "1.0.0",
       "schema": [
         {"name": "operation", "type": "string",
          "description": "Operation to perform: read, write, or list"},
         {"name": "path", "type": "string",
          "description": "File or directory path"},
         {"name": "content", "type": "string",
          "description": "Content to write (for write operation)"}
       ],
       "conf": {
         "base_dir": "/workspace",
         "max_file_size_kb": 1024
       }
     },
     {
       "toolClassName": "PythonTool",
       "displayName": "Python Tool",
       "description": "Executes Python code in an isolated subprocess and returns the output",
       "version": "1.0.0",
       "schema": [
         {"name": "code", "type": "string",
          "description": "The Python code to execute"}
       ],
       "conf": {
         "timeout": 60,
         "max_output_lines": 500
       }
     }
   ]
 }

utils/text_utils.py

def truncate_output(text, max_lines=200):
     if not text:
         return ""
     try:
         max_lines = int(max_lines)
     except (TypeError, ValueError):
         max_lines = 200
     lines = text.strip().split("\n")
     if len(lines) > max_lines:
         lines = lines[:max_lines] + [f"... ({len(lines) - max_lines} lines truncated)"]
     return "\n".join(lines)
 

def sanitize_path(base_dir, relative_path):
     import os
     if not relative_path:
         return base_dir
     full = os.path.normpath(os.path.join(base_dir, relative_path))
     if not full.startswith(os.path.normpath(base_dir)):
         return None
     return full

utils/__init__.py

# Empty file. Required for Python to treat utils/ as a package.

requirements.txt

# stdlib only

Después de cargar el ZIP, el separador Paquete muestra las tres herramientas detectadas y le permite activar o desactivar cada una. El separador Parámetros muestra una lista desplegable Clase de herramienta que cambia entre BashTool, FileTool y PythonTool, y muestra la configuración por herramienta (timeout, max_output_lines, base_dir, max_file_size_kb) a la derecha.

Agente con registro de herramientas en Oracle AI Data Platform Workbench

Oracle AI Data Platform Workbench admite la construcción flexible de agentes y la orquestación de herramientas internas. En este tema se proporciona un enfoque recomendado de ejemplo para definir, registrar y utilizar herramientas dentro de un agente.

1. Describir las herramientas mediante la configuración

Cada herramienta es un diccionario de Python:

my_tool = {
    "name": "blog_idea_tool",
    "description": "Generate blog ideas for a topic.",
    "class": "PromptTool",
    "conf": {...},  # tool-specific settings
    "params": [
        {"name": "topic", "type": "string", "description": "Blog topic"}
    ]
}

2. Registrar herramientas en un registro/config

Todas las herramientas de usuario se recopilan en un registro para la consulta de agentes:

tool_conf = {
    "blog_idea_tool": my_tool,
    "social_post_tool": another_tool,
    # ... more tools
}

3. Ajuste del marco: creación de objetos de herramientas consumibles por agentes

La construcción del agente requiere convertir estos dicts en objetos de herramientas ejecutables (StructuredTool o similares):

from langchain_core.tools import StructuredTool

def create_langgraph_tool(tool):
    def tool_fn(**kwargs):
        # Example implementation: you would use utils.call_tool_by_name/tool runner, etc.
        return f"Executed {tool['name']} with inputs: {kwargs}"
    return StructuredTool.from_function(
        func=tool_fn,
        name=tool['name'],
        description=tool['description'],
        args_schema=None,  # Build a pydantic schema if detailed validation required
        infer_schema=False
    )

4. Memoria y uso de un indicador de control

Los agentes de AI Data Platform Workbench suelen necesitar memoria para mantener el estado intermedio, permitir la reanudación y permitir la recuperación después de fallos o en flujos de trabajo de larga ejecución. El mecanismo típico es un objeto checkpointer, que guarda y restaura el estado del agente.

# Suppose you have a 'checkpointer' object available:
# It might be provided to your agent context directly, or created via aidp-agent-runtime utilities

# During agent run:
state = {"step": "tool_invoked", "result": tool_result}

if checkpointer:
    checkpointer.save(state)
    # To restore later:
    loaded_state = checkpointer.load()
    print(f"Restored state: {loaded_state}")

# You can persist any serializable agent context, params, or partial results

Patrón de uso:

Transfiera el 'checkpointer' al código/clase de agente en la construcción o como una variable global/contexto.
Guarde el estado después de cada evento de agente crítico, como la salida de la herramienta, el paso de petición de datos o la generación del LLM.
Restaurar estado al reiniciar el agente, si está disponible.

Fuentes típicas del indicador de comprobación:

En el código de demostración de AI Data Platform Workbench, se puede inyectar un `checkpointer` mediante la configuración del flujo de trabajo o los valores globales, por ejemplo, `checkpointer = globals().get("checkpointer", None)`
En los casos de uso complejos, el indicador de control puede encapsular el almacenamiento externo, las bases de datos o el estado de la nube para permitir una recuperación de fallos sólida.

# Inside agent code
checkpointer = globals().get("checkpointer", None)
if checkpointer:
    checkpointer.save({"step": "after_tool", "context": context_vars})
    # ...
    restored_state = checkpointer.load()

Observability: registro, rastreo y métricas

La observabilidad se integra perfectamente en las aplicaciones de Oracle AI Data Platform Workbench a través del paquete helpp_observability, lo que permite la recopilación automática de telemetría (logs, rastreos, métricas) con una configuración mínima.

Inicialización

Importe e inicialice como se muestra:

from observability.aidp_observability import AIDPObservability
from observability.config import CollectorConfig

config = CollectorConfig()
config.service_name = "dummy_name"
observability = AIDPObservability(config)
observability.initialize()

Tras la inicialización:

Se crean los exportadores de OpenTelemetry para rastreos, métricas y logs.
El punto final del recopilador está configurado para todos los datos de telemetría (puerto 4317, protocolo GRpc).
Se configuran los registradores de aplicaciones.
El modo de patio de recreo permite al exportador en memoria mostrar el rastreo instantáneo.
El recopilador está preconfigurado para la rotación de logs, el almacenamiento en buffer e incluye un fregadero para la exportación de telemetría.
Las métricas, los logs y los metadatos del área de trabajo de la plataforma de datos de IA por defecto se incluyen en todas las señales de telemetría.
Los atributos de período/sesión por defecto (por ejemplo, sessionId, traceId) se definen para la correlación.

Patrón de uso:

No se necesitan cambios en la lógica de la aplicación para emitir telemetría. Como usuario:

Utilice el medidor de OpenTelemetry para las métricas.
Utilice el `logging` estándar de Python para los logs.
Utilice el rastreador OpenTelemetry para los rastreos.

Ejemplo

import logging
import time
from opentelemetry import trace, metrics

tracer = trace.get_tracer(__name__)
meter = metrics.get_meter(__name__)

request_counter = meter.create_counter(
    name="requests_total",
    description="Number of requests processed",
    unit="1",
)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("sample-app")

def process_request(user_id: str):
    logger.info("Processing request for user %s", user_id)
    request_counter.add(1, {"user.id": user_id})
    with tracer.start_as_current_span("process_request") as span:
        span.set_attribute("user.id", user_id)
        time.sleep(0.1)
        span.add_event("request_completed", {"status": "ok"})

if __name__ == "__main__":
    for i in range(3):
        process_request(f"user-{i}")
        time.sleep(1)

Note:

La telemetría de aplicación se exporta automáticamente; el usuario no requiere ningún cambio de instrumentación. El paquete de observabilidad autoinstrumenta los marcos LLM y las aplicaciones LangGraph para la generación de informes de rastreo.

Instanciación y uso de agente con paquetes Aidputil

Los siguientes ejemplos muestran cómo puede crear y utilizar agentes con paquetes de Aidputil.

from aidputils.agents.toolkit.agent_helper import invoke, get_client
from aidputils.agents.toolkit.configs import OCIAIConf
from langchain_core.tools import StructuredTool
from langgraph.prebuilt import create_react_agent
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
from langchain_community.chat_models.oci_generative_ai import ChatOCIGenAI
import logging
import json
 
logger = logging.getLogger('muse_agent_flow')
checkpointer = globals().get("checkpointer", None)
 
########## Guardrails Configuration ################
guardrails_config = {
    "name" : "Default Guardrails",
    "description" : "Default empty guardrails configuration",
    "policies" : [ ]
  }
########## End Guardrails Configuration ############
 
########## Start Generated code for Agent Flow ################
##### Start Tool configuration for blog_idea_tool
##### Start PROMPT Tool configuration
blog_idea_tool_def = {
  "llm": {
    "model_id" : "<your-model-id>",
    "model_provider" : "cohere",
    "compartment_id" : "<your-compartment-ocid>",
    "endpoint" : "https://inference.generativeai.<oci-region>.oci.oraclecloud.com",
    "auth_type" : "SECURITY_TOKEN",
    "auth_profile" : "DEFAULT",
    "model_args" : {
      "temperature" : 1,
      "max_tokens" : 600,
      "frequency_penalty" : 0,
      "presence_penalty" : 0,
      "top_k" : 0,
      "top_p" : 0.75
    }
  }, "prompt_template": """
You are a master blog strategist.
Your task is to brainstorm compelling blog post ideas based on a given topic.
For the given {topic}, generate 5 unique blog post titles.
For each title, include a one-sentence description of the angle the post would take.
Present the output as a numbered list.
"""
}
 
blog_idea_tool_params = [ {
  "name" : "topic",
  "type" : "string",
  "description" : "The central theme or subject for which to generate blog ideas."
} ]
 
blog_idea_tool_dict = {
    "name": "blog_idea_tool",
    "description": "Use this tool to generate several distinct and engaging blog post titles and concepts based on a topic ",
    "tool_class": "PromptTool",
    "conf": blog_idea_tool_def,
    "params": blog_idea_tool_params
}
blog_idea_tool = create_langgraph_tool(blog_idea_tool_dict)
##### End PROMPT Tool configuration
# Set tool_var_name = blog_idea_tool
# set ns.tool_var_list = [blog_idea_tool]
##### End Tool configuration for Blog idea tool
##### End Tool configuration
 
##### Start tool List#############
tools_agent1 = [blog_idea_tool]
##### End tool List#############
 
 
########## Generated code for OCI Gen AI LLM
model_args = {
  "temperature" : 0.8,
  "max_tokens" : 500,
  "frequency_penalty" : 0,
  "presence_penalty" : 0,
  "top_p" : 1.0,
  "top_k" : 0
}

llm_conf = OCIAIConf(model_provider='cohere',
                     compartment_id='<your-compartment-ocid>',
                     auth_type='SECURITY_TOKEN',
                     auth_profile='DEFAULT',
                     model_args=model_args,
                     endpoint='https://inference.generativeai.<oci-region>.oci.oraclecloud.com',
                     model_id='<your-model-id>')
 
 
## Agent class definition
class MuseAgentFlow:
  def __init__(self) -> None:
    self.agent = None
 
  def setup(self) -> None:
    # TODO: Handle other kinds of llms, for example openAI or gemini
    oci_llm = init_oci_llm(llm_conf)
    system_prompt = """
**Task:**
For the given {topic}, generate 5 unique blog post titles. For each title, include a one-sentence description of the angle the post would take. Present the output as a numbered list.
 
**Example Input:**
topic: "AI in marketing"
 
**Example Output:**
1.  **Title:** "Beyond the Hype: 3 Practical Ways to Use AI in Your Marketing Today"
    * **Angle:** This post will focus on simple, actionable AI tools that small businesses can implement immediately.
2.  **Title:** "Is AI Coming for Your Marketing Job? A Realistic Look at the Future"
*   * **Angle:** This post will explore how AI will change marketing roles, not just replace them, focusing on new skills.
3.  **Title:** "We Let an AI Write Our Marketing Emails for a Week. Here's What Happened."
*   * **Angle:** A case-study style post detailing the results of an interesting experiment.
4.  **Title:** "The Ethics of AI Marketing: Are You Crossing a Line with Personalization?"
    * **Angle:** A thought-leadership piece that discusses the important ethical considerations of using AI.
5.  **Title:** "How to Personalize at Scale: A Guide to AI-Powered Customer Journeys"
    * **Angle:** A tactical guide on using AI to create highly personalized marketing campaigns.
"""
 
    try:
      if checkpointer:
        self.agent =create_react_agent(model=oci_llm, tools=tools_agent1, prompt=system_prompt, debug=True, checkpointer= checkpointer)
      else:
        self.agent  = self.agent = create_react_agent(model=oci_llm, tools=tools_agent1, prompt=system_prompt, debug=True)
    except Exception as e:
      # Fallback compile without checkpointer if wiring fails
      self.agent = create_react_agent(model=oci_llm, tools=tools_agent1, prompt=system_prompt, debug=True)
      logger.warning(f"Checkpointer could not be initialized {e}")
    logger.info(f"Setup for agent completed {self.agent}")
 
  async def invoke(self, user_query: str, **kwargs):
    try:
      return await self.agent.invoke(input=user_query, **kwargs)
    except Exception as e:
      logger.error(f"Exception while calling invoke {e}")
 
  def init_oci_llm(llm_conf: OCIAIConf):
 
    chat = ChatOCIGenAI(
        model_id='<your-model-id>',
        provider='cohere',
        service_endpoint='https://inference.generativeai.<oci-region>.oci.oraclecloud.com',
        compartment_id='<your-compartment-ocid>',
        client=get_client(llm_conf=llm_conf),
        model_kwargs=model_args
    )
 
    return chat
  
  def create_langgraph_tool(tool):
    def tool_fn(**kwargs):
        # Example implementation: you would use utils.call_tool_by_name/tool runner, etc.
        return f"Executed {tool['name']} with inputs: {kwargs}"
    return StructuredTool.from_function(
        func=tool_fn,
        name=tool['name'],
        description=tool['description'],
        args_schema=None,  # Build a pydantic schema if detailed validation required
        infer_schema=False
    )

Configuración de guías

Puede configurar barandillas mediante Aidputils como parte de la selección de un modelo básico mediante OCIAIConf().

La configuración de las guías se proporciona al seleccionar un modelo básico del servicio OCI Generative AI. En este ejemplo, seleccionamos el modelo xai.grok-4:

from aidputils.agents.toolkit.configs import OCIAIConf 
guardrails_config = { 
    "name" : "<guardrailsName>", 
    "description" : "<guardrailsDescription>", 
    "policies" : [ ] 
  } 
model_args = {} 
llm_conf = OCIAIConf(model_provider='generic', 
                     compartment_id='<compartment_ocid>', 
                     model_args=model_args, 
                     endpoint='https://inference.generativeai.<oci-region>.oci.oraclecloud.com', 
                     model_id='xai.grok-4', 
                     guardrails_config=guardrails_config)

La configuración de las guías de protección es una cadena similar a JSON que consta de una matriz de políticas. En el ejemplo anterior, se define en este bloque de código donde <guardrailsName> y <guardrailsDescription> son un nombre y una descripción definidos por el usuario:


guardrails_config = { 
    "name" : "<guardrailsName>", 
    "description" : "<guardrailsDescription>", 
    "policies" : [ ] 
  }

Cada política tiene las siguientes claves:

Tecla	Obligatorio	Descripción	Tipo de dato	Valor por defecto
`policyName`	N.º	Nombre personalizado para la política	Cadena	No disponible
`policyType`	Sí	Tipo de política de barandilla que aplicar. Los valores permitidos son: `CONTENT_MODERATION` `PROMPT_ATTACKS_PREVENTION` `PII_DETECTION`	ENUM
`policyDescription`	N.º	Descripción de la política	Cadena
`scope`	N.º	El ámbito define dónde se aplican las barandillas. Los valores permitidos son: `USER_REQUEST` `AGENT_RESPONSE` `BOTH`	ENUM
`action`	N.º	Acción que se debe realizar cuando se viola la política Los valores permitidos son: `INFORM` `BLOCK` `ALLOW MASK` (solo para `PII_DETECTION`)	ENUM
`threshold`	N.º	Umbral de detección. El rango es una probabilidad entre 0 y 1.	float
`piiCategories`	Sí	Categoría de datos de PII que se detectarán junto con su acción y habilitación.	Matriz

piiCategories también es una matriz de objetos similares a JSON que utiliza las siguientes claves:

Tecla	Obligatorio	Descripción	Tipo de dato	Valor por defecto
`category`	Sí	La categoría de PII que se debe detectar. Los valores permitidos son: `PERSON` `ADDRESS` `TELEPHONE_NUMBER` `EMAIL`	Cadena	No disponible
`isEnabled`	N.º	Permite activar la detección de la categoría de información de identificación personal. Los valores permitidos son: `True` `False`	ENUM
`action`	N.º	Acción que se debe realizar si se detecta una categoría de información de identificación personal. Sustituya la acción anterior. Los valores permitidos son: `INFORM` `BLOCK` `ALLOW` `MASK`	Cadena

Ejemplo: configuración completa de guías de protección

En este caso, aplicamos las tres políticas:

la moderación del contenido solo se aplica a la respuesta del agente,
la inyección de aviso bloqueará las solicitudes del usuario si se detectan,
La información de identificación personal se detecta tanto en la respuesta del agente como en la solicitud del usuario. Cada categoría de PII se trata de manera diferente.

guardrails_config = { 
    "policies" : [ { 
      "policyType" : "CONTENT_MODERATION", 
      "policyName" : "Content Moderation prevention", 
      "policyDescription" : "Choose an action to take when hate, sexual, violence, toxic, derogatory, or harassment content is detected in either the user input query or the agent response.", 
      "scope" : "AGENT_RESPONSE", 
      "action" : "INFORM", 
      "threshold" : 0.5, 
      "categories" : [ ] 
    }, { 
      "policyType" : "PROMPT_ATTACKS_PREVENTION", 
      "policyName" : "Prompt Injection prevention", 
      "policyDescription" : "Choose action when prompt injection is detected on the user query.", 
      "scope" : "USER_REQUEST", 
      "action" : "BLOCK", 
      "threshold" : 0.5 
    }, { 
      "policyType" : "PII_DETECTION", 
      "policyName" : "Personally Identifiable Information (PII) detection", 
      "policyDescription" : "Choose an action to take when PII entities are detected in either the user input query or the agent response.", 
      "scope" : "AGENT_RESPONSE", 
      "action" : "INFORM", 
      "threshold" : 0.5, 
      "piiCategories" : [ { 
        "category" : "PERSON", 
        "isEnabled" : False, 
        "action" : "INFORM" 
      }, { 
        "category" : "ADDRESS", 
        "isEnabled" : False, 
        "action" : "INFORM" 
      }, { 
        "category" : "TELEPHONE_NUMBER", 
        "isEnabled" : True, 
        "action" : "MASK" 
      }, { 
        "category" : "EMAIL", 
        "isEnabled" : True, 
        "action" : "MASK" 
      } ] 
    }, { 
      "policyType" : "PII_DETECTION", 
      "policyName" : "Personally Identifiable Information (PII) detection", 
      "policyDescription" : "Choose an action to take when PII entities are detected in either the user input query or the agent response.", 
      "scope" : "USER_REQUEST", 
      "action" : "INFORM", 
      "threshold" : 0.5, 
      "piiCategories" : [ { 
        "category" : "PERSON", 
        "isEnabled" : True, 
        "action" : "INFORM" 
      }, { 
        "category" : "ADDRESS", 
        "isEnabled" : True, 
        "action" : "INFORM" 
      }, { 
        "category" : "TELEPHONE_NUMBER", 
        "isEnabled" : True, 
        "action" : "BLOCK" 
      }, { 
        "category" : "EMAIL", 
        "isEnabled" : False, 
        "action" : "INFORM" 
      } ] 
    } ] 
  }