Compare commits

..

7 Commits

Author SHA1 Message Date
8d2b8b5be0 bugfix Alex 2026-06-30 2026-07-01 11:32:46 -03:00
9607c723a0 bugfix Alex 2026-06-30 2026-07-01 07:15:38 -03:00
d603a01039 bugfix Alex 2026-06-30 2026-07-01 07:15:04 -03:00
7893c4c8ab bugfix spans 2026-06-26 00:40:33 -03:00
ffcdbe5d85 bugfix spans 2026-06-25 23:46:09 -03:00
5b0bf3f826 bugfix Alex 2026-06-24 2026-06-24 14:24:45 -03:00
dc618bfe47 bugfix Alex 2026-06-24 2026-06-24 14:15:44 -03:00
21 changed files with 1598 additions and 342 deletions

164
.env Normal file
View File

@@ -0,0 +1,164 @@
###############################################################################
# AI AGENT PLATFORM - CONFIGURAÇÃO ÚNICA
# Este arquivo é lido por Pydantic Settings no framework e no backend template.
###############################################################################
APP_NAME=ai-agent-template
APP_ENV=local
LOG_LEVEL=INFO
API_HOST=0.0.0.0
API_PORT=8000
CORS_ORIGINS=http://localhost:5173,http://127.0.0.1:5173
###############################################################################
# LLM - OCI Generative AI como provider principal
###############################################################################
# Opções: mock, oci_openai, oci_sdk, openai_compatible
LLM_PROVIDER=oci_openai
LLM_TEMPERATURE=0.2
LLM_MAX_TOKENS=2048
LLM_TIMEOUT_SECONDS=120
# OCI OpenAI-compatible endpoint
OCI_GENAI_BASE_URL=https://inference.generativeai.us-chicago-1.oci.oraclecloud.com/openai/v1
OCI_GENAI_MODEL=openai.gpt-4.1
OCI_GENAI_API_KEY=sk-ph3FgX6iP3fxAQCXb9IpPIDTadkeeYAWntUWhzcWysIM6zsS
OCI_GENAI_PROJECT_OCID=
# OCI SDK / signer / profiles
OCI_CONFIG_FILE=~/.oci/config
OCI_PROFILE=LATINOAMERICA-Chicago
OCI_COMPARTMENT_ID=ocid1.compartment.oc1..aaaaaaaaexpiw4a7dio64mkfv2t273s2hgdl6mgfvvyv7tycalnjlvpvfl3q
OCI_REGION=us-chicago-1
###############################################################################
# Persistência
###############################################################################
# Opções: memory, autonomous, mongodb
SESSION_REPOSITORY_PROVIDER=sqlite
MEMORY_REPOSITORY_PROVIDER=sqlite
CHECKPOINT_REPOSITORY_PROVIDER=sqlite
SQLITE_DB_PATH=./data/agent_framework.db
# Autonomous Database
ADB_USER=admin
ADB_PASSWORD=Moniquinha1972
ADB_DSN=oradb23ai_high
ADB_WALLET_LOCATION=/mnt/d/Dropbox/ORACLE/LatinoAmerica/Wallet_ORADB23ai
ADB_WALLET_PASSWORD=Moniquinha1972
ADB_TABLE_PREFIX=AGENTFW
# MongoDB - também pode representar Autonomous usando API compatível com Mongo, se habilitada no ambiente
MONGODB_URI=mongodb://mongo:mongopassword@localhost:27017
MONGODB_DATABASE=agent_platform
# Redis
REDIS_URL=redis://localhost:6379/0
ENABLE_REDIS_CACHE=false
###############################################################################
# RAG / Vector / Graph
###############################################################################
VECTOR_STORE_PROVIDER=memory
GRAPH_STORE_PROVIDER=memory
RAG_TOP_K=5
EMBEDDING_PROVIDER=mock
OCI_EMBEDDING_MODEL=cohere.embed-multilingual-v3.0
RAG_FILE_GLOBS=*.md,*.txt,*.yaml,*.yml,*.json
###############################################################################
# Observabilidade
###############################################################################
ENABLE_LANGFUSE=true
LANGFUSE_TRACE_MODE=compact # Opcional: verbose, compact
LANGFUSE_PUBLIC_KEY=pk-lf-2f9da109-5b0f-4c78-b61d-9598ed787eba
LANGFUSE_SECRET_KEY=sk-lf-a4cb0cdd-f2ea-4468-9911-cebeb91ba944
LANGFUSE_HOST=http://localhost:3005
ENABLE_OTEL=false
OTEL_EXPORTER_OTLP_ENDPOINT=
OTEL_SERVICE_NAME=ai-agent-template
###############################################################################
# Analytics / Observer corporativo
###############################################################################
# Quando true, AgentObserver publica eventos IC.*, NOC.* e GRL.* nos providers abaixo.
ENABLE_ANALYTICS=false
# Providers aceitos: oci_streaming,pubsub,noop
ANALYTICS_PROVIDERS=pubsub
# Compatibilidade FIRST/TIM: pode informar AGENT_PUBSUB_TOPIC diretamente.
AGENT_PUBSUB_TOPIC=
GCP_PUBSUB_TOPIC_PATH=
GCP_PROJECT_ID=
GCP_PUBSUB_TOPIC=
GCP_PUBSUB_TIMEOUT_SECONDS=30
# Credencial GCP segue padrão Google:
# GOOGLE_APPLICATION_CREDENTIALS=/secrets/gcp-service-account.json
###############################################################################
# OCI Streaming
###############################################################################
ENABLE_OCI_STREAMING=false
OCI_STREAM_ENDPOINT=
OCI_STREAM_OCID=
OCI_STREAM_PARTITION_KEY=agent-events
###############################################################################
# Guardrails, Judges, Supervisor
###############################################################################
ENABLE_INPUT_GUARDRAILS=true
ENABLE_OUTPUT_GUARDRAILS=true
ENABLE_JUDGES=true
ENABLE_SUPERVISOR=true
ENABLE_OUTPUT_SUPERVISOR=true
ENABLE_PARALLEL_GUARDRAILS=true
GUARDRAILS_FAIL_FAST=true
OUTPUT_SUPERVISOR_MAX_RETRIES=3
GUARDRAILS_CONFIG_PATH=./config/guardrails.yaml
JUDGES_CONFIG_PATH=./config/judges.yaml
PROMPT_POLICY_PATH=./config/prompt_policy.yaml
###############################################################################
# Gateway de canais
###############################################################################
DEFAULT_CHANNEL=web
ENABLE_VOICE_ADAPTER=true
ENABLE_WHATSAPP_ADAPTER=true
ENABLE_TEXT_ADAPTER=true
#################################################
# ENTERPRISE ROUTING
#################################################
# Arquivo YAML com intents, keywords, políticas de estado e fallback.
ROUTING_CONFIG_PATH=./config/routing.yaml
# true = usa LLM para classificar quando keywords/estado não resolverem.
# Em produção, costuma ser útil; em desenvolvimento, false evita custo e latência.
ENABLE_LLM_ROUTER=true
###############################################################################
# MCP / Tools
###############################################################################
ENABLE_MCP_TOOLS=true
MCP_SERVERS_CONFIG_PATH=./agent_template_backend/config/mcp_servers.yaml
TOOLS_CONFIG_PATH=./agent_template_backend/config/tools.yaml
MCP_TOOL_TIMEOUT_SECONDS=30
# router = EnterpriseRouter seleciona um agente; supervisor = pode acionar múltiplos agentes
ROUTING_MODE=router
# Usage/cost accounting
USAGE_REPOSITORY_PROVIDER=sqlite
IDENTITY_CONFIG_PATH=./agent_template_backend/config/identity.yaml
MCP_PARAMETER_MAPPING_PATH=./agent_template_backend/config/mcp_parameter_mapping.yaml
# -----------------------------------------------------------------------------
# ConversationSummaryMemory / compressão de contexto conversacional
# -----------------------------------------------------------------------------
ENABLE_CONVERSATION_SUMMARY_MEMORY=true
MEMORY_CONTEXT_STRATEGY=summary
MEMORY_HISTORY_LIMIT=80
MEMORY_RECENT_MESSAGES_LIMIT=8
MEMORY_SUMMARY_TRIGGER_MESSAGES=20
MEMORY_MAX_SUMMARY_CHARS=6000
MEMORY_SUMMARY_USE_LLM=true
MEMORY_INJECT_RECENT_MESSAGES=true
MEMORY_INJECT_SUMMARY=true

View File

@@ -71,6 +71,9 @@ RAG_FILE_GLOBS=*.md,*.txt,*.yaml,*.yml,*.json
# Observabilidade
###############################################################################
ENABLE_LANGFUSE=true
LANGFUSE_TRACE_MODE=verbose # Opcional: verbose, compact
LANGFUSE_ROOT_SPAN_NAME=agent.gateway_message
LANGFUSE_LEGACY_IO_FALLBACK=true
LANGFUSE_PUBLIC_KEY=pk-lf-bd9b0c7e-2b8b-4e5b-a382-284a9b4413b3
LANGFUSE_SECRET_KEY=sk-lf-5f5cc18d-0bb5-424e-b5d0-cb3664d58c20
LANGFUSE_HOST=http://localhost:3005
@@ -85,7 +88,7 @@ ENABLE_LANGFUSE_OPENAI_AUTO_INSTRUMENTATION=true
# Quando true, AgentObserver publica eventos IC.*, NOC.* e GRL.* nos providers abaixo.
ENABLE_ANALYTICS=false
# Providers aceitos: oci_streaming,pubsub,noop
ANALYTICS_PROVIDERS=oci_streaming
ANALYTICS_PROVIDERS=pubsub
# Compatibilidade FIRST/TIM: pode informar AGENT_PUBSUB_TOPIC diretamente.
AGENT_PUBSUB_TOPIC=
GCP_PUBSUB_TOPIC_PATH=

118
.idea/workspace.xml generated
View File

