Knowledge Client
O KnowledgeClient realiza consultas RAG (Retrieval-Augmented Generation) às bases de conhecimento da plataforma Tech4AI. Use quando a action precisa buscar informações de documentos ou bases de dados vetorizadas.
Inicialização
from actions_sdk.core.types.action_context import ActionContext
from actions_sdk.features.knowledge import KnowledgeClient
from uuid import UUID
ctx: ActionContext
knowledge = KnowledgeClient(ctx)
O KnowledgeClient extrai ctx.knowledge_base_url, ctx.agent_config e ctx.client_slug automaticamente no construtor.
Exceções no construtor:
ValueError— sectx.knowledge_base_urlnão estiver definido
Métodos
query(query, knowledge_base_id)
Consulta a base de conhecimento com geração de resposta via LLM (RAG completo).
from uuid import UUID
knowledge = KnowledgeClient(ctx)
resultado = knowledge.query(
query="Qual é a política de reembolso?",
knowledge_base_id=UUID("550e8400-e29b-41d4-a716-446655440000"),
top_k=5,
)
print(resultado.answer) # Resposta gerada pelo LLM
print(resultado.status) # Status da consulta
Parâmetros:
| Parâmetro | Tipo | Padrão | Descrição |
|---|---|---|---|
query | str | — | Texto da consulta |
knowledge_base_id | UUID | — | ID da base de conhecimento a consultar |
top_k | int | 5 | Número máximo de chunks a recuperar |
rerank | bool | False | Ativar re-ranking dos resultados |
extra_context | dict | None | None | Contexto adicional para o payload da requisição |
project_prompt | str | None | None | Prompt de projeto injetado no RAG (glossário, terminologia) |
timeout | int | 30 | Timeout em segundos |
Retorna: QueryResponse
QueryResponse
Objeto com o resultado completo da consulta:
resultado = knowledge.query(query="...", knowledge_base_id=kb_id)
resultado.status # str — status geral da consulta
resultado.answer # str | None — resposta gerada pelo LLM
# Chunks recuperados
if resultado.chunks:
resultado.chunks.status # str — status do retrieval
resultado.chunks.score_threshold # float | None — limiar de score usado
resultado.chunks.chunk_search_results # list[ChunkSearchResult]
# Iterar sobre chunks
for chunk in resultado.chunks.chunk_search_results:
chunk.content # str — conteúdo do chunk
chunk.score # float — relevância do chunk
chunk.metadata # dict — metadados (source, page, etc.)
Erros e Exceções
| Exceção | Quando ocorre | Como tratar |
|---|---|---|
KnowledgeClientError | Falha na consulta à API de knowledge | Capture e retorne ResponseToAgent com ERROR |
ValueError | ctx.knowledge_base_url não definido no construtor | Indica problema de configuração do motor |
from actions_sdk.features.knowledge.client import KnowledgeClientError
try:
resultado = knowledge.query(query=query, knowledge_base_id=kb_id)
except KnowledgeClientError as erro:
return ResponseToAgent(
status=ResponseStatus.ERROR,
instruction=f"Falha na consulta à base de conhecimento: {erro}",
)
Exemplo completo
Baseado na action knowledge_search do agente de homologação:
import traceback
from uuid import UUID
from actions_sdk.core.types.action_context import ActionContext
from actions_sdk.core.types.responses import ResponseStatus, ResponseToAgent
from actions_sdk.features.conversational.state import StateManager
from actions_sdk.features.knowledge import KnowledgeClient
from actions_sdk.features.tracing import ActionSpan
ctx: ActionContext
TOP_K = 3
CHAVE_PROGRESSO = "action.knowledge.progresso"
def buscar_na_base(query: str, knowledge_base_id: str):
span = ActionSpan(ctx)
state = StateManager(ctx)
span.event("action.knowledge.inicio", {"query": query})
try:
kb_id = UUID(knowledge_base_id)
knowledge = KnowledgeClient(ctx)
resultado = knowledge.query(query=query, knowledge_base_id=kb_id, top_k=TOP_K)
chunks = (
resultado.chunks.chunk_search_results
if resultado and resultado.chunks
else []
)
state.update(CHAVE_PROGRESSO, "SUCESSO ✓")
span.event("action.knowledge.sucesso", {"chunks": len(chunks)})
return ResponseToAgent(
status=ResponseStatus.SUCCESS,
instruction=(
f"KnowledgeClient.query() concluído.\n"
f"Query: '{query}'\n"
f"Chunks retornados: {len(chunks)}\n"
f"Resposta: {resultado.answer}"
),
state_updates=state.get_full_state(),
)
except Exception as erro:
state.update(CHAVE_PROGRESSO, "ERRO ✗")
span.event("action.knowledge.erro", {"erro": str(erro)})
return ResponseToAgent(
status=ResponseStatus.ERROR,
instruction=f"Falha na consulta: {traceback.format_exc()}",
state_updates=state.get_full_state(),
)