에이전트 및 도구에 대한 보조 도구 샘플 코드

도구가 없는 에이전트

제공된 샘플 코드를 사용하여 프롬프트, 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 도구 테스트

이 예제 코드는 Aidputil을 사용하여 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) 도구 테스트

이 샘플 코드는 aidputils를 사용하여 프롬프트 도구를 테스트하는 방법을 보여줍니다.

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

이 샘플 코드는 Aidputils를 사용하여 사용자 정의 코드 도구를 테스트하는 방법을 보여줍니다.

Hello World 예제는 가능한 가장 간단한 사용자 정의 코드 도구입니다. 이름 매개변수를 받아들이고 인사말을 반환하는 단일 도구 클래스를 정의합니다. 자신의 도구의 시작점으로 사용합니다.

도구_구현.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}!"}

도구_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": {}
     }
   ]
 }

요구 사항.txt

# no deps

ZIP 아카이브의 루트에 세 개의 파일을 패키지화하고 패키지 탭을 통해 ZIP을 업로드합니다. 업로드한 후 매개변수 탭으로 전환하고, 기본값을 무효화하려면 설명을 채우고, 도구를 호출하려면 테스트 탭으로 전환합니다. name="Alice"를 사용할 경우 도구는 다음을 반환합니다.

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

Custom Code Tool - 개발자 툴킷

이 샘플 코드는 Aidputils를 사용하여 사용자 정의 코드 도구를 테스트하는 방법을 보여줍니다.

Developer Toolkit 예제는 utils/ 디렉토리에서 다중 도구 패키지 및 helper 모듈 사용을 보여줍니다. 이 패키지는 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)}

도구_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.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

uts/__init__.py

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

요구 사항.txt

# stdlib only

ZIP을 업로드한 후 Package(패키지) 탭에 검색된 세 가지 도구가 표시되고 각 도구를 사용 또는 사용 안함으로 설정할 수 있습니다. 매개변수 탭에는 BashTool, FileTool 및 PythonTool 간을 전환하고 오른쪽에 도구별 구성(timeout, max_output_lines, base_dir, max_file_size_kb)을 노출하는 Tool Class 드롭다운이 표시됩니다.

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 데이터 플랫폼 워크벤치의 에이전트는 종종 중간 상태를 유지하고, 재개를 가능하게 하며, 실패 후 또는 장기 실행 워크플로우에서 복구를 허용하기 위해 메모리가 필요합니다. 일반적인 방식은 에이전트 상태를 저장하고 복원하는 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

사용 패턴:

구성 시 에이전트 코드/클래스 또는 전역/컨텍스트 변수로 '체크포인터'를 전달합니다.
도구 출력, 프롬프트 단계 또는 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 데이터 플랫폼 워크벤치 메타데이터는 모든 원격 측정 신호에 포함됩니다.
기본 범위/세션 속성(예: 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 패키지 사용

다음 샘플은 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
    )

가드레일 구성

OCIAIConf()를 사용하여 기본 모델을 선택하는 과정에서 보조 장치를 사용하여 보호대를 구성할 수 있습니다.

OCI Generative AI 서비스에서 기본 모델을 선택할 때 Guardrails 구성이 제공됩니다. 이 예에서는 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`	아니요	정책의 사용자정의 이름	String	N/A
`policyType`	예	적용할 난간 정책의 유형입니다. 허용되는 값은 다음과 같습니다. `CONTENT_MODERATION` `PROMPT_ATTACKS_PREVENTION` `PII_DETECTION`	열거
`policyDescription`	아니요	정책에 대한 설명	String
`scope`	아니요	범위는 가드레일이 적용되는 위치를 정의합니다. 허용되는 값은 다음과 같습니다. `USER_REQUEST` `AGENT_RESPONSE` `BOTH`	열거
`action`	아니요	정책 위반 시 수행할 작업입니다. 허용되는 값은 다음과 같습니다. `INFORM` `BLOCK` `ALLOW MASK` (`PII_DETECTION`에만 해당)	열거
`threshold`	아니요	감지에 대한 임계값입니다. 범위는 0과 1 사이의 확률입니다.	float
`piiCategories`	예	해당 작업 및 사용으로 설정과 함께 감지될 PII 데이터의 범주입니다.	Array

piiCategories는 다음 키를 사용하는 JSON과 유사한 객체의 배열이기도 합니다.

핵심	필수사항	설명	데이터 유형	기본값
`category`	예	감지할 PII 범주입니다. 허용되는 값은 다음과 같습니다. `PERSON` `ADDRESS` `TELEPHONE_NUMBER` `EMAIL`	String	N/A
`isEnabled`	아니요	PII 범주의 감지를 사용으로 설정합니다. 허용되는 값은 다음과 같습니다. `True` `False`	열거
`action`	아니요	PII 범주가 감지된 경우 수행할 작업입니다. 위의 작업을 무효화합니다. 허용되는 값은 다음과 같습니다. `INFORM` `BLOCK` `ALLOW` `MASK`	String

예: Complete Guardrails 구성

이 경우 세 가지 정책을 모두 적용합니다.

콘텐츠 조정은 에이전트 응답에만 적용됩니다.
프롬프트 삽입은 감지된 경우 사용자 요청을 차단합니다.
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" 
      } ] 
    } ] 
  }