@@ -4,11 +4,8 @@
<option name="autoReloadType" value="SELECTIVE" />
</component>
<component name="ChangeListManager">
<list default="true" id="30a0e1d8-9d7d-469b-b241-f300911cee8a" name="Changes" comment="Ajustes na documentação e remanejamento dos folders">
<list default="true" id="30a0e1d8-9d7d-469b-b241-f300911cee8a" name="Changes" comment="bugfix Alex 2026-06-30">
<change beforePath="$PROJECT_DIR$/.idea/workspace.xml" beforeDir="false" afterPath="$PROJECT_DIR$/.idea/workspace.xml" afterDir="false" />
<change beforePath="$PROJECT_DIR$/README.md" beforeDir="false" afterPath="$PROJECT_DIR$/README.md" afterDir="false" />
<change beforePath="$PROJECT_DIR$/README_en.md" beforeDir="false" afterPath="$PROJECT_DIR$/README_en.md" afterDir="false" />
<change beforePath="$PROJECT_DIR$/specs/SPEC-003-AI-Gateway.md" beforeDir="false" />
</list>
<option name="SHOW_DIALOG" value="false" />
<option name="HIGHLIGHT_CONFLICTS" value="true" />
@@ -86,7 +83,8 @@
<workItem from="1782038604623" duration="6486000" />
<workItem from="1782047166074" duration="27000" />
<workItem from="1782047194363" duration="961000" />
<workItem from="1782048494672" duration="230000" />
<workItem from="1782048494672" duration="2790000" />
<workItem from="1782900499027" duration="911000" />
</task>
<task id="LOCAL-00001" summary="Ajustes na documentação e remanejamento dos folders">
<option name="closed" value="true" />
@@ -112,7 +110,111 @@
<option name="project" value="LOCAL" />
<updated>1782048304579</updated>
</task>
<option name="localTasksCounter" value="4" />
<task id="LOCAL-00004" summary="Ajustes na documentação e remanejamento dos folders">
<option name="closed" value="true" />
<created>1782048747421</created>
<option name="number" value="00004" />
<option name="presentableId" value="LOCAL-00004" />
<option name="project" value="LOCAL" />
<updated>1782048747421</updated>
</task>
<task id="LOCAL-00005" summary="bugfix Alex 2026-06-24">
<option name="closed" value="true" />
<created>1782321346355</created>
<option name="number" value="00005" />
<option name="presentableId" value="LOCAL-00005" />
<option name="project" value="LOCAL" />
<updated>1782321346355</updated>
</task>
<task id="LOCAL-00006" summary="bugfix Alex 2026-06-30">
<option name="closed" value="true" />
<created>1782900906490</created>
<option name="number" value="00006" />
<option name="presentableId" value="LOCAL-00006" />
<option name="project" value="LOCAL" />
<updated>1782900906490</updated>
</task>
<task id="LOCAL-00007" summary="bugfix Alex 2026-06-30">
<option name="closed" value="true" />
<created>1782900910793</created>
<option name="number" value="00007" />
<option name="presentableId" value="LOCAL-00007" />
<option name="project" value="LOCAL" />
<updated>1782900910793</updated>
</task>
<task id="LOCAL-00008" summary="bugfix Alex 2026-06-30">
<option name="closed" value="true" />
<created>1782900913579</created>
<option name="number" value="00008" />
<option name="presentableId" value="LOCAL-00008" />
<option name="project" value="LOCAL" />
<updated>1782900913579</updated>
</task>
<task id="LOCAL-00009" summary="bugfix Alex 2026-06-30">
<option name="closed" value="true" />
<created>1782900914906</created>
<option name="number" value="00009" />
<option name="presentableId" value="LOCAL-00009" />
<option name="project" value="LOCAL" />
<updated>1782900914906</updated>
</task>
<task id="LOCAL-00010" summary="bugfix Alex 2026-06-30">
<option name="closed" value="true" />
<created>1782900923834</created>
<option name="number" value="00010" />
<option name="presentableId" value="LOCAL-00010" />
<option name="project" value="LOCAL" />
<updated>1782900923834</updated>
</task>
<task id="LOCAL-00011" summary="bugfix Alex 2026-06-30">
<option name="closed" value="true" />
<created>1782900925229</created>
<option name="number" value="00011" />
<option name="presentableId" value="LOCAL-00011" />
<option name="project" value="LOCAL" />
<updated>1782900925229</updated>
</task>
<task id="LOCAL-00012" summary="bugfix Alex 2026-06-30">
<option name="closed" value="true" />
<created>1782900926577</created>
<option name="number" value="00012" />
<option name="presentableId" value="LOCAL-00012" />
<option name="project" value="LOCAL" />
<updated>1782900926577</updated>
</task>
<task id="LOCAL-00013" summary="bugfix Alex 2026-06-30">
<option name="closed" value="true" />
<created>1782900927457</created>
<option name="number" value="00013" />
<option name="presentableId" value="LOCAL-00013" />
<option name="project" value="LOCAL" />
<updated>1782900927457</updated>
</task>
<task id="LOCAL-00014" summary="bugfix Alex 2026-06-30">
<option name="closed" value="true" />
<created>1782900928320</created>
<option name="number" value="00014" />
<option name="presentableId" value="LOCAL-00014" />
<option name="project" value="LOCAL" />
<updated>1782900928320</updated>
</task>
<task id="LOCAL-00015" summary="bugfix Alex 2026-06-30">
<option name="closed" value="true" />
<created>1782900928490</created>
<option name="number" value="00015" />
<option name="presentableId" value="LOCAL-00015" />
<option name="project" value="LOCAL" />
<updated>1782900928490</updated>
</task>
<task id="LOCAL-00016" summary="bugfix Alex 2026-06-30">
<option name="closed" value="true" />
<created>1782900939364</created>
<option name="number" value="00016" />
<option name="presentableId" value="LOCAL-00016" />
<option name="project" value="LOCAL" />
<updated>1782900939364</updated>
</task>
<option name="localTasksCounter" value="17" />
<servers />
</component>
<component name="TypeScriptGeneratedFilesManager">
@@ -120,6 +222,8 @@
</component>
<component name="VcsManagerConfiguration">
<MESSAGE value="Ajustes na documentação e remanejamento dos folders" />
<option name="LAST_COMMIT_MESSAGE" value="Ajustes na documentação e remanejamento dos folders" />
<MESSAGE value="bugfix Alex 2026-06-24" />
<MESSAGE value="bugfix Alex 2026-06-30" />
<option name="LAST_COMMIT_MESSAGE" value="bugfix Alex 2026-06-30" />
</component>
</project>

View File

@@ -342,6 +342,7 @@ RAG_TOP_K=5
EMBEDDING_PROVIDER=mock
ENABLE_LANGFUSE=false
LANGFUSE_TRACE_MODE=compact # Opcional: verbose, compact
LANGFUSE_HOST=http://localhost:3005
ENABLE_OTEL=false
OTEL_SERVICE_NAME=ai-agent-template
@@ -414,6 +415,7 @@ Para usar Langfuse:
```env
ENABLE_LANGFUSE=true
LANGFUSE_TRACE_MODE=compact # Opcional: verbose, compact
LANGFUSE_PUBLIC_KEY=<public-key>
LANGFUSE_SECRET_KEY=<secret-key>
LANGFUSE_HOST=http://localhost:3005
@@ -8553,6 +8555,7 @@ Verifique:
```env
ENABLE_LANGFUSE=true
LANGFUSE_TRACE_MODE=compact # Opcional: verbose, compact
LANGFUSE_PUBLIC_KEY=<public-key>
LANGFUSE_SECRET_KEY=<secret-key>
LANGFUSE_HOST=http://localhost:3005

View File

@@ -340,6 +340,7 @@ RAG_TOP_K=5
EMBEDDING_PROVIDER=mock
ENABLE_LANGFUSE=false
LANGFUSE_TRACE_MODE=compact # Opcional: verbose, compact
LANGFUSE_HOST=http://localhost:3005
ENABLE_OTEL=false
OTEL_SERVICE_NAME=ai-agent-template
@@ -412,6 +413,7 @@ To use Langfuse:
```env
ENABLE_LANGFUSE=true
LANGFUSE_TRACE_MODE=compact # Opcional: verbose, compact
LANGFUSE_PUBLIC_KEY=<public-key>
LANGFUSE_SECRET_KEY=<secret-key>
LANGFUSE_HOST=http://localhost:3005
@@ -8461,6 +8463,7 @@ Check:
```env
ENABLE_LANGFUSE=true
LANGFUSE_TRACE_MODE=compact # Opcional: verbose, compact
LANGFUSE_PUBLIC_KEY=<public-key>
LANGFUSE_SECRET_KEY=<secret-key>
LANGFUSE_HOST=http://localhost:3005

View File

@@ -99,6 +99,9 @@ class Settings(BaseSettings):
OCI_EMBEDDING_MODEL: str = 'cohere.embed-multilingual-v3.0'
ENABLE_LANGFUSE: bool = False
LANGFUSE_TRACE_MODE: Literal['verbose','compact'] = 'verbose'
LANGFUSE_ROOT_SPAN_NAME: str = 'agent.gateway_message'
LANGFUSE_LEGACY_IO_FALLBACK: bool = True
LANGFUSE_PUBLIC_KEY: str | None = None
LANGFUSE_SECRET_KEY: str | None = None
LANGFUSE_HOST: str = 'https://cloud.langfuse.com'

View File

