Wednesday, 1 July 2026

Oracle Agentic AI Foundations and Agentic AI Question and Answer 2026

Agentic AI Foundations Associate (1Z0-1157-26)

Oracle Agentic AI Foundations - Oracle MyLearn 


Q: What is the fundamental difference between traditional GenAI and Agentic AI?

A: Traditional generative AI is "instruction-driven" and stops generating text once a prompt is completed. Agentic AI is "goal-driven". It operates autonomously by decomposing a broad objective into smaller tasks, determining the sequence of steps, executing them via external tools or APIs, evaluating the outcomes, and refining its strategy until the goal is met. 
Q: What are the foundational pillars of an AI Agent?
A: Every functioning AI agent relies on four core pillars: 
  • Planning: Task decomposition and breaking complex goals into manageable steps.
  • Memory: Remembering conversation context (short-term) and persisting historical data/preferences (long-term).
  • Tool Usage: Leveraging external APIs, databases, or code interpreters to take physical actions.
  • Reflection: Evaluating outputs, validating results, and self-correcting errors. 
Q: What is the ReAct framework in Agentic AI?
A: ReAct (Reasoning + Acting) is a pattern that enables an agent to alternate between "thinking" and "doing". The agent reasons about what to do, takes an action (like executing a SQL query), observes the result, and repeats the process until the task is complete. 
Oracle-Specific & Enterprise Implementation
Q: How does Oracle approach Agentic AI in enterprise software?
A: Oracle’s Agentic AI strategy focuses on bringing intelligent agents directly into the databases (e.g., Oracle Database 23ai Autonomous Vector Search) and business applications (Oracle Fusion Cloud) that organizations already use. These agents automate workflows like lead conversion, supply chain routing, and HR incident management using localized enterprise context. 
Q: What is the purpose of the Oracle AI Agent Studio (1Z0-1145-1)?
A: The Oracle AI Agent Studio allows administrators and developers to configure, test, and deploy pre-built AI agents. It is used to govern agent routing logic, set specific boundaries (guardrails), and integrate AI directly into ERP, HCM, or CX modules. 
Q: How does Database-Native RAG work in Oracle systems?
A: Retrieval-Augmented Generation (RAG) allows agents to query data securely stored within the Oracle Database. Instead of pulling data into an external model, the AI agent performs semantic (vector) searches directly on structured/unstructured data inside the database, ensuring high data governance, role-based access control, and verifiable citations. 
Q: What are the primary security risks when deploying autonomous AI agents in production?
A:
  • Prompt Injection: Malicious inputs tricking the agent into executing unwanted actions (e.g., deleting a database or altering permissions).
  • Data Leakage: Exposing sensitive information or PII through improper API calls.
  • Uncontrolled Decision Loops: The agent gets stuck in a cycle of repetitive, failing actions, rapidly consuming tokens or resources. 
Q: How do you add safeguards (guardrails) when giving an agent destructive tools (like modifying a database)?

A: You implement strict constraints such as read-only default permissions, "human-in-the-loop" (HITL) authorization for destructive operations, dry-run validation (testing a command before executing it), and strict token/execution limits

Question:"Explain how you would design a multi-step agentic workflow in Oracle to autonomously process and approve a vendor invoice. How do you ensure this agent is secure and doesn't execute unauthorized or incorrect actions?" 

Answer
In an Oracle environment (such as OCI Generative AI or Oracle Fusion Applications), this is achieved by defining a Reasoning and Acting (ReAct) loop that leverages enterprise data grounding. 
  1. Reasoning: The agent determines that an invoice must be approved.
  2. Tool Calling: It identifies the correct function (e.g., execute_invoice_approval) and prepares the payload.
  3. Action & Guardrails: Before pushing to Oracle ERP, it utilizes a "Human-in-the-Loop" (HITL) system and Role-Based Access Control (RBAC). It checks the user context and transaction amount. If the amount exceeds $5,000, it initiates an automatic escalation


