エージェントLangGraphコード・サンプル

提供されているLangGraphコード・サンプルを参照として使用したり、Oracle AI Data Platform Workbenchでエージェントの使用を開始したりできます。

トピック:

• LangGraphのHello World
• プロンプトツールLangGraphサンプルコード付きReActエージェント
• RAGツールLangGraphサンプルコード付きReActエージェント
• SQL Tool LangGraphを使用したReActエージェントのサンプル・コード
• カスタムツールLangGraphサンプルコード付きReActエージェント
• マルチエージェント・システム – 監督者パターン・サンプル・コード
• リモートOracle Analytics Cloud MCPサーバー接続のサンプル・コード

LangGraphのHello World

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

from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
from langgraph.graph import StateGraph, MessagesState, START, END

def mock_llm(state: MessagesState):
    return {"messages": [{"role": "ai", "content": "hello world"}]}

class AgentBasic:
  def __init__(self) -> None:
    self.graph = None

  def setup(self) -> None:
    self.graph = StateGraph(MessagesState)
    self.graph.add_node(mock_llm)
    self.graph.add_edge(START, "mock_llm")
    self.graph.add_edge("mock_llm", END)
    self.graph =     self.graph.compile()
    system_prompt = "Be a helpful assistant."

  async def invoke(self, user_query: str, **kwargs):
    user_message = HumanMessage(content=user_query)
    messages = {"messages": [dict(user_message)]}
    try:
      return self.graph.invoke(messages)
    except Exception as e:
      import traceback
      logger.error(f"Exception while calling invoke {e}", exc_info=True)
      print("Stack trace:\n", traceback.format_exc())

import asyncio
async def main():
    test_agent = AgentBasic()
    test_agent.setup()
    result = await test_agent.invoke("Hi there")
    print("Agent response:", result)
if __name__ == "__main__":
    asyncio.run(main())

プロンプトツールLangGraphサンプルコード付きReActエージェント

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

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.messages import AIMessage, HumanMessage, SystemMessage
import logging
import json

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

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

