에이전트 LangGraph 코드 샘플
제공된 LangGraph 코드 샘플을 참조로 사용하거나 Oracle AI Data Platform Workbench에서 에이전트를 시작할 수 있습니다.
Hello World와 LangGraph
에이전트 흐름에서 이 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 샘플 코드를 사용하여 반응 에이전트에서 프롬프트 툴을 테스트하고 디버그할 수 있습니다.
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 샘플 코드를 사용하여 반응 에이전트에서 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 도구 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, 'john.doe@acme.com' email_address, 75000 salary from DUAL
UNION
Select 102 employee_id, 'Jane' first_name, 'Smith' last_name, 'jane.smith@acme.com' email_address, 100000 salary from DUAL
UNION
Select 103 employee_id, 'Peter' first_name, 'Jones' last_name, 'peter.jones@acme.com' 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의 총 매출은 얼마입니까?
- 2024년 APAC 대비 EMEA의 매출은 어떻습니까?
이 에이전트 플로우의 설계는 다음과 같습니다.
- 라우터
- → SQL 에이전트
- → 비교 에이전트
- → 인사이트 에이전트
- → 최종
- → 최종
- → 비교 에이전트
- → 기타 에이전트 → 최종
- → 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())