24 Llamada a un agente desplegado
Puede llamar a la URL de punto final de su agente desde la aplicación de producción.
Independientemente de la interfaz programática que se utilice para llamar al punto final del agente, debe estar autenticado con OCI y tener los permisos pertinentes. En el caso de los puntos finales del agente, el emisor de la llamada debe tener al menos permiso USE en el punto final del agente.
El URI de punto final se documenta en el separador de detalles de la interfaz de usuario del agente. Puede copiar ese URI de punto final en el código para llamar al agente.

Métodos para Llamar a URI de Punto Final
Puede llamar al URI de punto final del agente a través de diferentes herramientas, SDK y CLI.
Los siguientes métodos le permiten llamar al URI de punto final en los agentes de Oracle AI Data Platform Workbench.
Llamar con la CLI de OCI
oci raw-request
--http-method POST
--target-uri <your-agent-flow-endpoint-uri>
--request-body '{"query":"Tell me about the Ryder Cup in 1985"}'
--auth <security_token>Llamar con biblioteca de solicitudes de Python
import oci import requests import json import uuid from contextlib import closing from requests import Request, Response
class AuthHelper: """ AuthHelper allows creating an OCI signer with either API key or security_token (which are short term sessions) """ def init(self, oci_profile: str, use_security_token:bool = True): config = oci.config.from_file(file_location="/Volumes/jr/default/misc/config",profile_name=oci_profile) if use_security_token: with open(config["security_token_file"], 'r') as f: token = f.read() private_key = oci.signer.load_private_key_from_file(config["key_file"])
self.signer = oci.auth.signers.SecurityTokenSigner(token, private_key) else: self.signer = oci.signer.Signer( tenancy=config["tenancy"], user=config["user"], fingerprint=config["fingerprint"], private_key_file_location=config["key_file"], #pass_phrase=config.get("pass_phrase"), #private_key_content=config.get("key_content") )
@property
def Signer(self):
return self.signer
class MyRawJsonRpcClient: """ Simple class using requests lib to post JSON to chat endpoint using OCI signing """ def init(self, chat_url:str, oci_profile: str, sessionKey:str, use_security_token:bool = True): self.authhelper = AuthHelper(oci_profile=oci_profile, use_security_token=use_security_token) self.authsigner = self.authhelper.Signer self.chat_url = chat_url self.sessionKey = sessionKey
def send(self, input:str) -> Response:
body = {
"isStreamEnabled" : False,
"sessionKey" : self.sessionKey,
"trace" : False,
"input" :[{
"role":"User",
"content":[{
"type" : "INPUT_TEXT",
"text" : input
}]
}]
}
response:Request = requests.post(
url =self.chat_url,
params = None,
auth = self.authsigner,
json=body,
headers={}
)
return response
sessionKey arbitrario. sessionKey es el identificador único de la sesión de usuario con el agente. Si sigue reutilizando el mismo sessionKey, los mensajes de usuario y las respuestas del agente se agregan a la misma conservación. client = MyRawJsonRpcClient(chat_url="<your-agent-flow-endpoint-uri>",
oci_profile = "DEFAULT",
sessionKey= “<your-session-key>”,
use_security_token = False )También puede proporcionar un mensaje de usuario y utilizar el cliente para enviar el mensaje al URI de punto final del agente: user_input = f"Hello, tell me a good dad joke."
r = client.send(input = user_input)
response_json = r.json()A través de APEX
Puede utilizar el ejemplo de código disponible en el repositorio de muestras de Github de AI Data Platform Workbench. El ejemplo le guía por el proceso de llamada a un punto final de despliegue de agente desde una aplicación APEX.
A través de Streamlit
Puede utilizar el ejemplo de código disponible en el repositorio de muestras de Github de AI Data Platform Workbench. El ejemplo le guía por el proceso de llamada a un punto final de despliegue de agente desde una aplicación Streamlit.
Mejores Prácticas: Respuestas Asíncronas y No Asíncronas
Le recomendamos que escriba su código de cliente suponiendo respuestas asíncronas. Por ejemplo:
import httpx
async def fetch_data():
async with httpx.AsyncClient() as client:
response = await client.get(URL)
return response.json()