Question : "How do you design, implement, and unit-test a goal-oriented AI Agent using the LangChain framework in Python that safely executes system commands or API lookups while maintaining a reasoning loop?" 
The Answer (Architectural Explanation)
An autonomous enterprise agent requires three core architectural layers: 
  1. The Core LLM / Brain: Processes information and plans tasks using strategies like ReAct (Reasoning and Acting). 
  2. Tools: Executable functions (e.g., executing OS commands, database lookups via Oracle Vector Search) that let the agent interact with its ecosystem. 
  3. Orchestration Loop: An execution pipeline that continuously processes the user prompt, determines the next tool call, parses the output, and updates the agent state until the final goal is met. 
To ensure production safety in an Oracle-centric cloud landscape, the agent must leverage custom guardrails to restrict unauthorized tool parameters or execution inputs. 
 Python & LangChain Implementation
This self-contained Python script uses LangChain Core Abstractions to construct a custom tool, declare an execution agent, enforce safe command validation, and handle state processing. 
python
import os
import subprocess
from typing import Dict, Any
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI

# =====================================================================
# 1. TOOL DEFINITION WITH SAFETY GUARDRAILS
# =====================================================================
@tool
def execute_system_command(command_type: str) -> str:
    """
    Executes a predefined, safe system diagnostic check.
    Allowed command types: 'disk_space', 'memory_usage'.
    """
    # Defensive Guardrail: Restrict arbitrary string injections
    allowed_commands = {
        "disk_space": ["df", "-h"],
        "memory_usage": ["free", "-m"]
    }
    
    if command_type not in allowed_commands:
        return f"Error: Command parameter '{command_type}' is unauthorized."
        
    try:
        # Secure execution utilizing tokenized arguments (prevents shell injection)
        result = subprocess.run(
            allowed_commands[command_type], 
            capture_output=True, 
            text=True, 
            timeout=5
        )
        return result.stdout if result.returncode == 0 else result.stderr
    except Exception as e:
        return f"Execution failed: {str(e)}"

# =====================================================================
# 2. AGENT ARCHITECTURE & REASONING LOOP
# =====================================================================
def build_agent_pipeline():
    # Instantiate the structural model and bind the execution tools
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    tools = [execute_system_command]
    llm_with_tools = llm.bind_tools(tools)
    
    # Structure the agent system prompt
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are an operations agent. You must resolve requests using tools. "
                   "If the user asks for unauthorized checks, explain why you cannot run them."),
        MessagesPlaceholder(variable_name="chat_history", optional=True),
        ("human", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad")
    ])
    
    # Custom output router handling Tool Calls vs Final Responses
    def route_agent_output(output) -> Any:
        if hasattr(output, "tool_calls") and output.tool_calls:
            # Route model execution to the respective tool
            return AgentAction(
                tool=output.tool_calls[0]["name"],
                tool_input=output.tool_calls[0]["args"],
                log=output.content
            )
        # Goal achieved: Return final response payload
        return AgentFinish(return_values={"output": output.content}, log=output.content)

    # Compile the runnable execution chain
    agent_chain = (
        RunnablePassthrough.assign(
            agent_scratchpad=lambda x: x.get("agent_scratchpad", [])
        )
        | prompt
        | llm_with_tools
        | route_agent_output
    )
    return agent_chain, tools

# =====================================================================
# 3. COMPACT EXECUTION ENGINE
# =====================================================================
def run_agent(user_query: str) -> str:
    agent, tools_list = build_agent_pipeline()
    tool_map = {t.name: t for t in tools_list}
    scratchpad = []
    
    # Manual loop simulating LangChain's internal orchestration lifecycle
    for _ in range(5):
        response = agent.invoke({"input": user_query, "agent_scratchpad": scratchpad})
        
        if isinstance(response, AgentFinish):
            return response.return_values["output"]
            
        if isinstance(response, AgentAction):
            # Locate tool, run context evaluation, and commit to intermediate memory
            target_tool = tool_map[response.tool]
            observation = target_tool.invoke(response.tool_input)
            scratchpad.append(response)
            # Standardize step feedback for next LLM context pass
            scratchpad.append(f"Observation: {observation}")
            
    return "Error: Agent timed out or exceeded max iterations."
 Automated Test Cases
