Agent LangGraph - Codebeispiele
Sie können die bereitgestellten LangGraph-Codebeispiele als Referenz verwenden oder mit Agents in Oracle AI Data Platform Workbench beginnen.
Themen:
- Hallo Welt mit LangGraph
- ReAct Agent mit Eingabeaufforderung Tool LangGraph Beispielcode
- ReAct Agent mit RAG Tool LangGraph Beispielcode
- ReAct Agent mit SQL Tool LangGraph Beispielcode
- ReAct Agent mit benutzerdefinierten Tools LangGraph Beispielcode
- Multi-Agent-System – Mustercode des Vorgesetzten
- Beispielcode für Remote-Oracle Analytics Cloud MCP-Serververbindung
Hallo Welt mit LangGraph
Sie können dieses LangGraph-Codebeispiel in einem Agent-Ablauf zum Testen und Debuggen der Ausgabe verwenden.
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())ReAct Agent mit Eingabeaufforderung Tool LangGraph Beispielcode
Mit diesem LangGraph-Beispielcode können Sie Prompt-Tools in Reaktionsagenten testen und debuggen.
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())ReAct Agent mit RAG Tool LangGraph Beispielcode
Mit diesem LangGraph-Beispielcode können Sie RAG-Tools in Reaktionsagenten testen und debuggen.
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())ReAct Agent mit SQL Tool LangGraph Beispielcode
Mit diesem LangGraph-Beispielcode können Sie SQL-Tools in ReAct-Agents testen und debuggen.
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())
ReAct Agent mit benutzerdefinierten Tools LangGraph Beispielcode
Sie können dieses LangGraph-Codebeispiel in einem Agent-Ablauf verwenden, um die Ausgabe für benutzerdefinierte Tools zu testen und zu debuggen.
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())Multi-Agent-System – Mustercode des Vorgesetzten
Dieses Beispiel folgt einem Supervisor-Agent-Muster und ist ein Data Agent, der SQLite verwendet. Die SQL-Funktionen können durch das SQL-Tool ersetzt werden und die Oracle-Datenbank nutzen. Beispiel: SQLite wird zur Demonstration verwendet, da es sofort einsatzbereit ist.
Versuchen:
- Wie hoch war der Gesamtumsatz in EMEA?
- Wie sind die Umsätze in EMEA im Vergleich zu APAC im Jahr 2024 verlaufen?
Das Design dieses Agent-Flusses sieht folgendermaßen aus:
- Router
- → SQL-Agent
- → Vergleichsagent
- → Insight-Agent
- → Endgültig
- → Endgültig
- → Vergleichsagent
- → Andere Agent → Endgültig
- → SQL-Agent
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())
Beispielcode für Remote-Oracle Analytics Cloud MCP-Serververbindung
Sie können diesen Beispielcode als Richtlinie zum Einrichten Ihrer eigenen Remote-MCP-Serververbindungen zu Oracle Analytics Cloud verwenden.
Hinweis:
Dieser Code hängt davon ab, dass ein vorhandener Oracle Analytics Cloud MCP-Server ausgeführt wird und wurde zur Veranschaulichung aufgenommen.langchain-mcp-adaptersEingabedatei
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())