カスタムツールLangGraphサンプルコード付きReActエージェント

このLangGraphコード・サンプルをエージェント・フローで使用して、カスタム・ツールの出力をテストおよびデバッグできます。

from aidputils.agents.toolkit.tool_helper import create_langgraph_tool
from aidputils.agents.toolkit.agent_helper import init_oci_llm, pre_invoke_setup
from aidputils.agents.toolkit.configs import AIDPToolConf, OCIAIConf, ModelArgs
from langgraph.prebuilt import create_react_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
import logging
import json

logger = logging.getLogger('agent_with_prompt_tool')
checkpointer = globals().get("checkpointer", None)

SYSTEM_PROMPT = (
    "You are a customer operations assistant.\n"
    "Use tools to gather accurate information.\n"
    "Think step by step.\n"
    "Respond clearly and concisely.\n"
)

########## Guardrails Configuration ################
guardrails_config = {
    "name" : "Default Guardrails",
    "description" : "Default empty guardrails configuration",
    "policies" : [ ]
  }
########## End Guardrails Configuration ############


@tool
def lookup_customer(customer_id: str) -> str:
    """Lookup customer profile (stub)"""
    return f"Customer {customer_id}: Enterprise, ARR $250k, healthy"


@tool
def fetch_usage(customer_id: str) -> str:
    """Fetch customer usage metrics (stub)"""
    return f"Customer {customer_id}: 42 jobs/day, 3.1TB processed"


@tool
def open_ticket(reason: str) -> str:
    """Create support ticket (stub)"""
    return f"Ticket created for issue: {reason}"


TOOLS = [
    lookup_customer,
    fetch_usage,
    open_ticket
]



model_args = {}
llm_conf = OCIAIConf(model_provider='generic',
                     compartment_id='<your-compartment-ocid>’,
                     model_args=model_args,
                     endpoint='https://inference.generativeai.<oci-region>.oci.oraclecloud.com',
                     model_id='xai.grok-4',
                     guardrails_config=guardrails_config)

## Agent class definition
class AgentWithTools:
  def __init__(self) -> None:
    self.agent = None

  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)

    try:
      if checkpointer:
        self.agent = create_react_agent(model=oci_llm, tools=TOOLS, prompt=SYSTEM_PROMPT, debug=True, checkpointer= checkpointer)
      else:
        self.agent  = create_react_agent(model=oci_llm, tools=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=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):
    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:
      import traceback
      logger.error(f"Exception while calling invoke {e}", exc_info=True)
      print("Stack trace:\n", traceback.format_exc())



# The following is used when executing the python from within the agent code editor
import asyncio
async def main():
    # Instantiate and initialize the agent
    test_agent = AgentWithTools()
    test_agent.setup()

    # You can customize this user query or prompt for input
    user_query = "Get customer 123 profile and usage"

    # Run the asynchronous invoke method and print the result
    result = await test_agent.invoke(user_query)
    print("Agent response:", result)

if __name__ == "__main__":
    asyncio.run(main())