This accompanying script uses Python's standard unittest library to validate correct workflow execution and confirm safety boundaries function as expected.
python
import unittest

class TestAgenticAIFoundations(unittest.TestCase):
    
    def test_successful_tool_execution(self):
        """Verify that valid system parameters safely execute and output logs."""
        result = execute_system_command.invoke({"command_type": "disk_space"})
        # Check standard Linux disk metrics or local platform mock fallbacks
        self.assertTrue(any(x in result for x in ["Filesystem", "Error", "failed"]))

    def test_guardrail_unauthorized_input(self):
        """Verify safety logic intercepts and denies unapproved shell inputs."""
        malicious_input = "rm -rf /"
        result = execute_system_command.invoke({"command_type": malicious_input})
        expected_err = f"Error: Command parameter '{malicious_input}' is unauthorized."
        self.assertEqual(result, expected_err)

if __name__ == "__main__":
    unittest.main(argv=['first-arg-is-ignored'], exit=False)
Command-Line Invocation Guide
Execute the setup environment, run configuration commands, and execute your script from the terminal with the following sequence:
bash
# Install the necessary underlying framework abstractions
pip install langchain-core langchain-openai

# Configure system environments for secure context processing
export OPENAI_API_KEY="your-api-key-here"

# Execute the main application runtime
python main_agent.py



An Agentic AI system represents a major shift from passive text completion to active, autonomous goal execution. Unlike basic chat interfaces that process static prompts, an Agentic AI system uses an LLM as a central brain to break down goals into subtasks, invoke external enterprise APIs (tool calling), evaluate intermediate observations, and maintain multi-turn memory state to execute business workflows independently. [1, 2, 3, 4, 5]
In an enterprise framework such as Oracle Cloud Infrastructure (OCI) Enterprise AI Platform or Oracle AI Database, agents leverage specialized frameworks: [1, 2, 3]
  1. Reasoning Engine (LLM): Processes user intent, maintains contextual state, and uses prompt patterns like Chain-of-Thought (CoT) to decide actions. [1, 2, 3]
  2. Tools (Action Layer): Python functions or microservices registered via frameworks like LangChain that bind to the model, giving it the ability to interact with real-world databases and APIs. [, 2, 3, 4]
  3. Orchestration & Reflection Loop: Uses a feedback mechanism (such as the ReAct framework) where the agent runs a continuous cycle of Thought → Action → Observation → Thought until a final answer or termination condition is reached. [1, 2, 3, 4]
Code Implementation (Python & LangChain)
Below is a production-style prototype of a LangChain Agent using the newer create_tool_calling_agent pattern. It binds an explicit execution tool, builds a localized memory buffer, and implements structured error handling. [1]
python
import os
import sys
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_openai import ChatOpenAI

# 1. Environment Setup (Ensure your API keys are configured)
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY", "mock-key-for-testing")

# 2. Tool Definition: Simulate querying an Oracle ERP database or external service
@tool
def query_enterprise_inventory(item_name: str) -> str:
    """Queries the central ERP database to find available stock counts for a specific item."""
    # Production application would use oci-cli or an Oracle python connector (oracledb)
    database_mock = {
        "autonomous_server": "In stock: 14 units. Location: Region-US-East.",
        "oci_load_balancer": "In stock: 3 units. Location: Region-EU-West.",
        "exadata_rack": "Out of stock. Backorder ETA: 5 days."
    }
    
    normalized_key = item_name.lower().replace(" ", "_")
    return database_mock.get(normalized_key, f"Item '{item_name}' was not found in the Oracle ERP system database.")

# 3. Agent Construction Class
class OracleAgenticEngine:
    def __init__(self):
        # Initialize a tool-compatible model engine
        self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
        self.tools = [query_enterprise_inventory]
        
        # Construct explicit system instructions defining the agentic boundaries
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", "You are an autonomous Oracle Enterprise AI assistant. "
                       "You must use available tools to look up factual database data before answering. "
                       "Provide direct, concise metrics based only on tool observations."),
            MessagesPlaceholder(variable_name="chat_history"),
            ("human", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad"),
        ])
        
        # Assemble the LangChain runtime components
        agent = create_tool_calling_agent(self.llm, self.tools, self.prompt)
        self.executor = AgentExecutor(
            agent=agent, 
            tools=self.tools, 
            verbose=True, 
            handle_parsing_errors=True
        )

    def run_workflow(self, user_query: str) -> str:
        # Initialize empty state context tracking per transaction loop
        response = self.executor.invoke({"input": user_query, "chat_history": []})
        return response["output"]

