Hilfsmittel für Agenten und Werkzeuge Beispielcode

Agent ohne Tools

Mit dem bereitgestellten Beispielcode können Sie einen KI-Agent von Oracle AI Data Platform testen, der keine Tools wie Prompt, SQL oder RAG enthält.

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

Dieser Beispielcode zeigt, wie Sie das SQL-Tool mit Hilfe von Hilfepaketen testen können.

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

Prompt-(LLM-)Tooltest

Dieser Beispielcode zeigt, wie Sie Hilfsmittel verwenden können, um Ihr Prompt-Tool zu testen.

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

Benutzerdefiniertes Code-Tool - Hello World

Dieser Beispielcode zeigt, wie Sie mit Hilfe von Hilfsmitteln Ihr Custom Code-Tool testen können.

Das Hello World-Beispiel ist das einfachste Custom Code-Tool. Sie definiert eine einzelne Toolklasse, die einen Namensparameter akzeptiert und eine Begrüßung zurückgibt. Verwenden Sie es als Ausgangspunkt für Ihr eigenes Tool.

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

Verpacken Sie die drei Dateien in der Root eines ZIP-Archivs, und laden Sie die ZIP-Datei über die Registerkarte Package hoch. Wechseln Sie nach dem Hochladen zur Registerkarte Parameter, geben Sie die Beschreibung ein, wenn Sie den Standardwert außer Kraft setzen möchten, und wechseln Sie zur Registerkarte Test, um das Tool aufzurufen. Bei name="Alice" gibt das Tool Folgendes zurück:

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

Tool für benutzerdefinierten Code - Entwickler-Toolkit

Dieser Beispielcode zeigt, wie Sie mit Hilfe von Hilfsmitteln Ihr Custom Code-Tool testen können.

Das Developer Toolkit-Beispiel zeigt ein Multi-Tool-Paket und die Verwendung von Helper-Modulen in einem utils/-Verzeichnis. Das Package registriert drei Tools – einen bash-Befehlsausläufer, ein Dateioperationstool und einen Python-Codeausläufer – und verwendet gemeinsam genutzte Helper-Funktionen für die Ausgabeabschneiden und Pfadbereinigung.

Hinweis:

Das Developer Toolkit ist ein Beispiel. Die Ausführung von Bash-Befehlen und die Ausführung von Python-Code haben erhebliche Auswirkungen auf die Sicherheit. Beschränken Sie in der Produktion die KI-Berechnung, sandboxen Sie die Vorgänge, und wenden Sie strenge Zulassungslisten für die Befehle und Codemuster an, die das Tool ausführen wird.

Paketlayout

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

Nach dem Hochladen der ZIP-Datei werden auf der Registerkarte Package die drei erkannten Tools angezeigt, mit denen Sie die einzelnen Tools aktivieren oder deaktivieren können. Auf der Registerkarte Parameter wird eine Dropdown-Liste Toolklasse angezeigt, die zwischen BashTool, FileTool und PythonTool wechselt und die Konfiguration pro Werkzeug (Timeout, max_output_lines, base_dir, max_file_size_kb) auf der rechten Seite anzeigt.

Agent mit Toolregistrierung in Oracle AI Data Platform Workbench

Oracle AI Data Platform Workbench unterstützt die flexible Agent-Erstellung und interne Toolorchestrierung. Dieses Thema enthält ein Beispiel für einen empfohlenen Ansatz zum Definieren, Registrieren und Verwenden von Tools in einem Agent.

1. Tools über die Konfiguration beschreiben

Jedes Tool ist ein Python-Dictionary:

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. Tools in einer Registry/Konfiguration registrieren

Alle Benutzertools werden in einer Registry für die Agent-Suche erfasst:

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

3. Framework-Wrapping: Agent-konsumierbare Toolobjekte erstellen

Die Agent-Konstruktion erfordert die Konvertierung dieser Dicts in ausführbare Toolobjekte (StructuredTool oder ähnliches):

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. Arbeitsspeicher und Checkpointer verwenden

Agents in AI Data Platform Workbench benötigen häufig Speicher, um den Zwischenstatus zu persistieren, die Wiederaufnahme zu ermöglichen und ein Recovery nach Ausfällen oder über Workflows mit langer Ausführungszeit zu ermöglichen. Der typische Mechanismus ist ein checkpointer-Objekt, das den Agent-Status speichert und wiederherstellt.

# 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

Verwendungsmuster:

Übergeben Sie den "checkpointer" an Agent-Code/Klasse bei der Konstruktion oder als globale/Kontextvariable.
Speichern Sie den Status nach jedem kritischen Agent-Ereignis, wie Toolausgabe, Prompt-Schritt oder LLM-Generierung.
Status bei Agent-Neustart wiederherstellen, falls verfügbar.

Typische Quellen des Checkpointers:

In AI Data Platform Workbench-Democode kann ein `checkpointer` über Workflowkonfiguration oder globale Elemente injiziert werden, z.B. `checkpointer = globals().get("checkpointer", None)`
Bei komplexen Anwendungsfällen kann der Checkpointer externen Speicher, Datenbanken oder Cloud-Status umschließen, um ein robustes Fehler-Recovery zu ermöglichen.

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

Beobachtbarkeit: Logging, Tracing und Metriken

