エージェントおよびツールの補助サンプル・コード

ツールのないエージェント

提供されているサンプル・コードを使用して、プロンプト、SQL、RAGなどのツールを含まないOracle AI Data Platform AIエージェントをテストできます。

# 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################

SQLツールのテスト

このサンプル・コードは、補助ファイルを使用して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}")

プロンプト(LLM)ツール・テスト

このサンプル・コードは、helpputilsを使用してプロンプト・ツールをテストする方法を示します。

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}")

カスタムコードツール- Hello World

このサンプル・コードは、ユーザのカスタム・コード・ツールをテストするための補助ツールの使用方法を示しています。

Hello Worldの例は、最も単純なカスタム・コード・ツールです。nameパラメータを受け入れて挨拶を返す単一のツール・クラスを定義します。独自のツールの開始点として使用します。

ツール_実装.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}!"}

ツール_構成.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": {}
     }
   ]
 }

要件.txt

# no deps

ZIPアーカイブのルートにある3つのファイルをパッケージ化し、「パッケージ」タブでZIPをアップロードします。アップロードしたら、「パラメータ」タブに切り替えて、デフォルトをオーバーライドする場合は「説明」を入力し、「テスト」タブに切り替えてツールを起動します。name="Alice"を指定すると、ツールは次を返します。

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

カスタム・コード・ツール- Developer Toolkit

このサンプル・コードは、ユーザのカスタム・コード・ツールをテストするための補助ツールの使用方法を示しています。

Developer Toolkitの例は、マルチツール・パッケージと、utils/ディレクトリでのヘルパー・モジュールの使用を示しています。このパッケージは、3つのツール(bashコマンド・ランナー、ファイル操作ツールおよびPythonコード・ランナー)を登録し、出力の切捨ておよびパス・サニタイズに共有ヘルパー関数を使用します。

ノート:

Developer Toolkitが例です。Bashコマンドの実行とPythonコードの実行には、重大なセキュリティ上の意味があります。本番環境では、AIコンピュートを制限し、操作をサンドボックス化し、ツールが実行するコマンドおよびコード・パターンに厳密な許可リストを適用します。

パッケージ・レイアウト

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

ツール_実装.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)}

ツール_構成.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

ユーティリティ/__init__.py

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

要件.txt

# stdlib only

ZIPをアップロードすると、「パッケージ」タブに検出された3つのツールが表示され、それぞれを有効または無効にできます。「パラメータ」タブには、BashTool、FileToolおよびPythonToolを切り替える「ツール・クラス」ドロップダウンが表示され、右側にツールごとの構成(timeout、max_output_lines、base_dir、max_file_size_kb)が公開されます。

Oracle AI Data Platform Workbenchでのツール登録によるエージェント

Oracle AI Data Platform Workbenchは、柔軟なエージェント構築と内部ツールのオーケストレーションをサポートします。このトピックでは、エージェント内でツールを定義、登録および使用するための推奨方法の例を示します。

1. 構成によるツールの説明

各ツールは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. レジストリ/構成へのツールの登録

すべてのユーザー・ツールは、エージェント検索用のレジストリに収集されます。

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

3. フレームワーク・ラッピング: エージェントが使用可能なツール・オブジェクトの作成

エージェントの構築では、次のディクトを実行可能なツール・オブジェクト(StructuredToolなど)に変換する必要があります。

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. メモリーとチェックポインタの使用

AI Data Platform Workbenchのエージェントは、多くの場合、中間状態を維持し、再開を可能にし、障害後または長時間実行されるワークフローにわたってリカバリできるようにするためのメモリーを必要とします。一般的なメカニズムは、エージェントの状態を保存およびリストアするcheckpointerオブジェクトです。

# 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

用途パターン:

コンストラクション時またはグローバル/コンテキスト変数として、'checkpointer'をエージェント・コード/クラスに渡します。
ツール出力、プロンプト・ステップ、LLM生成など、すべてのクリティカル・エージェント・イベントの後の状態を保存します。
エージェントの再起動時に状態をリストアします(使用可能な場合)。

チェックポインタの一般的なソース:

AI Data Platform Workbenchのデモ・コードでは、`checkpointer`はワークフロー構成またはグローバルを介して注入できます。たとえば、`checkpointer = globals().get("checkpointer"、 None)`
複雑なユースケースでは、チェックポインタが外部ストレージ、データベース、またはクラウドの状態をラップして、堅牢な障害回復を実現できます。

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