# Executable entry point for demonstration purposes
if __name__ == "__main__":
    if len(sys.argv) > 1:
        query = " ".join(sys.argv[1:])
        engine = OracleAgenticEngine()
        print(f"\n--- Executive Action Output ---\n{engine.run_workflow(query)}")

Execution Command
To invoke the autonomous agent workflow via your terminal window, save the code block above as agent_engine.py and run the following command:
bash
python agent_engine.py "Check inventory for the Autonomous Server and tell me if we need to restock"

Automated Test Cases
The following test suite leverages unittest to programmatically validate proper tool routing, agent fallback behaviors, and structured string evaluation. Save this script as test_agent.py. [1]
python
import unittest
from unittest.mock import MagicMock, patch
from agent_engine import OracleAgenticEngine, query_enterprise_inventory

class TestOracleAgenticEngine(unittest.TestCase):
    
    def test_tool_direct_execution(self):
        """Verifies the underlying tool function processes entries and queries strings directly."""
        result_valid = query_enterprise_inventory.invoke({"item_name": "Autonomous Server"})
        self.assertIn("14 units", result_valid)
        
        result_invalid = query_enterprise_inventory.invoke({"item_name": "Unknown Gadget"})
        self.assertIn("was not found", result_invalid)

    @patch('langchain_openai.ChatOpenAI')
    def test_agent_workflow_routing(self, mock_llm):
        """Mocks the LLM layer to confirm the engine processes and formats input requests safely."""
        # Setup mocking behaviors for the agent integration framework execution path
        mock_instance = MagicMock()
        mock_llm.return_value = mock_instance
        
        # Instantiate engine and patch executor behaviors manually for deterministic outcomes
        engine = OracleAgenticEngine()
        engine.executor.invoke = MagicMock(return_value={
            "output": "The Autonomous Server has 14 units available in Region-US-East."
        })
        
        output = engine.run_workflow("What is the current inventory status of our Autonomous Server?")
        
        # Verify execution integrity bounds
        self.assertIsInstance(output, str)
        self.assertIn("14 units", output)
        engine.executor.invoke.assert_called_once()

if __name__ == "__main__":
    unittest.main()

Run Test Suite Command:
bash
python -m unittest test_agent.py



Oracle Agentic AI Foundations
Q1: What is Oracle Agentic AI, and how does OCI Generative AI Agents service differ from standard LLM inference?
  • Answer: Oracle Agentic AI focuses on autonomous agents that can reason, use tools, 
and execute workflows to solve complex problems rather than just predicting text. 
The OCI Generative AI Agents service provides a fully managed infrastructure 
that wraps LLMs with built-in components like Retrieval-Augmented Generation (RAG), 
memory management, and enterprise data connectors (like Oracle Database) 
without requiring you to build the orchestration layer from scratch.
  • Command (OCI CLI Installation & Setup):
    bash
    pip install oci
oci setup config

    Code Example: Connecting to OCI Generative AI using the Python SDK.
  • python
    import oci


def get_oci_agent_response(agent_id, user_prompt):
    # Initializes configuration from default ~/.oci/config
    config = oci.config.from_file()
    # Create a generative AI agent runtime client
    client = oci.generative_ai_agent_runtime.GenerativeAiAgentRuntimeClient(
        config
    )

    request_details = (
        oci.generative_ai_agent_runtime.models.ChatDetails(
            user_message=user_prompt
        )
    )
    response = client.chat(agent_id=agent_id, chat_details=request_details)
    return response.data.chat_response.text

    Test Case:
  • python
    def test_oci_agent_response(mocker):
    # Mocking the OCI client response
    mock_client = mocker.patch(
        "oci.generative_ai_agent_runtime.GenerativeAiAgentRuntimeClient"
    )
    mock_instance = mock_client.return_value
    mock_instance.chat.return_value.data.chat_response.text = (
        "Hello from Oracle Agent!"
    )

    result = get_oci_agent_response("mock-agent-id", "Hi")
    assert result == "Hello from Oracle Agent!"



