在 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。記憶體與使用 Checkpointer

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 資料平台工作台」示範程式碼中，可以透過工作流程組態或全域插入 `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()