可観測性: ロギング、トレースおよびメトリック

可観測性は、aidp_observabilityパッケージを介してOracle AI Data Platform Workbenchアプリケーションにシームレスに統合され、最小限の設定で自動テレメトリ(ログ、トレース、メトリック)収集を実現します。

初期化

次に示すように、インポートと初期化を行います。

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

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

初期化時:

トレース、メトリックおよびログのOpenTelemetryエクスポータが作成されます。
コレクタ・エンドポイントは、すべてのテレメトリ・データ(ポート4317、GRpcプロトコル)に対して構成されます。
アプリケーション・ロガーが設定されます。
プレイグラウンド・モードでは、インスタント・トレース表示用のインメモリー・エクスポータが有効になります。
コレクタは、ログのローテーションおよびバッファリング用に事前構成されており、テレメトリ・エクスポート用のシンクが含まれています。
デフォルトのメトリック、ログおよびAI Data Platform Workbenchメタデータは、すべてのテレメトリ・シグナルに含まれています。
デフォルトのスパン/セッション属性(sessionId、traceIdなど)が相関用に設定されます。

用途パターン:

テレメトリを生成するためにアプリケーション・ロジックを変更する必要はありません。ユーザーとして:

メトリックにはOpenTelemetryメーターを使用します。
ログにはPythonの標準`logging`を使用します。
トレースにはOpenTelemetryトレーサを使用します。

例

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)

ノート:

アプリケーションの遠隔測定は自動的にエクスポートされます。計測の変更はユーザーが行う必要はありません。追跡レポート用の可観測性パッケージ自動計測LLMフレームワークおよびLangGraphアプリケーション。

Aidputilパッケージを使用したエージェントのインスタンス化と使用

次のサンプルは、helpputilパッケージでエージェントを作成および使用する方法を示しています。

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
    )

ガードレール構成

ガードレールは、OCIAIConf()を使用して基本モデルの選択の一部として、支援ユーティリティを使用して構成できます。

ガードレール構成は、OCI生成AIサービスから基本モデルを選択するときに提供されます。この例では、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)

ガードレール構成は、ポリシーの配列で構成されるJSONのような文字列です。前述の例では、<guardrailsName>および<guardrailsDescription>がユーザー定義の名前および説明である次のコード・ブロックで定義されています。


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

各ポリシーには次のキーがあります。

ヒント	必須	説明	データ・タイプ	デフォルト値
`policyName`	×	ポリシーのカスタム名	文字列	N/A
`policyType`	○	適用するガードレール・ポリシーのタイプ。指定できる値: `CONTENT_MODERATION` `PROMPT_ATTACKS_PREVENTION` `PII_DETECTION`	ENUM
`policyDescription`	×	ポリシーの説明です。	文字列
`scope`	×	スコープは、ガードレールの適用場所を定義します。指定できる値: `USER_REQUEST` `AGENT_RESPONSE` `BOTH`	ENUM
`action`	×	ポリシー違反時に実行するアクション指定できる値: `INFORM` `BLOCK` `ALLOW MASK` (`PII_DETECTION`の場合のみ)	ENUM
`threshold`	×	検出のしきい値。範囲は0から1までの確率です。	浮動小数
`piiCategories`	○	検出されるPIIデータのカテゴリとそのアクションおよび有効化。	配列

piiCategoriesは、次のキーを使用するJSONのようなオブジェクトの配列でもあります。

ヒント	必須	説明	データ・タイプ	デフォルト値
`category`	○	検出するPIIカテゴリ。指定できる値: `PERSON` `ADDRESS` `TELEPHONE_NUMBER` `EMAIL`	文字列	N/A
`isEnabled`	×	PIIカテゴリの検出を有効にします。指定できる値: `True` `False`	ENUM
`action`	×	PIIカテゴリが検出された場合に実行する処理。前述の処理を上書きします。指定できる値: `INFORM` `BLOCK` `ALLOW` `MASK`	文字列

例: ガードレール構成の完了

この場合、次の3つのポリシーすべてを適用します。

コンテンツのモデレーションは、エージェントの応答にのみ適用されます。
プロンプトインジェクションは、検出された場合、ユーザーの要求をブロックします。
PIIはエージェント・レスポンスとユーザー・リクエストの両方で検出されます。各PIIカテゴリは異なる方法で処理されます。

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" 
      } ] 
    } ] 
  }