#### Q2: How do you integrate Oracle Database 23ai AI Vector Search into an AI agent workflow?
*   **Answer**: Oracle Database 23ai introduces native `VECTOR` data types and distance functions (like `COSINE`). This allows enterprise agents to search unstructured data directly inside the relational database using standard SQL, eliminating the need for an external vector database.
*   **Command (Database Driver Installation)**:
    ```bash
    pip install oracledb
    ```
*   **Code Example**: Querying vectors using Python `oracledb` thick/thin driver.
    ```python
    import oracledb


    def query_oracle_vectors(connection_string, user_password, query_vector):
        connection = oracledb.connect(
            user="system", password=user_password, dsn=connection_string
        )
        cursor = connection.cursor()

        # SQL executing a similarity search using standard vector distance
        sql = """
            SELECT document_text 
            FROM enterprise_knowledge 
            ORDER BY VECTOR_DISTANCE(embedding, :1, COSINE) 
            FETCH FIRST 1 ROWS ONLY
        """

        cursor.execute(sql, [str(query_vector)])
        result = cursor.fetchone()
        cursor.close()
        connection.close()
        return result[0] if result else None
    ```
*   **Test Case**:
    ```python
    def test_query_oracle_vectors(mocker):
        mock_connect = mocker.patch("oracledb.connect")
        mock_cursor = mock_connect.return_value.cursor.return_value
        mock_cursor.fetchone.return_value = ("Oracle 23ai offers native vector search.",)

        res = query_oracle_vectors("localhost/XEPDB1", "password", [0.1, 0.2, 0.3])
        assert "native vector search" in res

LangChain Orchestration
Q3: What is LangChain Expression Language (LCEL) and how do you build a basic agent loop?
  • Answer: LCEL is a declarative way to compose chains of components. For agents, we use chains to link prompts, LLMs, and output parsers, then bind tools to the LLM so it can dynamically choose actions.
  • Command (LangChain Installation):
    bash
    pip install langchain langchain-core langchain-community

    Code Example: Constructing an agent chain with a custom tool via LCEL.
  • python
    from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool


@tool
def get_server_status(server_name: str) -> str:
    """Returns the status of an enterprise server."""
    if server_name == "OCI_DB_01":
        return "Active and Healthy"
    return "Unknown Host"


def build_agent_chain(llm):
    prompt = ChatPromptTemplate.from_messages(
        [
            (
                "system",
                "You are an operations assistant. Use tools if needed.",
            ),
            ("human", "{input}"),
        ]
    )
    # Bind the tool directly to the model supporting tool calling
    llm_with_tools = llm.bind_tools([get_server_status])
    chain = prompt | llm_with_tools
    return chain

    Test Case:
  • python
    def test_get_server_status_tool():
    assert get_server_status.invoke("OCI_DB_01") == "Active and Healthy"
    assert get_server_status.invoke("Unknown") == "Unknown Host"



#### Q4: How do you implement stateful Memory in a LangChain agent using LangGraph?
*   **Answer**: Standard conversational chains store history in strings. For complex agent behaviors, LangGraph is used to manage a persistent state graph where each turn updates a key-value store, maintaining long-term and short-term dialogue states.
*   **Command**:
    ```bash
    pip install langgraph
    ```
*   **Code Example**: Creating a minimal stateful agent loop.
    ```python
    from typing import Annotated, TypedDict
    from langgraph.graph import StateGraph, START, END
    from langgraph.graph.message import add_messages


    class AgentState(TypedDict):
        messages: Annotated[list, add_messages]


    def assistant_node(state: AgentState):
        # Simulates processing logic or LLM call
        new_message = {"role": "assistant", "content": "Processed your request."}
        return {"messages": [new_message]}


    def create_graph():
        workflow = StateGraph(AgentState)
        workflow.add_node("assistant", assistant_node)
        workflow.add_edge(START, "assistant")
        workflow.add_edge("assistant", END)
        return workflow.compile()
    ```
*   **Test Case**:
    ```python
    def test_graph_flow():
        app = create_graph()
        initial_state = {"messages": [{"role": "user", "content": "Hello"}]}
        output = app.invoke(initial_state)
        assert len(output["messages"]) == 2
        assert output["messages"][-1]["content"] == "Processed your request."

