Aidputils for Agents and Tools 示例代码

无工具的代理

您可以使用提供的示例代码来测试不包含提示、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) 工具测试

此示例代码演示了如何使用辅助工具测试提示工具。

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 参数并返回一个 greeting。使用它作为您自己的工具的起点。

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

要求 .txt

# no deps

将这三个文件打包到 ZIP 归档文件的根目录，然后通过 Package 选项卡上载 ZIP。上载后，切换到参数选项卡，如果要覆盖默认值，请填写“说明”，然后切换到测试选项卡以调用该工具。使用 name="Alice" 时，该工具将返回：

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

自定义代码工具 - 开发者工具包

此示例代码演示了如何使用辅助工具测试定制代码工具。

Developer Toolkit 示例演示了多工具包以及在 utils/ 目录中使用帮助程序模块。软件包注册了三个工具 - 一个 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

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.

要求 .txt

# stdlib only

上载 ZIP 后， Package（程序包）选项卡将显示搜索到的三个工具，并允许您启用或禁用每个工具。参数选项卡显示 Tool Class（工具类）下拉列表，该下拉列表在 BashTool、FileTool 和 PythonTool 之间切换，并在右侧公开按工具的配置（超时、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()

可观测性：日志记录、跟踪和度量

Observability 通过 helpp_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 Meter 获取指标。
将 Python 的标准 `logging` 用于日志。
使用 OpenTelemetry Tracer 进行跟踪。

范例

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 Generative 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)

Guardrails 配置是一个类似 JSON 的字符串，由一系列策略组成。在上面的示例中，它在代码块中定义，其中 <guardrailsName> 和 <guardrailsDescription> 是用户定义的名称和说明：


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

每个策略都有以下键：

关键信息	必需	说明	数据类型	默认值
`policyName`	无	策略的定制名称	字符串	不适用
`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 数据及其操作和启用的类别。	Array

piiCategories 也是使用以下键的类似 JSON 的对象的数组：

关键信息	必需	说明	数据类型	默认值
`category`	是	要检测的 PII 类别。允许的值包括： `PERSON` `ADDRESS` `TELEPHONE_NUMBER` `EMAIL`	字符串	不适用
`isEnabled`	无	启用 PII 类别的检测。允许的值包括： `True` `False`	ENUM
`action`	无	检测到 PII 类别时要执行的操作。覆盖上面的操作。允许的值包括： `INFORM` `BLOCK` `ALLOW` `MASK`	字符串

示例：完整的 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" 
      } ] 
    } ] 
  }