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