Python Core & System Design for Agents
Q5: How do you design thread-safe, concurrent tool execution in a production AI agent framework?
  • Answer: Real-world agents need to run multiple API calls or tool actions concurrently without blocking the main event loop. Python's asyncio combined with ThreadPoolExecutor (for CPU-bound or legacy synchronous tools) provides a non-blocking architecture.
  • Command: (No external libraries needed, core Python features)
  • Code Example: Async tool orchestrator.
    python
    import asyncio
import time


async def run_network_tool(tool_name: str, delay: int) -> dict:
    # Simulate an asymmetric network call
    await asyncio.sleep(delay)
    return {"tool": tool_name, "status": "Success", "timestamp": time.time()}


async def orchestrate_tools(tools_list):
    # Execute multiple tools concurrently
    tasks = [run_network_tool(name, 1) for name in tools_list]
    results = await asyncio.gather(*tasks)
    return results


  • Test Case:
    python
    import pytest


@pytest.mark.asyncio
async def test_orchestrate_tools():
    start_time = time.time()
    tools = ["Tool_A", "Tool_B", "Tool_C"]
    results = await orchestrate_tools(tools)
    duration = time.time() - start_time

    assert len(results) == 3
    # If running concurrently, total time should be close to the 1-second delay, not 3 seconds.
    assert duration < 1.5
    assert results[0]["status"] == "Success"

    {content: }



Question : What is load_dotenv 

load_dotenv is a function from the python-dotenv library that reads key-value pairs from a .env file and automatically sets them as system environment variables. It is heavily used in local development to store configurations and sensitive information like API keys, database credentials, and secrets without hardcoding them into your source code. [1, 2]
Quick Start Guide
1. Install the Library
Do not install dotenv directly, as it is a different, unrelated package. Make sure to download the correct library: [1, 2, 3]
bash
pip install python-dotenv

2. Create your .env File
Place a file named exactly .env in the root directory of your project. Populate it with your configurations: [1, 2, 3]
env
DATABASE_URL=postgres://user:password@localhost:5432/mydb
API_SECRET_KEY=your_super_secret_key_here
DEBUG_MODE=True

(Important Security Step: Always add your .env file to your .gitignore so you never push your raw secrets to GitHub or Git providers!) [1]
3. Load Variables in Python
Import the function and execute it at the very beginning of your application script: 
python
import os
from dotenv import load_dotenv

# Search for and load the .env file
load_dotenv()

# Access the variables using standard system utilities
db_url = os.getenv("DATABASE_URL")
api_key = os.getenv("API_SECRET_KEY")

print(f"Connected to database: {db_url}")

Core Parameters & Capabilities
  • override=True: By default, load_dotenv() will not overwrite an environment variable if it already exists in your local machine terminal environment. If you want the values in the .env file to forcefully take precedence, pass override=True.
  • dotenv_path: If your .env file lives outside the script's root path or has a distinct name (like .env.dev), manually target it by passing the path string: load_dotenv(dotenv_path="path/to/.env.dev").
  • Variable Expansion: The library allows you to cleanly stitch configurations together internally using POSIX syntax:
    env
    DOMAIN=example.com
API_URL=https://api.${DOMAIN}/v1


Q2: How do you implement Oracle AI Vector Search in Python to give an agent memory or context?
  • Example: You use oracledb to query embeddings stored in an Oracle Database. 
python
import oracledb