##### Start PROMPT Tool configuration
blogger_def = {
  "llm": {
    "model_id" : "xai.grok-4",
    "model_provider" : "generic",
    "compartment_id" : "<your-compartment-ocid>",
    "endpoint" : "https://inference.generativeai.<oci-region>.oci.oraclecloud.com"
  }, "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"
"""
}

blogger_params = [ {
  "name" : "topic",
  "type" : "string",
  "description" : "Blog topic",
  "defaultValue" : "golf"
} ]

blogger_conf= AIDPToolConf(name="blogger",
                                  description= "PROMPT_description_794368 ",
                                  tool_class = "PromptTool", conf=blogger_def, params=blogger_params)
blogger = create_langgraph_tool(blogger_conf.model_dump())
##### End PROMPT Tool configuration

##### Start tool List#############
tools_agent1 = [blogger]
##### End tool List#############


model_args = {}
llm_conf = OCIAIConf(model_provider='generic',
                     compartment_id='<your-oci-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 AgentWithPromptTool:
  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)
    system_prompt = """
Be a helpful assistant.
"""

    try:
      if checkpointer:
        self.agent = create_react_agent(model=oci_llm, tools=tools_agent1, prompt=system_prompt, debug=True, checkpointer= checkpointer)
      else:
        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):
    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 = AgentWithPromptTool()
    test_agent.setup()

    # You can customize this user query or prompt for input
    user_query = "Give me 3 ideas for a robotics blog"

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

RAGツールLangGraphサンプルコード付きReActエージェント

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

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.messages import AIMessage, HumanMessage, SystemMessage
import logging
import json

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

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

##### Start RAG Tool configuration

rag_params = [ {
  "name" : "query",
  "type" : "string",
  "description" : "RAG query",
  "defaultValue" : "find matching doc artifacts for …"
} ]

conf = {
     "catalog": "default",
     "schema": "default",
     "knowledgeBase": "acme_kb",
     "top_k": 5,
     "llm": {
       "model_id" : "xai.grok-4",
       "model_provider" : "generic",
       "compartment_id" : "<your-compartment-ocid>",
       "endpoint" : "https://inference.generativeai.<oci-region>.oci.oraclecloud.com"
     }
   }
rag_conf= AIDPToolConf(name="rag",
                                  description= "RAG Tool ",
                                  tool_class = "RAGTool", conf=conf, params=rag_params)
rag_tool = create_langgraph_tool(rag_conf.model_dump())


##### End RAG Tool configuration

##### Start tool List#############
tools_agent1 = [rag_tool]
##### End tool List#############


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 AgentWithRAGTool:
  def __init__(self) -> None:
    self.agent = None

  def setup(self) -> None:
    logger.info(llm_conf)
    oci_llm = init_oci_llm(llm_conf)
    system_prompt = """
Be a helpful assistant.
"""

    try:
      if checkpointer:
        self.agent = create_react_agent(model=oci_llm, tools=tools_agent1, prompt=system_prompt, debug=True, checkpointer= checkpointer)
      else:
        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):
    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())




import asyncio
async def main():
    # Instantiate and initialize the agent
    test_agent = AgentWithRAGTool()
    test_agent.setup()

    # You can customize this user query or prompt for input
    user_query = "Summarize SOX Compliance and Reconciliation at Acme Corp"

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

SQL Tool LangGraphを使用したReActエージェントのサンプル・コード

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

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.messages import AIMessage, HumanMessage, SystemMessage
import logging
import json
import os

compartment_id = os.getenv('AIDP_USER_COMPARTMENT_ID')
compartment_id="<your-compartment-ocid>"
logger = logging.getLogger('agent_with_sql_tool')
checkpointer = globals().get("checkpointer", None)

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

#####
##### Start SQL Tool configuration

sql_params = [ {
  "name" : "MAX_SALARY",
  "type" : "string",
  "description" : "Maximum salary",
  "defaultValue" : "50000"
} ]


conf = {
    "catalogKey": "adw23ai_phx",
    "schemaKey": "gold",
    "query": """ Select * from (
  Select 101 employee_id, 'John' first_name, 'Doe' last_name, '[e-mail]' email_address, 75000 salary from DUAL
  UNION
  Select 102 employee_id, 'Jane' first_name, 'Smith' last_name, '[e-mail]' email_address, 100000 salary from DUAL
  UNION
  Select 103 employee_id, 'Peter' first_name, 'Jones' last_name, '[e-mail]' email_address, 45000 salary from DUAL
) employees where salary >= {{MAX_SALARY}}
"""
}

sql_conf= AIDPToolConf(name="query_employees",
                                  description= "Query employees using SQL Tool ",
                                  tool_class = "SQLTool", conf=conf, params=sql_params)
sql_tool = create_langgraph_tool(sql_conf.model_dump())


##### End SQL Tool configuration

##### Start tool List#############
tools_agent1 = [sql_tool]
##### End tool List#############


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 AgentWithSQLTool:
  def __init__(self) -> None:
    self.agent = None

  def setup(self) -> None:
    logger.info(llm_conf)
    oci_llm = init_oci_llm(llm_conf)
    system_prompt = """
Be a helpful assistant.
"""

    try:
      if checkpointer:
        self.agent = create_react_agent(model=oci_llm, tools=tools_agent1, prompt=system_prompt, debug=True, checkpointer= checkpointer)
      else:
        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):
    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())




import asyncio
async def main():
    # Instantiate and initialize the agent
    test_agent = AgentWithSQLTool()
    test_agent.setup()

    # You can customize this user query or prompt for input
    user_query = "Which employees have a salary greater than 50000"

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

カスタムツール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())

マルチエージェント・システム – 監督者パターン・サンプル・コード

この例は、スーパーバイザ・エージェント・パターンに従い、SQLiteを使用するデータ・エージェントです。SQL関数は、SQLツールに置き換えて、Oracleデータベースを利用できます。たとえば、初期状態で動作するため、デモンストレーションにSQLiteを使用します。

次のことを試してください。

• EMEAの収益合計はいくらでしたか?
• EMEAの収益は、2024年のAPACをどのように比較しましたか?

このエージェント・フローの設計は次のようになります。

• ルーター
  • →SQLエージェント
    • → 比較エージェント
      • → インサイトエージェント
      • → 最終
    • → 最終
  • → その他のエージェント → 最終

from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
from langgraph.graph import StateGraph, MessagesState, START, END
import sqlite3
from typing import TypedDict, Literal
from aidputils.agents.toolkit.agent_helper import init_oci_llm, pre_invoke_setup
from aidputils.agents.toolkit.configs import AIDPToolConf, OCIAIConf, ModelArgs
from typing import TypedDict, Literal
from typing import List
from langchain_core.messages import BaseMessage

##  Replace compartment id and endpoint
##
compartment_id = '<your-compartment-ocid>'
endpoint = 'https://inference.generativeai.<oci-region>.oci.oraclecloud.com'
####

checkpointer = globals().get("checkpointer", None)
conn = sqlite3.connect("file::memory:?cache=shared", uri=True)

def setup_db():
  cur = conn.cursor()

  cur.execute("""
    CREATE TABLE IF NOT EXISTS orders (
    order_id INTEGER,
    region TEXT,
    revenue REAL,
    order_date TEXT
  )
  """)

  cur.executemany(
    "INSERT INTO orders VALUES (?, ?, ?, ?)",
    [
        (1, "EMEA", 1200, "2024-10-01"),
        (2, "EMEA", 900, "2024-10-02"),
        (3, "AMER", 1500, "2024-10-01"),
        (4, "EMEA", 400, "2024-10-03"),
        (5, "APAC", 800, "2024-10-02"),
        (6, "EMEA", 1400, "2025-10-01"),
        (7, "AMER", 1200, "2025-10-01"),
        (8, "EMEA", 900, "2025-10-03"),
        (9, "APAC", 300, "2025-10-02"),    ],
  )

  conn.commit()
  #conn.close()

class State(TypedDict):
    question: str
    route: Literal["sql", "other"]
    sql: str
    rows: list
    content: str
    comparison: str
    insight: str
    messages: List[BaseMessage]

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

llm_conf = OCIAIConf(model_provider='generic',
                     compartment_id='<your-compartment-ocid>',
                     model_args=model_args,
                     endpoint=endpoint,
                     model_id='xai.grok-4',
                     guardrails_config=guardrails_config)
llm = init_oci_llm(llm_conf)


def supervisor(state: State) -> State:
    messages = [
        HumanMessage(
            content=(
                "Decide whether the following question requires SQL analysis.\n\n"
                "Respond with ONLY one word:\n"
                "- sql\n"
                "- other\n\n"
                f"Question:\n{state['question']}"
            )
        )
    ]

    response: AIMessage = llm.invoke(messages)
    route = response.content.strip().lower()

    return {"route": route}


def other_agent(state: State) -> State:
    response: AIMessage = llm.invoke(
        [HumanMessage(content=state["question"])]
    )
    return {"content": response.content.strip()}

def final(state: State) -> State:
    messages = state.get("messages", [])

    combined_answer = state["content"]
    print(state.get("insight"))
    if state.get("insight") and state["insight"] != "No additional insight.":
        combined_answer += f"\n\nInsight: {state['insight']}"

    messages.append(HumanMessage(content=state["question"]))
    messages.append(AIMessage(content=combined_answer))

    return {
        **state,
        "messages": messages,
    }



def execute_sql(query: str):
    conn = sqlite3.connect("file::memory:?cache=shared", uri=True)
    cur = conn.cursor()
    cur.execute(query)

    columns = [desc[0] for desc in cur.description]
    rows = cur.fetchall()

    conn.close()

    return columns, rows



def sql_agent(state: State) -> State:
    # 1. Generate SQL
    sql_messages = [
        HumanMessage(
            content=(
                "You are a senior data analyst.\n\n"
                "Database schema:\n"
                "orders(order_id, region, revenue, order_date)\n\n"
                "Write a SQLite-compatible SQL query that contents the question below.\n"
                "Return ONLY the SQL starting with the SELECT statement.\n\n"
                f"Question:\n{state['question']}"
            )
        )
    ]

    sql_response: AIMessage = llm.invoke(sql_messages)
    sql = sql_response.content.strip()

    # 2. Execute SQL
    columns, rows = execute_sql(sql)
    results = [dict(zip(columns, row)) for row in rows]

    # 3. Format content (LLM owns presentation)
    format_messages = [
        HumanMessage(
            content=(
                "You are a professional analytics assistant.\n\n"
                "The following data is the FINAL, correct result.\n\n"
                f"Data:\n{results}\n\n"
                f"User Question:\n{state['question']}\n\n"
                "Formatting Rules:\n"
                "- Format entire response using Markdown syntax. Include headers, bold text, and a bulleted list\n"
                "- Start with a short headline (max 12 words)\n"
                "- On the next line, give a complete sentence contenting the question\n"
                "- Clearly state the numeric value\n"
                "- Use commas in numbers\n"
                "- Do NOT mention SQL, databases, tables, or queries\n"
                "- Do NOT explain how the data was obtained\n"
                "- Do NOT add disclaimers\n\n"
                "Output Format (EXACT):\n"
                "<Headline>\n"
                "<Sentence>"

            )
        )
    ]

    formatted_response: AIMessage = llm.invoke(format_messages)
    content = formatted_response.content.strip()

    return { 
        "sql": sql,
        "rows": results,
        "content": content,
   }


def comparison_agent(state: State) -> State:
    rows = state.get("rows", [])

    messages = [
        HumanMessage(
            content=(
                "You are a business analyst.\n\n"
                "Analyze the result data and produce a comparative insight.\n\n"
                f"Result Data:\n{rows}\n\n"
                "Rules:\n"
                "- Format entire response using Markdown syntax. Include headers, bold text, and a bulleted list\n"
                "- If multiple rows exist, identify the highest, lowest, or notable difference\n"
                "- If only one value exists, explain what it represents and how it could be compared\n"
                "- Do NOT mention SQL or databases\n"
                "- One concise sentence\n"
                "- Never say 'no additional insight'\n"
            )
        )
    ]

    response: AIMessage = llm.invoke(messages)

    return {
        "comparison": response.content.strip()
    }

def insight_agent(state: State) -> State:
    messages = [
        HumanMessage(
            content=(
                "You are a senior analytics advisor.\n\n"
                f"User Question:\n{state['question']}\n\n"
                f"Comparison Insight:\n{state.get('comparison', '')}\n\n"
                "Turn this into a clear business insight.\n"
                "- One sentence\n"
                "- No speculation\n"
                "- No technical language\n"
            )
        )
    ]

    response: AIMessage = llm.invoke(messages)

    return {
        "insight": response.content.strip()
    }



class AgentBasic:
  def __init__(self) -> None:
    self.graph = None

  def setup(self) -> None:
      setup_db()
      builder = StateGraph(State)

      builder.add_node("supervisor", supervisor)
      builder.add_node("sql_agent", sql_agent)
      builder.add_node("comparison_agent", comparison_agent)
      builder.add_node("insight_agent", insight_agent)
      builder.add_node("other_agent", other_agent)
      builder.add_node("final", final)
      
      builder.set_entry_point("supervisor")

      builder.add_conditional_edges(
        "supervisor",
        lambda s: s["route"],
        {
          "sql": "sql_agent",
          "other": "other_agent",
        },
      )

      builder.add_edge("sql_agent", "comparison_agent")
      builder.add_edge("comparison_agent", "insight_agent")
      builder.add_edge("insight_agent", "final")
      builder.add_edge("other_agent", "final")
      builder.add_edge("final", END)

      if checkpointer:
        self.graph = builder.compile(checkpointer= checkpointer)
      else:
        self.graph = builder.compile()



  async def invoke(self, user_query: str, **kwargs):
    config = pre_invoke_setup(**kwargs)
    initial_state = {
        "question": user_query
    }
    try:
      return await self.graph.ainvoke(initial_state, 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())

import asyncio

async def main():
    test_agent = AgentBasic()
    test_agent.setup()
    result = await test_agent.invoke("What was the total revenue in EMEA?")
    print("Agent response:", result)
if __name__ == "__main__":
    asyncio.run(main())

リモートOracle Analytics Cloud MCPサーバー接続のサンプル・コード

このサンプル・コードは、Oracle Analytics Cloudへの独自のリモートMCPサーバー接続を設定するためのガイドラインとして使用できます。

ノート:

このコードは、既存のOracle Analytics Cloud MCPサーバーが実行中であり、説明のために含まれています。

langchain-mcp-adapters

エントリ・ファイル

ACCESS_TOKEN = "enter_your_key"

from aidputils.agents.toolkit.agent_helper import init_oci_llm, pre_invoke_setup
from aidputils.agents.toolkit.configs import OCIAIConf, ModelArgs
from aidp_flowutils.configs import AIDPToolConf, OCIAIConf, ModelArgs
from langgraph.prebuilt import create_react_agent
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
from langchain_core.messages import BaseMessage, AIMessage, HumanMessage, SystemMessage
from aidp_flowutils.agent_helper import pre_tool_setup, post_tool_setup, parse_stream_response
from typing import AsyncGenerator, Dict, Union
import logging
from langchain_mcp_adapters.client import MultiServerMCPClient 
import uuid
import os
import asyncio
from concurrent.futures import ThreadPoolExecutor

logger = logging.getLogger('test')
checkpointer = globals().get("checkpointer", None)
_executor = ThreadPoolExecutor(1)

def async_to_sync(awaitable):
    loop = asyncio.new_event_loop()
    return _executor.submit(loop.run_until_complete, awaitable).result()

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

MCP_URL = "https://<your-oac-instance-url>/api/mcp"
client = MultiServerMCPClient({ "oac": { "transport":"streamable_http", "url":MCP_URL, "headers":{ "Authorization": f"Bearer {ACCESS_TOKEN}", "Content-Type": "application/json" }}})

tools_agent = async_to_sync(client.get_tools())
##### End tool List#############

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-code-fast-1',
                     guardrails_config=guardrails_config)

## Agent class definition
class Test:
  def __init__(self) -> None:
    self.agent = None
  """
  Setup for LangGraph agent.
  """
  def setup(self) -> None:
    logger.info(llm_conf)
    oci_llm = init_oci_llm(llm_conf)
    system_prompt = """
        Be a helpful and useful assistant. Use the Oracle Analytics MCP Tools to help answer data related information.
        """
    try:
      mem_url = os.getenv("MEMORY_SERVER_URL") or os.getenv("MEMORY_URL") or "http://127.0.0.1:21100"
      from oracle_memory_clients.client import ProxyStoreClient, AsyncProxyCheckpointClient  # type: ignore
      checkpointer = AsyncProxyCheckpointClient(base_url=mem_url, agent="basic demo agent")

      if checkpointer:
        self.agent = create_react_agent(model=oci_llm, tools=tools_agent, prompt=system_prompt, debug=True, checkpointer= checkpointer)
      else:
        self.agent = create_react_agent(model=oci_llm, tools=tools_agent, 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_agent, 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) -> Union[Dict, AsyncGenerator[BaseMessage, None]]:
    config = pre_invoke_setup(**kwargs)
    user_message = HumanMessage(content=user_query)
    message = {"messages": [dict(user_message)]}
    is_stream = bool(kwargs.get("stream", False))
    if is_stream:
      return self._stream_messages(input=message, config=config, kwargs=kwargs)
    else:
      token = pre_tool_setup(**kwargs)
      try:
        agent_response = await self.agent.ainvoke(input=message, config = config)
        final_response = {**agent_response, "messages": agent_response.get("messages", [])[-1:]}
        return final_response
      except Exception as e:
        logger.error(f"Exception while calling invoke {e}")
        raise

  async def _stream_messages(self, input, config, kwargs) -> AsyncGenerator[BaseMessage, None]:
    """
    Stream messages as they are generated by the agent.
    Yields BaseMessage objects as new messages are added by nodes.
    Ensures the auth context (ContextVar) remains active during streaming and is cleaned up after.
    """
    logger.info("Starting _stream_messages")
    token = pre_tool_setup(**kwargs)

    try:
      # Determine streaming mode based on availability of StreamingData class
      try:
        from aidp_auth.client.generative_ai_inference_v2_client import StreamingData
        stream = self.agent.astream(input=input, config=config, stream_mode="messages")
      except ImportError:
        stream = self.agent.astream(input=input, config=config)

      async for chunk in parse_stream_response(stream):
        yield chunk
    except Exception:
      logger.exception("Streaming error")
      raise

import asyncio
async def main():
    # Instantiate and initialize the agent
    demo_agent = Test()
    demo_agent.setup()
    # You can customize this user query or prompt for input
    user_query = "What datasets are available in my analytics instance?"
    #user_query = "Give me a summary of the recent Ice Cream Sales data, providing highlights and lowlights in the summary."
	
    # Run the asynchronous invoke method and print the result
    result = await demo_agent.invoke(user_query, thread_id=uuid.uuid4().hex)
    print("Agent response:", result)

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