Die Beobachtbarkeit ist über das helpp_observability-Package nahtlos in Oracle AI Data Platform Workbench-Anwendungen integriert und ermöglicht die automatische Erfassung von Telemetrie (Logs, Traces, Metriken) bei minimalem Setup.

Initialisierung

Import und Initialisierung wie dargestellt:

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

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

Bei der Initialisierung:

OpenTelemetry-Exporteure für Traces, Metriken und Logs werden erstellt.
Der Collector-Endpunkt ist für alle Telemetriedaten konfiguriert (Port 4317, GRpc-Protokoll).
Anwendungslogger sind eingerichtet.
Playground-Modus ermöglicht In-Memory-Exporter für sofortige Trace-Anzeige.
Der Collector ist für Logrotation, -pufferung vorkonfiguriert und enthält eine Senke für den Telemetrieexport.
Standardmetriken, -logs und AI Data Platform Workbench-Metadaten sind in allen Telemetriesignalen enthalten.
Standardspalten/Sessionattribute (z.B. sessionId, traceId) sind für die Korrelation festgelegt.

Verwendungsmuster:

In der Anwendungslogik sind keine Änderungen erforderlich, um Telemetrie auszugeben. Als Benutzer:

Verwenden Sie den OpenTelemetry Meter für Metriken.
Verwenden Sie das Standard-Logging von Python für Logs.
Verwenden Sie den OpenTelemetry Tracer für Traces.

Beispiel

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)

Hinweis:

Die Anwendungstelemetrie wird automatisch exportiert. Der Benutzer benötigt keine Instrumentierungsänderungen. Das Beobachtbarkeitspaket automatisiert LLM-Frameworks und LangGraph-Anwendungen für das Trace-Reporting.

Agent-Instanziierung und -Nutzung mit Aidputil-Paketen

In den folgenden Beispielen wird gezeigt, wie Sie Agents mit Hilfe-Paketen erstellen und verwenden können.

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
    )

Guardrails konfigurieren

Sie können Guardrails mit Hilfe von Hilfsmitteln konfigurieren, wenn Sie ein grundlegendes Modell mit OCIAIConf() auswählen.

Guardrails-Konfiguration wird bereitgestellt, wenn ein grundlegendes Modell aus dem OCI Generative AI-Service ausgewählt wird. In diesem Beispiel wählen wir das xai.grok-4 Modell:

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)

Die Guardrails-Konfiguration ist eine JSON-ähnliche Zeichenfolge, die aus einem Array von Policys besteht. Im obigen Beispiel wird es in diesem Codeblock definiert, wobei <guardrailsName> und <guardrailsDescription> ein benutzerdefinierter Name und eine benutzerdefinierte Beschreibung sind:


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

Jede Policy verfügt über die folgenden Schlüssel:

Legende	Erforderlich	Beschreibung	Datentyp	Standardwert
`policyName`	Nr.	Benutzerdefinierter Name für die Policy	String (Zeichenfolgendatentyp)	N/V
`policyType`	Ja	Art der anzuwendenden Leitplankenrichtlinie. Zulässige Werte sind: `CONTENT_MODERATION` `PROMPT_ATTACKS_PREVENTION` `PII_DETECTION`	ENUM
`policyDescription`	Nr.	Eine Beschreibung der Richtlinie	String (Zeichenfolgendatentyp)
`scope`	Nr.	Der Geltungsbereich definiert, wo die Leitschienen angewendet werden. Zulässige Werte sind: `USER_REQUEST` `AGENT_RESPONSE` `BOTH`	ENUM
`action`	Nr.	Die auszuführende Aktion, wenn gegen die Policy verstoßen wird Zulässige Werte sind: `INFORM` `BLOCK` `ALLOW MASK` (nur für `PII_DETECTION`)	ENUM
`threshold`	Nr.	Schwellenwert für die Erkennung. Der Bereich ist eine Wahrscheinlichkeit zwischen 0 und 1.	Fließkomma
`piiCategories`	Ja	Kategorie der zu ermittelnden personenbezogenen Daten sowie deren Aktion und Aktivierung.	Array

piiCategories ist auch ein Array von JSON-ähnlichen Objekten, das die folgenden Schlüssel verwendet:

Legende	Erforderlich	Beschreibung	Datentyp	Standardwert
`category`	Ja	Die zu ermittelnde personenbezogene Datenkategorie. Zulässige Werte sind: `PERSON` `ADDRESS` `TELEPHONE_NUMBER` `EMAIL`	String (Zeichenfolgendatentyp)	N/V
`isEnabled`	Nr.	Aktivieren Sie die Erkennung der Kategorie "PII". Zulässige Werte sind: `True` `False`	ENUM
`action`	Nr.	Aktion, die ausgeführt wird, wenn personenbezogene Daten ermittelt werden. Überschreiben Sie die oben genannte Aktion. Zulässige Werte sind: `INFORM` `BLOCK` `ALLOW` `MASK`	String (Zeichenfolgendatentyp)

Beispiel: Vollständige Guardrails-Konfiguration

In diesem Fall wenden wir alle drei Richtlinien an:

Die Inhaltsmoderation wird nur auf die Agent-Antwort angewendet.
Prompt-Injection blockiert Benutzeranfragen, wenn erkannt,
PII wird sowohl in der Agent-Antwort als auch in der Benutzeranforderung ermittelt. Jede PII-Kategorie wird unterschiedlich behandelt.

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