def search_oracle_vector(query_embedding, connection_params):
    # Establish connection to Oracle Database 23ai
    conn = oracledb.connect(**connection_params)
    cursor = conn.cursor()
    
    # SQL query utilizing vector_distance
    sql = """
    SELECT document_text 
    FROM enterprise_knowledge 
    ORDER BY vector_distance(embedding_col, :1, COSINE) 
    FETCH FIRST 3 ROWS ONLY
    """
    
    cursor.execute(sql, [str(query_embedding)])
    results = cursor.fetchall()
    cursor.close()
    conn.close()
    return [row[0] for row in results]

  • Test Case:
python
def test_search_oracle_vector():
    mock_params = {"user": "test", "password": "pwd", "dsn": "localhost/XEPDB1"}
    fake_embedding = [0.1, 0.2, 0.3]
    # In a real test, mock the oracledb connection and cursor
    # Assert that a list of strings is returned
    assert isinstance(fake_embedding, list)

LangChain Agents & Tools
Q3: How do you create a custom LangChain tool and bind it to an Agent?
  • Command: Install LangChain packages first. 
bash
pip install langchain langchain-core langchain-openai

  • Example: Define a tool that fetches enterprise server status and bind it to an OpenAI agent.
python
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

@tool
def check_server_status(server_id: str) -> str:
    """Useful for checking the live operational status of an enterprise server."""
    # Mocking internal API call
    if server_id == "SRV-404":
        return "Offline - Maintenance required"
    return "Online - Operational"

# Initialize LLM and bind tools
llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [check_server_status]

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful IT support agent."),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

  • Test Case:
python
def test_check_server_status_tool():
    # Direct tool execution test
    offline_res = check_server_status.invoke({"server_id": "SRV-404"})
    online_res = check_server_status.invoke({"server_id": "SRV-200"})
    
    assert offline_res == "Offline - Maintenance required"
    assert online_res == "Online - Operational"

Q4: What is LangGraph, and when should you use it over standard LangChain Agents?

Answer: Standard LangChain agents operate in a linear loops 
(Thought/Action/Observation). LangGraph is used when your agent needs complex, 
cyclical state machines, multi-agent collaboration, or explicit human-in-the-loop validation steps. 
Python Advanced Concepts for AI
Q5: How do you handle asynchronous tool execution in Python to prevent blocking an AI Agent's main thread?
  • Example: Use asyncio to run long-running agent tool operations concurrently. 
python
import asyncio

async def fetch_data_from_source_a():
    await asyncio.sleep(1)  # Simulating network latency
    return "Data A"

async def fetch_data_from_source_b():
    await asyncio.sleep(2)  # Simulating database latency
    return "Data B"

async def concurrent_agent_tools():
    # Run both data-gathering tasks at the same time
    results = await asyncio.gather(fetch_data_from_source_a(), fetch_data_from_source_b())
    return results

  • Test Case:
python
import pytest

@pytest.mark.asyncio
async def test_concurrent_agent_tools():
    results = await concurrent_agent_tools()
    assert len(results) == 2
    assert "Data A" in results
    assert "Data B" in results

Q6: How do you use Pydantic (v2) to validate JSON payloads returned by an LLM?
  • Example: Ensure the agent extracts structured information correctly.
python
from pydantic import BaseModel, Field

class ExtractionSchema(BaseModel):
    action_item: str = Field(description="The task that needs to be done.")
    assignee: str = Field(description="The person responsible for the task.")
    priority_score: int = Field(description="Priority rating from 1 to 10.")

def validate_llm_output(raw_json: dict) -> ExtractionSchema:
    return ExtractionSchema(**raw_json)

  • Test Case:
python
import pytest
from pydantic import ValidationError

def test_validation_success():
    valid_data = {"action_item": "Fix database", "assignee": "Alice", "priority_score": 9}
    parsed = validate_llm_output(valid_data)
    assert parsed.assignee == "Alice"

def test_validation_failure():
    invalid_data = {"action_item": "Fix database", "assignee": "Alice", "priority_score": "high"}
    with pytest.raises(ValidationError):
        validate_llm_output(invalid_data)
on

No comments:

Post a Comment

Subscribe to: Post Comments (Atom)