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