@@ -74,20 +74,23 @@ class MockLLMProvider(LLMProvider):
profile_found = kwargs.get("profile_found")
profiles_enabled = kwargs.get("profiles_enabled")
profiles_path = kwargs.get("profiles_path")
last = messages[-1].get("content", "") if messages else ""
answer = f"[mock-llm] Resposta simulada para: {last[:300]}"
usage = {"prompt_tokens": max(1, len(str(messages))//4), "completion_tokens": max(1, len(answer)//4), "total_tokens": max(2, (len(str(messages))+len(answer))//4), "cost_usd": 0.0, "cost_brl": 0.0}
llm_metadata = {"provider": "mock", "profile_name": profile_name, "component": component_name, "model": model, "profile_source": profile_source, "profile_found": profile_found, "profiles_enabled": profiles_enabled, "profiles_path": profiles_path}
async with _maybe_generation(
self.telemetry,
name=generation_name,
model=model,
input=messages,
metadata=llm_metadata,
model_parameters={},
) as generation:
last = messages[-1].get("content", "") if messages else ""
answer = f"[mock-llm] Resposta simulada para: {last[:300]}"
usage = {"prompt_tokens": max(1, len(str(messages))//4), "completion_tokens": max(1, len(answer)//4), "total_tokens": max(2, (len(str(messages))+len(answer))//4), "cost_usd": 0.0, "cost_brl": 0.0}
generation.set_output(answer)
generation.set_usage(usage)
generation.set_metadata(**usage)
if self.usage_repository:
await self.usage_repository.record(UsageRecord.from_usage("mock", model, generation_name, usage, {"provider":"mock", "profile_name": profile_name, "component": component_name, "model": model, "profile_source": profile_source, "profile_found": profile_found, "profiles_enabled": profiles_enabled, "profiles_path": profiles_path}))
if self.telemetry:
await self.telemetry.generation(
name=generation_name,
model=model,
input=messages,
output=answer,
metadata={"provider": "mock", "profile_name": profile_name, "component": component_name, "model": model, "profile_source": profile_source, "profile_found": profile_found, "profiles_enabled": profiles_enabled, "profiles_path": profiles_path, **usage},
usage=usage,
)
await self.usage_repository.record(UsageRecord.from_usage("mock", model, generation_name, usage, llm_metadata))
return answer
@@ -259,6 +262,22 @@ class OCICompatibleOpenAIProvider(LLMProvider):
for optional_key in ("top_p", "frequency_penalty", "presence_penalty"):
if effective.get(optional_key) is not None:
request_kwargs[optional_key] = effective[optional_key]
model_parameters = {
key: value
for key, value in request_kwargs.items()
if key not in {"model", "messages"} and value is not None
}
llm_metadata = {
"provider": provider,
"model": model,
"component": component_name,
"profile_name": resolved_profile_name,
"requested_profile_name": requested_profile_name,
"profile_source": profile_source,
"profile_found": profile_found,
"profiles_enabled": bool(effective.get("profiles_enabled")),
"profiles_path": effective.get("profiles_path"),
}
async with _maybe_span(
self.telemetry,
@@ -275,47 +294,35 @@ class OCICompatibleOpenAIProvider(LLMProvider):
profiles_enabled=bool(effective.get("profiles_enabled")),
):
try:
resp = await client.chat.completions.create(**request_kwargs)
answer = resp.choices[0].message.content or ""
async with _maybe_generation(
self.telemetry,
name=generation_name,
model=model,
input=messages,
metadata=llm_metadata,
model_parameters=model_parameters,
) as generation:
resp = await client.chat.completions.create(**request_kwargs)
answer = resp.choices[0].message.content or ""
usage_metadata = self.token_collector.enrich(model, getattr(resp, "usage", None))
usage_metadata.update({
"profile_name": resolved_profile_name,
"requested_profile_name": requested_profile_name,
"profile_source": profile_source,
"profile_found": profile_found,
"component": component_name,
"model": model,
"provider": provider,
"temperature": temperature,
"max_tokens": max_tokens,
})
llm_metadata = {
"provider": provider,
"model": model,
"component": component_name,
"profile_name": resolved_profile_name,
"requested_profile_name": requested_profile_name,
"profile_source": profile_source,
"profile_found": profile_found,
"profiles_enabled": bool(effective.get("profiles_enabled")),
"profiles_path": effective.get("profiles_path"),
"temperature": temperature,
"max_tokens": max_tokens,
}
usage_metadata = self.token_collector.enrich(model, getattr(resp, "usage", None))
usage_metadata.update({
"profile_name": resolved_profile_name,
"requested_profile_name": requested_profile_name,
"profile_source": profile_source,
"profile_found": profile_found,
"component": component_name,
"model": model,
"provider": provider,
**model_parameters,
})
generation.set_output(answer)
generation.set_usage(usage_metadata)
generation.set_metadata(**usage_metadata)
if self.usage_repository:
await self.usage_repository.record(
UsageRecord.from_usage(provider, model, generation_name, usage_metadata, llm_metadata)
)
if self.telemetry:
await self.telemetry.generation(
name=generation_name,
model=model,
input=messages,
output=answer,
metadata={**llm_metadata, **usage_metadata},
usage=usage_metadata,
)
return answer
except Exception as exc:
@@ -550,6 +557,19 @@ class OCISDKProvider(LLMProvider):
)
service_endpoint = self._resolve_endpoint(self.settings, endpoint)
model_parameters = {
"temperature": temperature,
"max_tokens": max_tokens,
}
llm_metadata = {
"provider": "oci_sdk",
"model": model,
"endpoint_id": endpoint_id,
"service_endpoint": service_endpoint,
"component": component_name,
"profile_name": profile_name,
"auth_mode": getattr(self.settings, "OCI_AUTH_MODE", "config_file"),
}
async with _maybe_span(
self.telemetry,
@@ -575,26 +595,30 @@ class OCISDKProvider(LLMProvider):
max_tokens=max_tokens,
)
response = await asyncio.to_thread(client.chat, details)
answer = self._extract_answer(response)
async with _maybe_generation(
self.telemetry,
name=generation_name,
model=model,
input=messages,
metadata=llm_metadata,
model_parameters=model_parameters,
) as generation:
response = await asyncio.to_thread(client.chat, details)
answer = self._extract_answer(response)
usage_metadata = {
"prompt_tokens": max(1, len(str(messages)) // 4),
"completion_tokens": max(1, len(answer) // 4),
"total_tokens": max(2, (len(str(messages)) + len(answer)) // 4),
"cost_usd": 0.0,
"cost_brl": 0.0,
"estimated_usage": True,
"provider": "oci_sdk",
"model": model,
"endpoint_id": endpoint_id,
"service_endpoint": service_endpoint,
"component": component_name,
"profile_name": profile_name,
"auth_mode": getattr(self.settings, "OCI_AUTH_MODE", "config_file"),
}
llm_metadata = dict(usage_metadata)
usage_metadata = {
"prompt_tokens": max(1, len(str(messages)) // 4),
"completion_tokens": max(1, len(answer) // 4),
"total_tokens": max(2, (len(str(messages)) + len(answer)) // 4),
"cost_usd": 0.0,
"cost_brl": 0.0,
"estimated_usage": True,
**llm_metadata,
**model_parameters,
}
generation.set_output(answer)
generation.set_usage(usage_metadata)
generation.set_metadata(**usage_metadata)
if self.usage_repository:
await self.usage_repository.record(
@@ -607,16 +631,6 @@ class OCISDKProvider(LLMProvider):
)
)
if self.telemetry:
await self.telemetry.generation(
name=generation_name,
model=model,
input=messages,
output=answer,
metadata=llm_metadata,
usage=usage_metadata,
)
return answer
@@ -656,3 +670,36 @@ class _maybe_span:
if self.cm:
return await self.cm.__aexit__(exc_type, exc, tb)
return False
class _NoopGeneration:
def set_output(self, output: Any) -> None:
pass
def set_usage(self, usage: dict[str, Any] | None) -> None:
pass
def set_metadata(self, **metadata: Any) -> None:
pass
def set_model_parameters(self, **model_parameters: Any) -> None:
pass
class _maybe_generation:
def __init__(self, telemetry, **attrs: Any):
self.telemetry = telemetry
self.attrs = attrs
self.cm = None
self.noop = _NoopGeneration()
async def __aenter__(self):
if not self.telemetry or not hasattr(self.telemetry, "generation_span"):
return self.noop
self.cm = self.telemetry.generation_span(**self.attrs)
return await self.cm.__aenter__()
async def __aexit__(self, exc_type, exc, tb):
if self.cm:
return await self.cm.__aexit__(exc_type, exc, tb)
return False

View File

@@ -21,6 +21,7 @@ _workflow_id: ContextVar[str | None] = ContextVar("workflow_id", default=None)
_message_id: ContextVar[str | None] = ContextVar("message_id", default=None)
_trace_id: ContextVar[str | None] = ContextVar("trace_id", default=None)
_current_observation_id: ContextVar[str | None] = ContextVar("current_observation_id", default=None)
_current_span_events: ContextVar[list[dict[str, Any]] | None] = ContextVar("current_span_events", default=None)
@dataclass(slots=True)
class ObservabilityContext:
@@ -66,6 +67,31 @@ def reset_current_observation_id(token) -> None:
_current_observation_id.set(None)
def get_current_span_events() -> list[dict[str, Any]] | None:
"""Return the mutable aggregate-event bucket for the active macro span."""
return _current_span_events.get()
def set_current_span_events(events: list[dict[str, Any]] | None):
"""Set aggregate-event bucket and return ContextVar token."""
return _current_span_events.set(events)
def reset_current_span_events(token) -> None:
"""Restore previous aggregate-event bucket."""
try:
_current_span_events.reset(token)
except Exception:
_current_span_events.set(None)
def record_current_span_event(event: dict[str, Any]) -> None:
"""Append an event summary to the active macro span, if one exists."""
events = _current_span_events.get()
if events is not None:
events.append(event)
def set_observability_context(**kwargs: Any) -> ObservabilityContext:
if not kwargs.get("request_id") and not _request_id.get():
kwargs["request_id"] = str(uuid4())
@@ -82,7 +108,7 @@ def set_observability_context(**kwargs: Any) -> ObservabilityContext:
def clear_observability_context() -> None:
for var in (_request_id, _session_id, _user_id, _tenant_id, _agent_id, _channel, _ura_call_id, _workflow_id, _message_id, _trace_id, _current_observation_id):
for var in (_request_id, _session_id, _user_id, _tenant_id, _agent_id, _channel, _ura_call_id, _workflow_id, _message_id, _trace_id, _current_observation_id, _current_span_events):
var.set(None)

View File

@@ -3,6 +3,36 @@ import time
from contextlib import asynccontextmanager
from typing import Any
_LANGGRAPH_STEP_ORDER = {
"__start__": 0,
"input_guardrails": 1,
"routing_decision": 2,
"billing_agent": 3,
"product_agent": 3,
"orders_agent": 3,
"support_agent": 3,
"handoff": 3,
"supervisor_agent": 3,
"output_supervisor": 4,
"output_guardrails": 5,
"judge": 6,
"supervisor_review": 7,
"persist": 8,
"__end__": 9,
}
def _langgraph_step(name: str, state: dict[str, Any]) -> int:
explicit_steps = state.get("langgraph_steps")
if isinstance(explicit_steps, dict) and name in explicit_steps:
try:
return int(explicit_steps[name])
except (TypeError, ValueError):
pass
return _LANGGRAPH_STEP_ORDER.get(name, 50)
class LangGraphDeepTelemetry:
"""Eventos profundos do LangGraph no padrão FIRST.
@@ -16,7 +46,16 @@ class LangGraphDeepTelemetry:
async def node(self, name: str, state: dict[str, Any] | None = None):
state=state or {}
session_id=state.get('conversation_key') or state.get('session_id')
payload={'node': name, 'session_id': session_id, 'agent_id': state.get('agent_id'), 'tenant_id': state.get('tenant_id'), 'input_size': len(str(state.get('user_text') or state.get('sanitized_input') or ''))}
payload={
'node': name,
'langgraph_node': name,
'langgraph_step': _langgraph_step(name, state),
'framework': 'langgraph',
'session_id': session_id,
'agent_id': state.get('agent_id'),
'tenant_id': state.get('tenant_id'),
'input_size': len(str(state.get('user_text') or state.get('sanitized_input') or '')),
}
start=time.time()
await self.telemetry.event('langgraph.node.started', payload, kind='langgraph')
async with self.telemetry.span(f'langgraph.node.{name}', **payload):

View File

@@ -15,15 +15,20 @@ import logging
import re
import time
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from typing import Any
from uuid import uuid4
from .context import (
context_metadata,
get_current_observation_id,
get_current_span_events,
get_observability_context,
record_current_span_event,
reset_current_observation_id,
reset_current_span_events,
set_current_observation_id,
set_current_span_events,
set_observability_context,
)
from .event_bus import TelemetryEventBus
@@ -32,6 +37,24 @@ from .otel import OpenTelemetryProvider
logger = logging.getLogger("agent_framework.telemetry")
_LANGFUSE_OBSERVATION_TYPES = {"span", "generation", "agent", "tool", "chain", "retriever", "embedding", "evaluator", "guardrail"}
_LANGFUSE_START_OBSERVATION_KWARGS = {
"trace_context",
"name",
"as_type",
"input",
"output",
"metadata",
"version",
"level",
"status_message",
"completion_start_time",
"model",
"model_parameters",
"usage_details",
"cost_details",
"prompt",
"end_on_exit",
}
def _langfuse_type(kind: str | None) -> str:
# Langfuse SDKs do not accept arbitrary event types such as "event"; FIRST pattern
@@ -42,6 +65,19 @@ def _langfuse_type(kind: str | None) -> str:
_LANGFUSE_TRACE_ID_RE = re.compile(r"^[0-9a-f]{32}$")
_COMPACT_SUPPRESSED_SPAN_PREFIXES = (
"llm.chat_completion",
"workflow.agent.",
"workflow.handoff",
"workflow.input_guardrails",
"workflow.judge",
"workflow.output_guardrails",
"workflow.output_supervisor",
"workflow.persist",
"workflow.routing_decision",
"workflow.supervisor_review",
)
_COMPACT_VISIBLE_EVENT_PREFIXES = ("AGA.", "NOC.")
def _raw_correlation_id(attrs: dict[str, Any] | None = None) -> str | None:
@@ -100,6 +136,7 @@ def _inject_langfuse_trace_context(kwargs: dict[str, Any], attrs: dict[str, Any]
which grouped everything in one trace while flattening the tree.
"""
attrs = attrs or kwargs.get("metadata") or {}
ignore_current_parent = bool(attrs.get("_ignore_current_parent") or kwargs.get("_ignore_current_parent"))
raw_id = _raw_correlation_id(attrs)
trace_id = _langfuse_trace_id(raw_id)
parent_id = (
@@ -107,7 +144,7 @@ def _inject_langfuse_trace_context(kwargs: dict[str, Any], attrs: dict[str, Any]
or attrs.get("parent_span_id")
or kwargs.get("parent_observation_id")
or kwargs.get("parent_span_id")
or get_current_observation_id()
or (None if ignore_current_parent else get_current_observation_id())
)
if trace_id:
trace_context = dict(kwargs.get("trace_context") or {})
@@ -121,6 +158,8 @@ def _inject_langfuse_trace_context(kwargs: dict[str, Any], attrs: dict[str, Any]
metadata.setdefault("langfuse_trace_id", trace_id)
if parent_id:
metadata.setdefault("parent_observation_id", str(parent_id))
metadata.pop("_ignore_current_parent", None)
kwargs.pop("_ignore_current_parent", None)
return kwargs
@@ -150,6 +189,128 @@ def _extract_observation_id(observation: Any) -> str | None:
pass
return None
def _is_compact_visible_event(name: str) -> bool:
return str(name or "").startswith(_COMPACT_VISIBLE_EVENT_PREFIXES)
class _SpanHandle:
"""Mutable handle yielded by Telemetry.span for setting final output."""
def __init__(self, observation: Any | None = None) -> None:
self.observation = observation
self.output: Any = None
self.has_output = False
self.metadata: dict[str, Any] = {}
def set_observation(self, observation: Any | None) -> None:
self.observation = observation
def set_output(self, output: Any) -> None:
self.output = output
self.has_output = True
def set_metadata(self, **metadata: Any) -> None:
self.metadata.update({k: v for k, v in metadata.items() if v is not None})
def __getattr__(self, name: str) -> Any:
if self.observation is None:
raise AttributeError(name)
return getattr(self.observation, name)
class _GenerationHandle:
"""Mutable handle yielded by Telemetry.generation_span."""
def __init__(self, observation: Any | None = None) -> None:
self.observation = observation
self.output: Any = None
self.has_output = False
self.metadata: dict[str, Any] = {}
self.usage: dict[str, Any] | None = None
self.model_parameters: dict[str, Any] = {}
def set_observation(self, observation: Any | None) -> None:
self.observation = observation
def set_output(self, output: Any) -> None:
self.output = output
self.has_output = True
def set_usage(self, usage: dict[str, Any] | None) -> None:
self.usage = dict(usage or {})
def set_metadata(self, **metadata: Any) -> None:
self.metadata.update({k: v for k, v in metadata.items() if v is not None})
def set_model_parameters(self, **model_parameters: Any) -> None:
self.model_parameters.update({k: v for k, v in model_parameters.items() if v is not None})
def __getattr__(self, name: str) -> Any:
if self.observation is None:
raise AttributeError(name)
return getattr(self.observation, name)
def _usage_details_from_usage(usage: dict[str, Any] | None) -> dict[str, int] | None:
if not isinstance(usage, dict):
return None
def int_value(*keys: str) -> int | None:
for key in keys:
value = usage.get(key)
if value is None:
continue
try:
return int(value)
except (TypeError, ValueError):
continue
return None
input_tokens = int_value("input", "input_tokens", "prompt_tokens")
output_tokens = int_value("output", "output_tokens", "completion_tokens")
total_tokens = int_value("total", "total_tokens")
# Langfuse self-hosted versions may sum all custom usage keys into totalUsage.
# Send split fields only when available; send total only when there is no split.
details: dict[str, int] = {}
if input_tokens is not None:
details["input"] = input_tokens
if output_tokens is not None:
details["output"] = output_tokens
if not details and total_tokens is not None:
details["total"] = total_tokens
return details or None
def _cost_details_from_usage(usage: dict[str, Any] | None) -> dict[str, float] | None:
if not isinstance(usage, dict):
return None
details: dict[str, float] = {}
if usage.get("cost_usd") is not None:
try:
details["total"] = float(usage["cost_usd"])
except (TypeError, ValueError):
pass
if usage.get("cost_brl") is not None:
try:
details["total_brl"] = float(usage["cost_brl"])
except (TypeError, ValueError):
pass
return details or None
def _clean_mapping(value: dict[str, Any] | None) -> dict[str, Any] | None:
if not isinstance(value, dict):
return None
clean = {k: v for k, v in value.items() if v is not None}
return clean or None
def _utc_iso_ms() -> str:
return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
class Telemetry:
def __init__(self, settings):
self.settings = settings
@@ -188,6 +349,15 @@ class Telemetry:
def is_enabled(self) -> bool:
return bool(self.enabled and self.langfuse)
def is_compact_mode(self) -> bool:
mode = getattr(self.settings, "LANGFUSE_TRACE_MODE", "verbose") or "verbose"
return str(mode).lower() == "compact"
def _should_emit_langfuse_span(self, name: str) -> bool:
if not self.is_compact_mode():
return True
return not str(name).startswith(_COMPACT_SUPPRESSED_SPAN_PREFIXES)
def bind_context(self, **kwargs: Any):
return set_observability_context(**kwargs)
@@ -199,6 +369,10 @@ class Telemetry:
"""Cria span correlacionado em logs, Langfuse e OpenTelemetry."""
start = time.time()
attrs = context_metadata(attrs)
attrs.setdefault("_span_name", name)
is_root_span = bool(attrs.get("_root_span")) or name == "agent.gateway_message"
if self.is_compact_mode() and is_root_span and not attrs.get("parent_observation_id"):
attrs["_ignore_current_parent"] = True
if not attrs.get("request_id"):
attrs["request_id"] = str(uuid4())
if not attrs.get("trace_id"):
@@ -206,45 +380,90 @@ class Telemetry:
set_observability_context(request_id=attrs.get("request_id"), trace_id=attrs.get("trace_id"))
observation_cm = None
observation = None
handle = _SpanHandle()
observation_token = None
parent_observation_id = attrs.get("parent_observation_id") or get_current_observation_id()
propagation_cm = None
legacy_io_update: dict[str, Any] | None = None
ignore_current_parent = bool(attrs.get("_ignore_current_parent"))
parent_observation_id = attrs.get("parent_observation_id")
if not parent_observation_id and not ignore_current_parent:
parent_observation_id = get_current_observation_id()
if parent_observation_id:
attrs.setdefault("parent_observation_id", str(parent_observation_id))
logger.info("span.start %s %s", name, _safe(attrs))
otel_cm = self.otel.span(name, attrs)
otel_span = otel_cm.__enter__()
if self.is_enabled():
emit_langfuse_span = self.is_enabled() and self._should_emit_langfuse_span(name)
span_events: list[dict[str, Any]] | None = [] if emit_langfuse_span and self.is_compact_mode() else None
span_events_token = set_current_span_events(span_events) if span_events is not None else None
observation_metadata = {k: v for k, v in attrs.items() if k != "input" and not str(k).startswith("_")}
if emit_langfuse_span:
observation_cm = self._start_observation(
name=name,
as_type="span",
input=attrs.get("input"),
metadata={k: v for k, v in attrs.items() if k != "input"},
metadata=observation_metadata,
_ignore_current_parent=attrs.get("_ignore_current_parent"),
)
try:
if observation_cm is not None:
observation = observation_cm.__enter__()
handle.set_observation(observation)
observation_id = _extract_observation_id(observation)
if observation_id:
observation_token = set_current_observation_id(observation_id)
attrs.setdefault("observation_id", observation_id)
self._update_trace_from_attrs(observation, attrs)
if is_root_span:
self._update_trace_from_attrs(observation, attrs)
self._set_trace_io(observation, input=attrs.get("input"))
propagation_cm = self._start_trace_attribute_propagation(name, attrs)
if propagation_cm is not None:
propagation_cm.__enter__()
# Publish span.started only after the Langfuse observation is current,
# so secondary analytics/exporters can attach it as a child instead
# of creating a sibling/root entry.
await self.event_bus.publish(f"{name}.started", attrs, kind="span")
yield observation
yield handle
duration_ms = int((time.time() - start) * 1000)
out = {"status": "ok", "duration_ms": duration_ms}
self._update_observation(observation, output=out, metadata={"duration_ms": duration_ms})
status = {"status": "ok", "duration_ms": duration_ms}
out = handle.output if handle.has_output else status
metadata = {**observation_metadata, **status, **handle.metadata}
if span_events is not None:
metadata["aggregated_event_count"] = len(span_events)
metadata["aggregated_events"] = span_events
self._update_observation(observation, input=attrs.get("input"), output=out, metadata=metadata)
if is_root_span:
self._set_trace_io(observation, input=attrs.get("input"), output=out)
legacy_io_update = {
"input": attrs.get("input"),
"output": out,
"metadata": metadata,
}
if otel_span is not None:
otel_span.set_attribute("duration_ms", duration_ms)
await self.event_bus.publish(f"{name}.completed", {**attrs, **out}, kind="span")
completed_payload = {**attrs, **status}
if handle.has_output:
completed_payload["output"] = out
await self.event_bus.publish(f"{name}.completed", completed_payload, kind="span")
logger.info("span.end %s duration_ms=%s", name, duration_ms)
except Exception as exc:
duration_ms = int((time.time() - start) * 1000)
out = {"status": "error", "error": str(exc), "duration_ms": duration_ms}
self._update_observation(observation, level="ERROR", status_message=str(exc), output=out, metadata={"duration_ms": duration_ms})
metadata = {**observation_metadata, "duration_ms": duration_ms}
if span_events is not None:
metadata["aggregated_event_count"] = len(span_events)
metadata["aggregated_events"] = span_events
self._update_observation(observation, level="ERROR", status_message=str(exc), input=attrs.get("input"), output=out, metadata=metadata)
if is_root_span:
self._set_trace_io(observation, input=attrs.get("input"), output=out)
legacy_io_update = {
"input": attrs.get("input"),
"output": out,
"metadata": metadata,
"level": "ERROR",
"status_message": str(exc),
}
if otel_span is not None:
try:
otel_span.record_exception(exc)
@@ -255,11 +474,23 @@ class Telemetry:
logger.exception("span.error %s %s", name, exc)
raise
finally:
if propagation_cm is not None:
try: propagation_cm.__exit__(None, None, None)
except Exception: logger.debug("Falha ao encerrar propagação Langfuse", exc_info=True)
if observation_cm is not None:
try: observation_cm.__exit__(None, None, None)
except Exception: logger.exception("Falha ao finalizar span Langfuse %s", name)
if legacy_io_update is not None:
self._legacy_observation_update(
observation,
observation_type="span",
name=name,
**legacy_io_update,
)
if observation_token is not None:
reset_current_observation_id(observation_token)
if span_events_token is not None:
reset_current_span_events(span_events_token)
try: otel_cm.__exit__(None, None, None)
except Exception: logger.debug("Falha ao fechar span OTEL", exc_info=True)
@@ -267,6 +498,24 @@ class Telemetry:
payload = context_metadata(payload or {})
logger.info("event %s %s", name, _safe(payload))
await self.event_bus.publish(name, payload, kind=kind)
if self.is_compact_mode():
if get_current_span_events() is not None:
record_current_span_event({
"name": name,
"kind": kind,
"payload": payload,
})
if not _is_compact_visible_event(name) or not self.is_enabled():
return
try:
metadata = {**payload, "event_kind": kind}
cm = self._start_observation(name=name, as_type="span", input=payload, metadata=metadata)
if cm is not None:
with cm as obs:
self._update_observation(obs, input=payload, output={"status": "ok"}, metadata=metadata)
except Exception:
logger.exception("Falha ao enviar event compacto via observation")
return
if not self.is_enabled():
return
# IMPORTANT: do not call ``langfuse.event(...)`` directly here. In SDK
@@ -274,54 +523,179 @@ class Telemetry:
# a new trace row for every telemetry event. We create a correlated
# observation instead, using request_id/trace_id as the stable trace id.
try:
cm = self._start_observation(name=name, as_type=_langfuse_type(kind), metadata={**payload, "event_kind": kind})
metadata = {**payload, "event_kind": kind}
if self.is_compact_mode():
metadata["_ignore_current_parent"] = True
cm = self._start_observation(name=name, as_type=_langfuse_type(kind), metadata=metadata)
if cm is not None:
with cm: pass
except Exception:
logger.exception("Falha ao enviar event via observation")
async def generation(self, name: str, model: str, input: list | dict | str, output: str,
metadata: dict[str, Any] | None = None, usage: dict[str, Any] | None = None):
@asynccontextmanager
async def generation_span(
self,
name: str,
model: str,
input: list | dict | str,
*,
metadata: dict[str, Any] | None = None,
usage: dict[str, Any] | None = None,
model_parameters: dict[str, Any] | None = None,
):
metadata = context_metadata(metadata or {})
# Keep the actual LLM model visible both in Langfuse's generation.model field
# and in metadata for filtering/debugging across SDK versions.
metadata.setdefault("model", model)
metadata.setdefault("llm_model", model)
metadata.setdefault("component", metadata.get("profile_name") or name)
if usage:
metadata["usage"] = usage
logger.info("generation %s model=%s component=%s profile=%s metadata=%s", name, model, metadata.get("component"), metadata.get("profile_name"), _safe(metadata))
await self.event_bus.publish(name, {"model": model, "llm_model": model, "output_chars": len(output or ""), **metadata}, kind="generation")
if not self.is_enabled():
return
clean_model_parameters = _clean_mapping(model_parameters)
if clean_model_parameters:
metadata.setdefault("model_parameters", clean_model_parameters)
handle = _GenerationHandle()
observation_cm = None
observation = None
observation_token = None
legacy_io_update: dict[str, Any] | None = None
logger.info("generation.start %s model=%s component=%s profile=%s metadata=%s", name, model, metadata.get("component"), metadata.get("profile_name"), _safe(metadata))
try:
kwargs = dict(name=name, as_type="generation", input=input, output=output, model=model, metadata=metadata)
if usage:
kwargs["usage"] = usage
kwargs["usage_details"] = {k: usage.get(k) for k in ("prompt_tokens", "completion_tokens", "total_tokens", "cached_tokens", "reasoning_tokens") if k in usage}
# Prefer current/correlated generation APIs. Avoid raw
# ``langfuse.generation(...)`` first because it can create a separate
# trace row per LLM call when no current observation exists.
if hasattr(self.langfuse, "start_as_current_generation"):
clean = {k: v for k, v in kwargs.items() if k != "as_type" and v is not None}
clean = _inject_langfuse_trace_context(clean, metadata)
if self.is_enabled():
try:
with self.langfuse.start_as_current_generation(**clean) as obs:
self._update_observation(obs, output=output, model=model, metadata=metadata)
return
except TypeError:
clean.pop("trace_context", None)
with self.langfuse.start_as_current_generation(**clean) as obs:
self._update_observation(obs, output=output, model=model, metadata=metadata)
return
observation_cm = self._start_observation(
name=name,
as_type="generation",
input=input,
model=model,
model_parameters=clean_model_parameters,
usage_details=_usage_details_from_usage(usage),
cost_details=_cost_details_from_usage(usage),
metadata=metadata,
)
if observation_cm is not None:
observation = observation_cm.__enter__()
handle.set_observation(observation)
observation_id = _extract_observation_id(observation)
if observation_id:
observation_token = set_current_observation_id(observation_id)
except Exception:
observation_cm = None
observation = None
logger.exception("Falha ao iniciar generation Langfuse %s", name)
yield handle
final_usage = handle.usage if handle.usage is not None else usage
final_model_parameters = {
**(clean_model_parameters or {}),
**handle.model_parameters,
} or None
final_metadata = {**metadata, **handle.metadata}
if final_usage:
final_metadata["usage"] = final_usage
output = handle.output if handle.has_output else None
usage_details = _usage_details_from_usage(final_usage)
cost_details = _cost_details_from_usage(final_usage)
self._update_observation(
observation,
input=input,
output=output,
model=model,
metadata=final_metadata,
model_parameters=final_model_parameters,
usage_details=usage_details,
cost_details=cost_details,
)
legacy_io_update = {
"input": input,
"output": output,
"model": model,
"metadata": final_metadata,
"model_parameters": final_model_parameters,
"usage_details": usage_details,
"cost_details": cost_details,
}
await self.event_bus.publish(
name,
{
"model": model,
"llm_model": model,
"output_chars": len(output or "") if isinstance(output, str) else 0,
**final_metadata,
},
kind="generation",
)
logger.info("generation.end %s model=%s", name, model)
except Exception as exc:
final_usage = handle.usage if handle.usage is not None else usage
final_model_parameters = {
**(clean_model_parameters or {}),
**handle.model_parameters,
} or None
final_metadata = {**metadata, **handle.metadata}
if final_usage:
final_metadata["usage"] = final_usage
usage_details = _usage_details_from_usage(final_usage)
cost_details = _cost_details_from_usage(final_usage)
output = handle.output if handle.has_output else None
self._update_observation(
observation,
level="ERROR",
status_message=str(exc),
input=input,
output=output,
model=model,
metadata=final_metadata,
model_parameters=final_model_parameters,
usage_details=usage_details,
cost_details=cost_details,
)
legacy_io_update = {
"input": input,
"output": output,
"model": model,
"metadata": final_metadata,
"model_parameters": final_model_parameters,
"usage_details": usage_details,
"cost_details": cost_details,
"level": "ERROR",
"status_message": str(exc),
}
await self.event_bus.publish(f"{name}.failed", {"model": model, "llm_model": model, "error": str(exc), **final_metadata}, kind="generation")
logger.exception("generation.error %s model=%s exc=%s", name, model, exc)
raise
finally:
if observation_cm is not None:
try: observation_cm.__exit__(None, None, None)
except Exception: logger.exception("Falha ao finalizar generation Langfuse %s", name)
if legacy_io_update is not None:
self._legacy_observation_update(
observation,
observation_type="generation",
name=name,
**legacy_io_update,
)
if observation_token is not None:
reset_current_observation_id(observation_token)
cm = self._start_observation(**kwargs)
if cm is not None:
with cm as obs:
self._update_observation(obs, output=output, model=model, metadata=metadata)
except Exception:
logger.exception("Falha ao registrar generation no Langfuse")
async def generation(
self,
name: str,
model: str,
input: list | dict | str,
output: str,
metadata: dict[str, Any] | None = None,
usage: dict[str, Any] | None = None,
model_parameters: dict[str, Any] | None = None,
):
async with self.generation_span(
name=name,
model=model,
input=input,
metadata=metadata,
usage=usage,
model_parameters=model_parameters,
) as generation:
generation.set_output(output)
if usage:
generation.set_usage(usage)
async def rag_event(self, name: str, query: str, results_count: int, metadata: dict[str, Any] | None = None):
await self.event(f"rag.{name}", {"query": query, "results_count": results_count, **(metadata or {})}, kind="rag")
@@ -364,10 +738,16 @@ class Telemetry:
def _start_observation(self, **kwargs):
if not self.is_enabled(): return None
if hasattr(self.langfuse, "start_as_current_observation"):
clean = {k: v for k, v in kwargs.items() if v is not None}
clean = {k: v for k, v in kwargs.items() if v is not None and k in _LANGFUSE_START_OBSERVATION_KWARGS}
if "as_type" in clean:
clean["as_type"] = _langfuse_type(clean.get("as_type"))
clean = _inject_langfuse_trace_context(clean, clean.get("metadata") or {})
if self.is_compact_mode():
clean.pop("_ignore_current_parent", None)
else:
clean = _inject_langfuse_trace_context(clean, clean.get("metadata") or {})
metadata = clean.get("metadata")
if isinstance(metadata, dict):
clean["metadata"] = {k: v for k, v in metadata.items() if not str(k).startswith("_")}
try:
return self.langfuse.start_as_current_observation(**clean)
except (TypeError, ValueError):
@@ -413,9 +793,66 @@ class Telemetry:
if hasattr(observation, "update"): observation.update(**clean)
except Exception: logger.debug("Observation update não suportado", exc_info=True)
def _legacy_observation_update(self, observation, *, observation_type: str, name: str, **kwargs):
"""Compatibility fallback for Langfuse servers that drop OTEL observation I/O."""
if not self.is_enabled() or not bool(getattr(self.settings, "LANGFUSE_LEGACY_IO_FALLBACK", True)):
return
if observation is None:
return
obs_id = _extract_observation_id(observation)
trace_id = getattr(observation, "trace_id", None)
if not obs_id or not trace_id:
return
api = getattr(self.langfuse, "api", None)
ingestion = getattr(api, "ingestion", None)
if ingestion is None or not hasattr(ingestion, "batch"):
return
clean = {k: v for k, v in kwargs.items() if v is not None}
if not any(k in clean for k in ("input", "output", "metadata")):
return
try:
if hasattr(self.langfuse, "flush"):
self.langfuse.flush()
if observation_type == "generation":
from langfuse.api.ingestion.types import (
IngestionEvent_GenerationUpdate,
UpdateGenerationBody,
)
body = UpdateGenerationBody(id=str(obs_id), trace_id=str(trace_id), name=name, **clean)
event = IngestionEvent_GenerationUpdate(
id=str(uuid4()),
timestamp=_utc_iso_ms(),
body=body,
metadata={"source": "agent_framework", "fallback": "legacy_observation_io"},
)
else:
from langfuse.api.ingestion.types import IngestionEvent_SpanUpdate, UpdateSpanBody
body = UpdateSpanBody(id=str(obs_id), trace_id=str(trace_id), name=name, **clean)
event = IngestionEvent_SpanUpdate(
id=str(uuid4()),
timestamp=_utc_iso_ms(),
body=body,
metadata={"source": "agent_framework", "fallback": "legacy_observation_io"},
)
response = ingestion.batch(
batch=[event],
metadata={"source": "agent_framework", "fallback": "legacy_observation_io"},
)
if getattr(response, "errors", None):
logger.debug("Langfuse legacy I/O fallback retornou erros: %s", response.errors)
except Exception:
logger.debug("Falha no fallback legado de input/output Langfuse", exc_info=True)
def _update_trace_from_attrs(self, observation, attrs: dict[str, Any]):
if observation is None: return
trace_attrs = {}
if attrs.get("_span_name"):
trace_attrs["name"] = attrs["_span_name"]
for key in ("session_id", "user_id"):
if attrs.get(key): trace_attrs[key] = attrs[key]
if attrs.get("input"): trace_attrs["input"] = attrs["input"]
@@ -427,6 +864,43 @@ class Telemetry:
if hasattr(observation, "update_trace"): observation.update_trace(**trace_attrs)
except Exception: logger.debug("Trace update não suportado", exc_info=True)
def _set_trace_io(self, observation, *, input: Any | None = None, output: Any | None = None):
if observation is None: return
try:
if hasattr(observation, "set_trace_io"):
observation.set_trace_io(input=input, output=output)
return
if hasattr(observation, "update_trace"):
payload = {}
if input is not None:
payload["input"] = input
if output is not None:
payload["output"] = output
if payload:
observation.update_trace(**payload)
except Exception: logger.debug("Trace input/output update não suportado", exc_info=True)
def _start_trace_attribute_propagation(self, name: str, attrs: dict[str, Any]):
if not self.is_enabled() or not hasattr(self.langfuse, "propagate_attributes"):
return None
metadata = {
k: attrs.get(k)
for k in ("request_id", "trace_id", "agent_id", "tenant_id", "channel", "message_id", "ura_call_id", "workflow_id")
if attrs.get(k)
}
tags = attrs.get("tags") if isinstance(attrs.get("tags"), list) else None
try:
return self.langfuse.propagate_attributes(
user_id=str(attrs["user_id"]) if attrs.get("user_id") is not None else None,
session_id=str(attrs["session_id"]) if attrs.get("session_id") is not None else None,
metadata=metadata or None,
tags=[str(tag) for tag in tags] if tags else None,
trace_name=name,
)
except Exception:
logger.debug("Trace attribute propagation não suportada", exc_info=True)
return None
class _LegacyObservationContext:
def __init__(self, observation): self.observation = observation
def __enter__(self): return self.observation

View File

@@ -25,8 +25,6 @@ OCI_GENAI_MODEL=openai.gpt-4.1
OCI_GENAI_API_KEY=sk-ph3FgX6ph3FgX6ph3FgX6ph3FgX6ph3FgX6ph3FgX6
OCI_GENAI_PROJECT_OCID=
# OCI_AUTH_MODE=config_file|instance_principal|resource_principal
OCI_AUTH_MODE=config_file
# OCI SDK / signer / profiles
OCI_CONFIG_FILE=~/.oci/config
OCI_PROFILE=DEFAULT
@@ -37,9 +35,10 @@ OCI_REGION=us-chicago-1
# Persistência
###############################################################################
# Opções: memory, autonomous, mongodb
SESSION_REPOSITORY_PROVIDER=autonomous
MEMORY_REPOSITORY_PROVIDER=autonomous
CHECKPOINT_REPOSITORY_PROVIDER=autonomous
SESSION_REPOSITORY_PROVIDER=sqlite
MEMORY_REPOSITORY_PROVIDER=sqlite
CHECKPOINT_REPOSITORY_PROVIDER=sqlite
SQLITE_DB_PATH=./data/agent_framework.db
# Autonomous Database
ADB_USER=admin
@@ -60,8 +59,8 @@ ENABLE_REDIS_CACHE=false
###############################################################################
# RAG / Vector / Graph
###############################################################################
VECTOR_STORE_PROVIDER=memory
GRAPH_STORE_PROVIDER=memory
VECTOR_STORE_PROVIDER=sqlite
GRAPH_STORE_PROVIDER=sqlite
RAG_TOP_K=5
EMBEDDING_PROVIDER=mock
OCI_EMBEDDING_MODEL=cohere.embed-multilingual-v3.0
@@ -71,8 +70,9 @@ RAG_FILE_GLOBS=*.md,*.txt,*.yaml,*.yml,*.json
# Observabilidade
###############################################################################
ENABLE_LANGFUSE=true
LANGFUSE_PUBLIC_KEY=pk-lf-bd9b0c7e-2b8b-4e5b-a382-284a9b4413b3
LANGFUSE_SECRET_KEY=sk-lf-5f5cc18d-0bb5-424e-b5d0-cb3664d58c20
LANGFUSE_TRACE_MODE=compact # Opcional: verbose, compact
LANGFUSE_PUBLIC_KEY=pk-lf-2f9da109-5b0f-4c78-b61d-9598ed787eba
LANGFUSE_SECRET_KEY=sk-lf-a4cb0cdd-f2ea-4468-9911-cebeb91ba944
LANGFUSE_HOST=http://localhost:3005
ENABLE_OTEL=false
OTEL_EXPORTER_OTLP_ENDPOINT=
@@ -85,7 +85,7 @@ ENABLE_LANGFUSE_OPENAI_AUTO_INSTRUMENTATION=true
# Quando true, AgentObserver publica eventos IC.*, NOC.* e GRL.* nos providers abaixo.
ENABLE_ANALYTICS=false
# Providers aceitos: oci_streaming,pubsub,noop
ANALYTICS_PROVIDERS=oci_streaming
ANALYTICS_PROVIDERS=pubsub
# Compatibilidade FIRST/TIM: pode informar AGENT_PUBSUB_TOPIC diretamente.
AGENT_PUBSUB_TOPIC=
GCP_PUBSUB_TOPIC_PATH=
@@ -150,7 +150,7 @@ MCP_TOOL_TIMEOUT_SECONDS=30
ROUTING_MODE=router
# Usage/cost accounting
USAGE_REPOSITORY_PROVIDER=autonomous
USAGE_REPOSITORY_PROVIDER=sqlite
IDENTITY_CONFIG_PATH=./config/identity.yaml
MCP_PARAMETER_MAPPING_PATH=./config/mcp_parameter_mapping.yaml

View File

@@ -74,6 +74,7 @@ logger.info("Framework channel input mode: %s", gateway.input_mode)
@app.middleware("http")
async def observability_context_middleware(request: Request, call_next):
clear_observability_context()
request_id = request.headers.get("x-request-id") or str(uuid4())
set_observability_context(
request_id=request_id,
@@ -99,6 +100,8 @@ async def observability_context_middleware(request: Request, call_next):
"duration_ms": int((time.time() - started) * 1000),
}, kind="http")
raise
finally:
clear_observability_context()
class GatewayRequest(BaseModel):
@@ -108,6 +111,31 @@ class GatewayRequest(BaseModel):
tenant_id: str | None = None
def _metadata_value(payload: dict, key: str):
metadata = payload.get("metadata")
if isinstance(metadata, dict):
return metadata.get(key)
return None
def _extract_workflow_id(payload: dict) -> str | None:
return (
payload.get("workflow_id")
or payload.get("workflowId")
or _metadata_value(payload, "workflow_id")
or _metadata_value(payload, "workflowId")
)
def _format_root_span_name(template: str | None, values: dict) -> str:
template = template or "agent.gateway_message"
try:
return template.format(**{k: v or "unknown" for k, v in values.items()})
except Exception:
logger.warning("LANGFUSE_ROOT_SPAN_NAME inválido: %s", template)
return "agent.gateway_message"
def _resolve_identity(req: GatewayRequest, msg) -> tuple[AgentIdentity, dict, BusinessContext, list[str]]:
payload = req.payload or {}
context = dict(msg.context or {})
@@ -143,9 +171,11 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
msg = await gateway.normalize(req.channel, req.payload)
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
payload = req.payload or {}
identity, normalized_context, business_context, missing_identity_keys = _resolve_identity(req, msg)
agent_session_id = identity.conversation_key()
message_id = (req.payload or {}).get("message_id") or str(uuid4())
message_id = payload.get("message_id") or str(uuid4())
workflow_id = _extract_workflow_id(payload)
set_observability_context(
session_id=agent_session_id,
user_id=msg.user_id,
@@ -153,7 +183,8 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
agent_id=identity.agent_id,
channel=msg.channel,
message_id=message_id,
ura_call_id=(req.payload or {}).get("ura_call_id") or normalized_context.get("ura_call_id") or business_context.interaction_key,
workflow_id=workflow_id,
ura_call_id=payload.get("ura_call_id") or normalized_context.get("ura_call_id") or business_context.interaction_key,
)
stream = sse_hub.stream_for(agent_session_id)
@@ -209,34 +240,56 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
await sse_hub.emit(agent_session_id, "message.received", {"session_id": agent_session_id, "role": "user"}) if emit_sse else None
history = [m.model_dump(mode="json") for m in await memory.list(agent_session_id)]
trace_input = {
cms_input = {
"channel": req.channel,
"tenant_id": req.tenant_id,
"agent_id": req.agent_id,
"payload": payload,
}
trace_context = {
"text": msg.text,
"channel": msg.channel,
"channel_id": msg.channel_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"conversation_key": agent_session_id,
"workflow_id": workflow_id,
"message_id": message_id,
"business_context": business_context.model_dump(),
"identity_missing": missing_identity_keys,
}
root_span_name = _format_root_span_name(
getattr(settings, "LANGFUSE_ROOT_SPAN_NAME", "agent.gateway_message"),
{
"workflow_id": workflow_id,
"channel": msg.channel,
"agent_id": identity.agent_id,
"tenant_id": identity.tenant_id,
},
)
root_tags = ["agent-template", msg.channel, f"agent:{identity.agent_id}", f"tenant:{identity.tenant_id}"]
if workflow_id:
root_tags.append(f"workflow:{workflow_id}")
async with telemetry.span(
"agent.gateway_message",
root_span_name,
session_id=agent_session_id,
user_id=session.user_id,
channel=msg.channel,
input=trace_input,
tags=["agent-template", msg.channel, f"agent:{identity.agent_id}", f"tenant:{identity.tenant_id}"],
):
await telemetry.event("gateway.message.received", trace_input)
await sse_hub.emit(agent_session_id, "workflow.started", trace_input) if emit_sse else None
workflow_id=workflow_id,
input=cms_input,
tags=root_tags,
_root_span=True,
) as root_span:
await telemetry.event("gateway.message.received", trace_context)
await sse_hub.emit(agent_session_id, "workflow.started", trace_context) if emit_sse else None
result = await workflow.ainvoke(
{
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"session_id": agent_session_id,
"conversation_key": agent_session_id,
"workflow_id": workflow_id,
"agent_profile": normalized_context["agent_profile"],
"user_text": msg.text,
"history": history,
@@ -246,6 +299,7 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
"original_session_id": msg.session_id,
"session_id": agent_session_id,
"conversation_key": agent_session_id,
"workflow_id": workflow_id,
"user_id": session.user_id,
"channel": msg.channel,
"message_id": message_id,
@@ -256,66 +310,68 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
}
)
await checkpoints.put(agent_session_id, {"state": result, "message_id": message_id})
await sse_hub.emit(agent_session_id, "workflow.completed", {"session_id": agent_session_id, "route": result.get("route"), "intent": result.get("intent")}) if emit_sse else None
await checkpoints.put(agent_session_id, {"state": result, "message_id": message_id})
await sse_hub.emit(agent_session_id, "workflow.completed", {"session_id": agent_session_id, "route": result.get("route"), "intent": result.get("intent")}) if emit_sse else None
answer = result.get("final_answer") or result.get("answer") or ""
await memory.append(
agent_session_id,
ChatMessage(
role="assistant",
content=answer,
metadata={
answer = result.get("final_answer") or result.get("answer") or ""
await memory.append(
agent_session_id,
ChatMessage(
role="assistant",
content=answer,
metadata={
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"message_id": f"assistant-{message_id}",
"route": result.get("route"),
"intent": result.get("intent"),
"route_decision": result.get("route_decision"),
"judges": result.get("judge_results"),
},
),
)
await telemetry.event(
"gateway.message.responded",
{
"session_id": agent_session_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"message_id": f"assistant-{message_id}",
"route": result.get("route"),
"intent": result.get("intent"),
"answer_chars": len(answer),
},
)
response = ChannelResponse(
channel=msg.channel,
session_id=agent_session_id,
text=answer,
metadata={
"channel_id": msg.channel_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"original_session_id": msg.session_id,
"conversation_key": agent_session_id,
"workflow_id": workflow_id,
"message_id": message_id,
"route": result.get("route"),
"intent": result.get("intent"),
"route_decision": result.get("route_decision"),
"domain": result.get("domain"),
"mcp_tools": result.get("mcp_tools"),
"mcp_results": result.get("mcp_results"),
"business_context": business_context.model_dump(),
"identity_missing": missing_identity_keys,
"judges": result.get("judge_results"),
"guardrails": result.get("guardrail_decisions"),
},
),
)
await telemetry.event(
"gateway.message.responded",
{
"session_id": agent_session_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"route": result.get("route"),
"intent": result.get("intent"),
"answer_chars": len(answer),
},
)
response = ChannelResponse(
channel=msg.channel,
session_id=agent_session_id,
text=answer,
metadata={
"channel_id": msg.channel_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"original_session_id": msg.session_id,
"conversation_key": agent_session_id,
"message_id": message_id,
"route": result.get("route"),
"intent": result.get("intent"),
"route_decision": result.get("route_decision"),
"domain": result.get("domain"),
"mcp_tools": result.get("mcp_tools"),
"mcp_results": result.get("mcp_results"),
"business_context": business_context.model_dump(),
"identity_missing": missing_identity_keys,
"judges": result.get("judge_results"),
"guardrails": result.get("guardrail_decisions"),
},
)
rendered = await gateway.render(response)
await sse_hub.emit(agent_session_id, "message.responded", rendered) if emit_sse else None
await sse_hub.emit(agent_session_id, "flow.end", {"session_id": agent_session_id, "message_id": message_id}) if emit_sse else None
return rendered
)
rendered = await gateway.render(response)
root_span.set_output(rendered)
await sse_hub.emit(agent_session_id, "message.responded", rendered) if emit_sse else None
await sse_hub.emit(agent_session_id, "flow.end", {"session_id": agent_session_id, "message_id": message_id}) if emit_sse else None
return rendered
@app.get("/health")

View File

@@ -6,6 +6,7 @@ class AgentState(TypedDict, total=False):
agent_id: str
session_id: str
conversation_key: str
workflow_id: str
agent_profile: dict[str, Any]
user_text: str
sanitized_input: str

View File

@@ -320,41 +320,36 @@ class AgentWorkflow:
}
async def billing_agent(self, state):
async with self.langgraph_telemetry.node("billing_agent", state):
async with self.telemetry.span(
"workflow.agent.billing",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.billing.run(state)
async with self.telemetry.span(
"workflow.agent.billing",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.billing.run(state)
async def product_agent(self, state):
async with self.langgraph_telemetry.node("product_agent", state):
async with self.telemetry.span(
"workflow.agent.product",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.product.run(state)
async with self.telemetry.span(
"workflow.agent.product",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.product.run(state)
async def orders_agent(self, state):
async with self.langgraph_telemetry.node("orders_agent", state):
async with self.telemetry.span(
"workflow.agent.orders",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.orders.run(state)
async with self.telemetry.span(
"workflow.agent.orders",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.orders.run(state)
async def support_agent(self, state):
async with self.langgraph_telemetry.node("support_agent", state):
async with self.telemetry.span(
"workflow.agent.support",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.support.run(state)
async with self.telemetry.span(
"workflow.agent.support",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.support.run(state)
async def supervisor_agent(self, state):
"""Executa um ou mais agentes no modo supervisor e consolida a resposta.

View File

@@ -25,8 +25,6 @@ OCI_GENAI_MODEL=openai.gpt-4.1
OCI_GENAI_API_KEY=sk-ph3FgX6ph3FgX6ph3FgX6ph3FgX6ph3FgX6ph3FgX6
OCI_GENAI_PROJECT_OCID=
# OCI_AUTH_MODE=config_file|instance_principal|resource_principal
OCI_AUTH_MODE=config_file
# OCI SDK / signer / profiles
OCI_CONFIG_FILE=~/.oci/config
OCI_PROFILE=DEFAULT
@@ -37,9 +35,10 @@ OCI_REGION=us-chicago-1
# Persistência
###############################################################################
# Opções: memory, autonomous, mongodb
SESSION_REPOSITORY_PROVIDER=autonomous
MEMORY_REPOSITORY_PROVIDER=autonomous
CHECKPOINT_REPOSITORY_PROVIDER=autonomous
SESSION_REPOSITORY_PROVIDER=sqlite
MEMORY_REPOSITORY_PROVIDER=sqlite
CHECKPOINT_REPOSITORY_PROVIDER=sqlite
SQLITE_DB_PATH=./data/agent_framework.db
# Autonomous Database
ADB_USER=admin
@@ -60,8 +59,8 @@ ENABLE_REDIS_CACHE=false
###############################################################################
# RAG / Vector / Graph
###############################################################################
VECTOR_STORE_PROVIDER=memory
GRAPH_STORE_PROVIDER=memory
VECTOR_STORE_PROVIDER=sqlite
GRAPH_STORE_PROVIDER=sqlite
RAG_TOP_K=5
EMBEDDING_PROVIDER=mock
OCI_EMBEDDING_MODEL=cohere.embed-multilingual-v3.0
@@ -71,8 +70,9 @@ RAG_FILE_GLOBS=*.md,*.txt,*.yaml,*.yml,*.json
# Observabilidade
###############################################################################
ENABLE_LANGFUSE=true
LANGFUSE_PUBLIC_KEY=pk-lf-bd9b0c7e-2b8b-4e5b-a382-284a9b4413b3
LANGFUSE_SECRET_KEY=sk-lf-5f5cc18d-0bb5-424e-b5d0-cb3664d58c20
LANGFUSE_TRACE_MODE=compact # Opcional: verbose, compact
LANGFUSE_PUBLIC_KEY=pk-lf-2f9da109-5b0f-4c78-b61d-9598ed787eba
LANGFUSE_SECRET_KEY=sk-lf-a4cb0cdd-f2ea-4468-9911-cebeb91ba944
LANGFUSE_HOST=http://localhost:3005
ENABLE_OTEL=false
OTEL_EXPORTER_OTLP_ENDPOINT=
@@ -85,7 +85,7 @@ ENABLE_LANGFUSE_OPENAI_AUTO_INSTRUMENTATION=true
# Quando true, AgentObserver publica eventos IC.*, NOC.* e GRL.* nos providers abaixo.
ENABLE_ANALYTICS=false
# Providers aceitos: oci_streaming,pubsub,noop
ANALYTICS_PROVIDERS=oci_streaming
ANALYTICS_PROVIDERS=pubsub
# Compatibilidade FIRST/TIM: pode informar AGENT_PUBSUB_TOPIC diretamente.
AGENT_PUBSUB_TOPIC=
GCP_PUBSUB_TOPIC_PATH=
@@ -122,9 +122,6 @@ PROMPT_POLICY_PATH=./config/prompt_policy.yaml
# Gateway de canais
###############################################################################
DEFAULT_CHANNEL=web
# embedded = backend may parse simple/native channel payloads.
# external = backend only accepts GatewayRequest normalized by an external Channel Gateway.
FRAMEWORK_CHANNEL_INPUT_MODE=embedded
ENABLE_VOICE_ADAPTER=true
ENABLE_WHATSAPP_ADAPTER=true
ENABLE_TEXT_ADAPTER=true
@@ -150,7 +147,7 @@ MCP_TOOL_TIMEOUT_SECONDS=30
ROUTING_MODE=router
# Usage/cost accounting
USAGE_REPOSITORY_PROVIDER=autonomous
USAGE_REPOSITORY_PROVIDER=sqlite
IDENTITY_CONFIG_PATH=./config/identity.yaml
MCP_PARAMETER_MAPPING_PATH=./config/mcp_parameter_mapping.yaml

View File

@@ -4,7 +4,7 @@ import logging
from uuid import uuid4
import time
from fastapi import FastAPI, Request
from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
@@ -52,7 +52,7 @@ summary_memory = create_conversation_summary_memory(settings, message_history=me
sessions = create_session_repository(settings)
checkpoints = create_checkpoint_repository(settings)
cache = create_cache(settings, telemetry=telemetry)
gateway = ChannelGateway()
gateway = ChannelGateway(input_mode=settings.FRAMEWORK_CHANNEL_INPUT_MODE)
analytics = create_analytics_publisher(settings)
observer = TelemetryBackedAgentObserver(telemetry=telemetry)
configure_global_observer({
@@ -70,9 +70,11 @@ logger.info("LLM provider carregado: %s", llm.__class__.__name__)
logger.info("Langfuse habilitado: %s host=%s", telemetry.is_enabled(), settings.LANGFUSE_HOST)
logger.info("Analytics habilitado: %s providers=%s", getattr(settings, "ENABLE_ANALYTICS", False), getattr(settings, "ANALYTICS_PROVIDERS", ""))
logger.info("Agentes disponíveis: %s", [p.agent_id for p in agent_profiles.list_profiles()])
logger.info("Framework channel input mode: %s", gateway.input_mode)
@app.middleware("http")
async def observability_context_middleware(request: Request, call_next):
clear_observability_context()
request_id = request.headers.get("x-request-id") or str(uuid4())
set_observability_context(
request_id=request_id,
@@ -98,6 +100,8 @@ async def observability_context_middleware(request: Request, call_next):
"duration_ms": int((time.time() - started) * 1000),
}, kind="http")
raise
finally:
clear_observability_context()
class GatewayRequest(BaseModel):
@@ -107,6 +111,31 @@ class GatewayRequest(BaseModel):
tenant_id: str | None = None
def _metadata_value(payload: dict, key: str):
metadata = payload.get("metadata")
if isinstance(metadata, dict):
return metadata.get(key)
return None
def _extract_workflow_id(payload: dict) -> str | None:
return (
payload.get("workflow_id")
or payload.get("workflowId")
or _metadata_value(payload, "workflow_id")
or _metadata_value(payload, "workflowId")
)
def _format_root_span_name(template: str | None, values: dict) -> str:
template = template or "agent.gateway_message"
try:
return template.format(**{k: v or "unknown" for k, v in values.items()})
except Exception:
logger.warning("LANGFUSE_ROOT_SPAN_NAME inválido: %s", template)
return "agent.gateway_message"
def _resolve_identity(req: GatewayRequest, msg) -> tuple[AgentIdentity, dict, BusinessContext, list[str]]:
payload = req.payload or {}
context = dict(msg.context or {})
@@ -138,10 +167,15 @@ def _resolve_identity(req: GatewayRequest, msg) -> tuple[AgentIdentity, dict, Bu
async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False) -> dict:
msg = await gateway.normalize(req.channel, req.payload)
try:
msg = await gateway.normalize(req.channel, req.payload)
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
payload = req.payload or {}
identity, normalized_context, business_context, missing_identity_keys = _resolve_identity(req, msg)
agent_session_id = identity.conversation_key()
message_id = (req.payload or {}).get("message_id") or str(uuid4())
message_id = payload.get("message_id") or str(uuid4())
workflow_id = _extract_workflow_id(payload)
set_observability_context(
session_id=agent_session_id,
user_id=msg.user_id,
@@ -149,7 +183,8 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
agent_id=identity.agent_id,
channel=msg.channel,
message_id=message_id,
ura_call_id=(req.payload or {}).get("ura_call_id") or normalized_context.get("ura_call_id") or business_context.interaction_key,
workflow_id=workflow_id,
ura_call_id=payload.get("ura_call_id") or normalized_context.get("ura_call_id") or business_context.interaction_key,
)
stream = sse_hub.stream_for(agent_session_id)
@@ -205,34 +240,56 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
await sse_hub.emit(agent_session_id, "message.received", {"session_id": agent_session_id, "role": "user"}) if emit_sse else None
history = [m.model_dump(mode="json") for m in await memory.list(agent_session_id)]
trace_input = {
cms_input = {
"channel": req.channel,
"tenant_id": req.tenant_id,
"agent_id": req.agent_id,
"payload": payload,
}
trace_context = {
"text": msg.text,
"channel": msg.channel,
"channel_id": msg.channel_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"conversation_key": agent_session_id,
"workflow_id": workflow_id,
"message_id": message_id,
"business_context": business_context.model_dump(),
"identity_missing": missing_identity_keys,
}
root_span_name = _format_root_span_name(
getattr(settings, "LANGFUSE_ROOT_SPAN_NAME", "agent.gateway_message"),
{
"workflow_id": workflow_id,
"channel": msg.channel,
"agent_id": identity.agent_id,
"tenant_id": identity.tenant_id,
},
)
root_tags = ["agent-template", msg.channel, f"agent:{identity.agent_id}", f"tenant:{identity.tenant_id}"]
if workflow_id:
root_tags.append(f"workflow:{workflow_id}")
async with telemetry.span(
"agent.gateway_message",
root_span_name,
session_id=agent_session_id,
user_id=session.user_id,
channel=msg.channel,
input=trace_input,
tags=["agent-template", msg.channel, f"agent:{identity.agent_id}", f"tenant:{identity.tenant_id}"],
):
await telemetry.event("gateway.message.received", trace_input)
await sse_hub.emit(agent_session_id, "workflow.started", trace_input) if emit_sse else None
workflow_id=workflow_id,
input=cms_input,
tags=root_tags,
_root_span=True,
) as root_span:
await telemetry.event("gateway.message.received", trace_context)
await sse_hub.emit(agent_session_id, "workflow.started", trace_context) if emit_sse else None
result = await workflow.ainvoke(
{
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"session_id": agent_session_id,
"conversation_key": agent_session_id,
"workflow_id": workflow_id,
"agent_profile": normalized_context["agent_profile"],
"user_text": msg.text,
"history": history,
@@ -242,6 +299,7 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
"original_session_id": msg.session_id,
"session_id": agent_session_id,
"conversation_key": agent_session_id,
"workflow_id": workflow_id,
"user_id": session.user_id,
"channel": msg.channel,
"message_id": message_id,
@@ -252,66 +310,68 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
}
)
await checkpoints.put(agent_session_id, {"state": result, "message_id": message_id})
await sse_hub.emit(agent_session_id, "workflow.completed", {"session_id": agent_session_id, "route": result.get("route"), "intent": result.get("intent")}) if emit_sse else None
await checkpoints.put(agent_session_id, {"state": result, "message_id": message_id})
await sse_hub.emit(agent_session_id, "workflow.completed", {"session_id": agent_session_id, "route": result.get("route"), "intent": result.get("intent")}) if emit_sse else None
answer = result.get("final_answer") or result.get("answer") or ""
await memory.append(
agent_session_id,
ChatMessage(
role="assistant",
content=answer,
metadata={
answer = result.get("final_answer") or result.get("answer") or ""
await memory.append(
agent_session_id,
ChatMessage(
role="assistant",
content=answer,
metadata={
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"message_id": f"assistant-{message_id}",
"route": result.get("route"),
"intent": result.get("intent"),
"route_decision": result.get("route_decision"),
"judges": result.get("judge_results"),
},
),
)
await telemetry.event(
"gateway.message.responded",
{
"session_id": agent_session_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"message_id": f"assistant-{message_id}",
"route": result.get("route"),
"intent": result.get("intent"),
"answer_chars": len(answer),
},
)
response = ChannelResponse(
channel=msg.channel,
session_id=agent_session_id,
text=answer,
metadata={
"channel_id": msg.channel_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"original_session_id": msg.session_id,
"conversation_key": agent_session_id,
"workflow_id": workflow_id,
"message_id": message_id,
"route": result.get("route"),
"intent": result.get("intent"),
"route_decision": result.get("route_decision"),
"domain": result.get("domain"),
"mcp_tools": result.get("mcp_tools"),
"mcp_results": result.get("mcp_results"),
"business_context": business_context.model_dump(),
"identity_missing": missing_identity_keys,
"judges": result.get("judge_results"),
"guardrails": result.get("guardrail_decisions"),
},
),
)
await telemetry.event(
"gateway.message.responded",
{
"session_id": agent_session_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"route": result.get("route"),
"intent": result.get("intent"),
"answer_chars": len(answer),
},
)
response = ChannelResponse(
channel=msg.channel,
session_id=agent_session_id,
text=answer,
metadata={
"channel_id": msg.channel_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"original_session_id": msg.session_id,
"conversation_key": agent_session_id,
"message_id": message_id,
"route": result.get("route"),
"intent": result.get("intent"),
"route_decision": result.get("route_decision"),
"domain": result.get("domain"),
"mcp_tools": result.get("mcp_tools"),
"mcp_results": result.get("mcp_results"),
"business_context": business_context.model_dump(),
"identity_missing": missing_identity_keys,
"judges": result.get("judge_results"),
"guardrails": result.get("guardrail_decisions"),
},
)
rendered = await gateway.render(response)
await sse_hub.emit(agent_session_id, "message.responded", rendered) if emit_sse else None
await sse_hub.emit(agent_session_id, "flow.end", {"session_id": agent_session_id, "message_id": message_id}) if emit_sse else None
return rendered
)
rendered = await gateway.render(response)
root_span.set_output(rendered)
await sse_hub.emit(agent_session_id, "message.responded", rendered) if emit_sse else None
await sse_hub.emit(agent_session_id, "flow.end", {"session_id": agent_session_id, "message_id": message_id}) if emit_sse else None
return rendered
@app.get("/health")
@@ -331,6 +391,8 @@ async def health():
"usage_repository": settings.USAGE_REPOSITORY_PROVIDER,
"identity_config_path": settings.IDENTITY_CONFIG_PATH,
"mcp_parameter_mapping_path": settings.MCP_PARAMETER_MAPPING_PATH,
"framework_channel_input_mode": settings.FRAMEWORK_CHANNEL_INPUT_MODE,
"legacy_channel_gateway_mode": settings.CHANNEL_GATEWAY_MODE,
}
@@ -354,6 +416,8 @@ async def debug_env():
"AGENTS_CONFIG_PATH": settings.AGENTS_CONFIG_PATH,
"ROUTING_CONFIG_PATH": settings.ROUTING_CONFIG_PATH,
"ROUTING_MODE": settings.ROUTING_MODE,
"FRAMEWORK_CHANNEL_INPUT_MODE": settings.FRAMEWORK_CHANNEL_INPUT_MODE,
"CHANNEL_GATEWAY_MODE": settings.CHANNEL_GATEWAY_MODE,
}

View File

@@ -6,6 +6,7 @@ class AgentState(TypedDict, total=False):
agent_id: str
session_id: str
conversation_key: str
workflow_id: str
agent_profile: dict[str, Any]
user_text: str
sanitized_input: str

View File

@@ -187,6 +187,16 @@ class AgentWorkflow:
input=state.get("user_text"),
):
history_texts = [m.get("content", "") for m in state.get("history", [])]
await self.observer.emit_grl(
"001",
{
"session_id": state.get("conversation_key") or state.get("session_id"),
"tenant_id": state.get("tenant_id"),
"agent_id": state.get("agent_id"),
"phase": "input",
},
component="workflow.input_guardrails.start",
)
sanitized, decisions = await self.guardrails.run_input(
state["user_text"],
{
@@ -199,6 +209,19 @@ class AgentWorkflow:
)
for _decision in decisions:
await self.guardrail_telemetry.evaluated("input", _decision)
await self.observer.emit_grl(
"002" if _decision.allowed else "004",
{
"session_id": state.get("conversation_key") or state.get("session_id"),
"tenant_id": state.get("tenant_id"),
"agent_id": state.get("agent_id"),
"phase": "input",
"rail_code": getattr(_decision, "code", None),
"allowed": bool(_decision.allowed),
"reason": getattr(_decision, "reason", None),
},
component="workflow.input_guardrails.decision",
)
if not _decision.allowed:
await self.guardrail_telemetry.blocked("input", _decision)
await self.telemetry.event(
@@ -210,6 +233,18 @@ class AgentWorkflow:
"decisions": [d.model_dump() for d in decisions],
},
)
await self.observer.emit_grl(
"009",
{
"session_id": state.get("conversation_key") or state.get("session_id"),
"tenant_id": state.get("tenant_id"),
"agent_id": state.get("agent_id"),
"phase": "input",
"blocked": any(not d.allowed for d in decisions),
"decision_count": len(decisions),
},
component="workflow.input_guardrails.final",
)
if any(not d.allowed for d in decisions):
return {
"sanitized_input": sanitized,
@@ -285,41 +320,36 @@ class AgentWorkflow:
}
async def billing_agent(self, state):
async with self.langgraph_telemetry.node("billing_agent", state):
async with self.telemetry.span(
"workflow.agent.billing",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.billing.run(state)
async with self.telemetry.span(
"workflow.agent.billing",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.billing.run(state)
async def product_agent(self, state):
async with self.langgraph_telemetry.node("product_agent", state):
async with self.telemetry.span(
"workflow.agent.product",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.product.run(state)
async with self.telemetry.span(
"workflow.agent.product",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.product.run(state)
async def orders_agent(self, state):
async with self.langgraph_telemetry.node("orders_agent", state):
async with self.telemetry.span(
"workflow.agent.orders",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.orders.run(state)
async with self.telemetry.span(
"workflow.agent.orders",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.orders.run(state)
async def support_agent(self, state):
async with self.langgraph_telemetry.node("support_agent", state):
async with self.telemetry.span(
"workflow.agent.support",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.support.run(state)
async with self.telemetry.span(
"workflow.agent.support",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.support.run(state)
async def supervisor_agent(self, state):
"""Executa um ou mais agentes no modo supervisor e consolida a resposta.
@@ -419,6 +449,21 @@ class AgentWorkflow:
},
)
await self.observer.emit_ic(
"IC.OUTPUT_SUPERVISOR_COMPLETED",
{
"session_id": context["session_id"],
"tenant_id": state.get("tenant_id"),
"agent_id": state.get("agent_id"),
"route": state.get("route"),
"intent": state.get("intent"),
"action": action,
"approved": decision.approved,
"result_count": len(decision.results),
},
component="workflow.output_supervisor",
)
if decision.action in {RailAction.ALLOW, RailAction.SANITIZE, RailAction.OBSERVE}:
final_answer = decision.candidate
elif decision.action == RailAction.HANDOVER:
@@ -457,11 +502,36 @@ class AgentWorkflow:
session_id=state.get("conversation_key") or state.get("session_id"),
input=state.get("answer"),
):
await self.observer.emit_grl(
"001",
{
"session_id": state.get("conversation_key") or state.get("session_id"),
"tenant_id": state.get("tenant_id"),
"agent_id": state.get("agent_id"),
"phase": "output",
"route": state.get("route"),
"intent": state.get("intent"),
},
component="workflow.output_guardrails.start",
)
final, decisions = await self.guardrails.run_output(
state["answer"], state.get("context", {})
)
for _decision in decisions:
await self.guardrail_telemetry.evaluated("output", _decision)
await self.observer.emit_grl(
"002" if _decision.allowed else "004",
{
"session_id": state.get("conversation_key") or state.get("session_id"),
"tenant_id": state.get("tenant_id"),
"agent_id": state.get("agent_id"),
"phase": "output",
"rail_code": getattr(_decision, "code", None),
"allowed": bool(_decision.allowed),
"reason": getattr(_decision, "reason", None),
},
component="workflow.output_guardrails.decision",
)
if not _decision.allowed:
await self.guardrail_telemetry.blocked("output", _decision)
await self.telemetry.event(
@@ -473,6 +543,18 @@ class AgentWorkflow:
"decisions": [d.model_dump() for d in decisions],
},
)
await self.observer.emit_grl(
"009",
{
"session_id": state.get("conversation_key") or state.get("session_id"),
"tenant_id": state.get("tenant_id"),
"agent_id": state.get("agent_id"),
"phase": "output",
"blocked": any(not d.allowed for d in decisions),
"decision_count": len(decisions),
},
component="workflow.output_guardrails.final",
)
return {
"final_answer": final,
"guardrail_decisions": state.get("guardrail_decisions", [])

View File

@@ -0,0 +1,194 @@
from __future__ import annotations
import pytest
from agent_framework.observability.context import clear_observability_context
from agent_framework.observability.telemetry import Telemetry
class Settings:
ENABLE_LANGFUSE = False
ENABLE_OTEL = False
LANGFUSE_TRACE_MODE = "compact"
class FakeObservation:
_next_id = 1
def __init__(self, kwargs):
self.kwargs = kwargs
self.id = f"obs-{FakeObservation._next_id}"
self.trace_id = "trace-123"
FakeObservation._next_id += 1
self.updates = []
self.trace_updates = []
self.trace_io_updates = []
def update(self, **kwargs):
self.updates.append(kwargs)
def update_trace(self, **kwargs):
self.trace_updates.append(kwargs)
def set_trace_io(self, **kwargs):
self.trace_io_updates.append(kwargs)
class FakeContextManager:
def __init__(self, observation):
self.observation = observation
def __enter__(self):
return self.observation
def __exit__(self, exc_type, exc, tb):
return False
class FakePropagationContext:
def __init__(self, owner, kwargs):
self.owner = owner
self.kwargs = kwargs
def __enter__(self):
self.owner.propagations.append(self.kwargs)
def __exit__(self, exc_type, exc, tb):
return False
class FakeLangfuse:
def __init__(self, *, legacy_api: bool = False):
self.observations = []
self.propagations = []
self.flush_count = 0
self.api = FakeApi() if legacy_api else None
def start_as_current_observation(self, **kwargs):
observation = FakeObservation(kwargs)
self.observations.append(observation)
return FakeContextManager(observation)
def propagate_attributes(self, **kwargs):
return FakePropagationContext(self, kwargs)
def flush(self):
self.flush_count += 1
class FakeIngestionResponse:
errors = []
successes = []
class FakeIngestion:
def __init__(self):
self.batches = []
def batch(self, *, batch, metadata=None):
self.batches.append({"batch": batch, "metadata": metadata})
return FakeIngestionResponse()
class FakeApi:
def __init__(self):
self.ingestion = FakeIngestion()
def telemetry_with_fake_langfuse(*, legacy_api: bool = False):
FakeObservation._next_id = 1
telemetry = Telemetry(Settings())
telemetry.enabled = True
telemetry.langfuse = FakeLangfuse(legacy_api=legacy_api)
return telemetry
@pytest.mark.asyncio
async def test_compact_keeps_root_output_and_shows_only_aga_noc_events():
clear_observability_context()
telemetry = telemetry_with_fake_langfuse()
async with telemetry.span("agent.gateway_message", session_id="s1", input={"request": "cms"}, _root_span=True) as span:
await telemetry.event("IC.INTERNAL", {"step": "hidden"}, kind="ic")
await telemetry.event("NOC.001", {"step": "visible"}, kind="noc")
await telemetry.event("AGA.010", {"step": "visible"}, kind="ic")
span.set_output({"answer": "ok"})
names = [obs.kwargs["name"] for obs in telemetry.langfuse.observations]
assert names == ["agent.gateway_message", "NOC.001", "AGA.010"]
root = telemetry.langfuse.observations[0]
assert root.updates[-1]["input"] == {"request": "cms"}
assert root.updates[-1]["output"] == {"answer": "ok"}
assert root.trace_io_updates[-1] == {"input": {"request": "cms"}, "output": {"answer": "ok"}}
assert telemetry.langfuse.observations[1].kwargs.get("trace_context") is None
assert telemetry.langfuse.observations[2].kwargs.get("trace_context") is None
assert telemetry.langfuse.propagations[-1]["trace_name"] == "agent.gateway_message"
aggregated = root.updates[-1]["metadata"]["aggregated_events"]
assert [event["name"] for event in aggregated] == ["IC.INTERNAL", "NOC.001", "AGA.010"]
@pytest.mark.asyncio
async def test_compact_generation_records_io_model_parameters_and_usage_details():
clear_observability_context()
telemetry = telemetry_with_fake_langfuse()
async with telemetry.span("agent.gateway_message", session_id="s1", input={"request": "cms"}, _root_span=True):
async with telemetry.generation_span(
name="llm.test",
model="test-model",
input=[{"role": "user", "content": "ping"}],
metadata={"profile_name": "test"},
model_parameters={"temperature": 0.2, "max_tokens": 100},
) as generation:
generation.set_output("pong")
generation.set_usage({"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2, "cost_usd": 0.01})
generation = telemetry.langfuse.observations[1]
assert generation.kwargs["name"] == "llm.test"
assert generation.kwargs["as_type"] == "generation"
assert generation.kwargs["input"] == [{"role": "user", "content": "ping"}]
assert generation.kwargs["model"] == "test-model"
assert generation.kwargs["model_parameters"] == {"temperature": 0.2, "max_tokens": 100}
assert "usage" not in generation.kwargs
assert "usage_details" not in generation.kwargs
assert generation.updates[-1]["input"] == [{"role": "user", "content": "ping"}]
assert generation.updates[-1]["output"] == "pong"
assert generation.updates[-1]["usage_details"] == {"input": 1, "output": 1}
assert generation.updates[-1]["cost_details"] == {"total": 0.01}
assert generation.kwargs.get("trace_context") is None
@pytest.mark.asyncio
async def test_legacy_io_fallback_updates_same_root_and_generation_observations():
clear_observability_context()
telemetry = telemetry_with_fake_langfuse(legacy_api=True)
async with telemetry.span("agent.gateway_message", session_id="s1", input={"request": "cms"}, _root_span=True) as root:
async with telemetry.generation_span(
name="llm.test",
model="test-model",
input=[{"role": "user", "content": "ping"}],
model_parameters={"temperature": 0.2},
) as generation:
generation.set_output("pong")
generation.set_usage({"prompt_tokens": 2, "completion_tokens": 3, "total_tokens": 5})
root.set_output({"answer": "ok"})
batches = telemetry.langfuse.api.ingestion.batches
assert [event.type for item in batches for event in item["batch"]] == ["generation-update", "span-update"]
generation_event = batches[0]["batch"][0]
assert generation_event.body.id == "obs-2"
assert generation_event.body.trace_id == "trace-123"
assert generation_event.body.input == [{"role": "user", "content": "ping"}]
assert generation_event.body.output == "pong"
assert generation_event.body.usage_details == {"input": 2, "output": 3}
root_event = batches[1]["batch"][0]
assert root_event.body.id == "obs-1"
assert root_event.body.trace_id == "trace-123"
assert root_event.body.input == {"request": "cms"}
assert root_event.body.output == {"answer": "ok"}
assert len(telemetry.langfuse.observations) == 2