Framework-agnostic observability SDK for multi-agent workflows
Project description
ZENTAXA Python SDK
Framework-agnostic observability SDK for multi-agent workflows.
Live Platform: https://zentaxaapp.azurewebsites.net
Features
- Universal Telemetry: Unified API for all major frameworks
- 5 Framework Integrations: LangChain, LangGraph, CrewAI, AutoGen, LlamaIndex
- Real-time Tracking: Agent runs, steps, LLM calls, tools
- Multi-Step Visualization: Track and visualize complex agent pipelines
- Easy Integration: Add 2 lines of code to your existing agents
- Async Support: Non-blocking telemetry capture
- Error Handling: Automatic retry logic with exponential backoff
Installation
pip install zentaxa
Framework-specific installation
# LangChain / LangGraph
pip install zentaxa[langchain]
# CrewAI
pip install zentaxa[crewai]
# AutoGen
pip install zentaxa[autogen]
# LlamaIndex
pip install zentaxa[llamaindex]
# All frameworks
pip install zentaxa[all]
Quick Start
Basic Usage
from zentaxa import ZentaxaClient
# Default: connects to https://zentaxaapp.azurewebsites.net
client = ZentaxaClient()
# Start an agent run
response = client.trace(
run_id=None,
agent_id="my-agent",
framework="custom",
event_type="agent_start",
metadata={"input": "Your query here"}
)
run_id = response["run_id"]
# Log steps
client.log(run_id, "info", "Processing step 1")
# End the run with results
client.trace(
run_id=run_id,
agent_id="my-agent",
framework="custom",
event_type="agent_end",
metadata={
"input": "Your query",
"output": "Agent response",
"steps": [
{"step_number": 1, "step_name": "Analyze", "input": "...", "output": "...", "latency_ms": 1500, "status": "completed"}
],
"latency_ms": 1500
}
)
LangChain
from zentaxa import ZentaxaClient
from zentaxa.integrations.langchain import ZentaxaCallbackHandler
from langchain_openai import ChatOpenAI
# Initialize client (default connects to live platform)
client = ZentaxaClient()
# Create callback handler
handler = ZentaxaCallbackHandler(client=client, agent_id="my-agent")
# Use with LangChain
llm = ChatOpenAI(callbacks=[handler])
result = llm.invoke("What is quantum computing?")
LangGraph
from zentaxa.integrations.langgraph import LangGraphObserver
from langgraph.graph import StateGraph
observer = LangGraphObserver(client=client, agent_id="my-graph")
graph = StateGraph(state_schema)
graph.add_node("research", research_node, callbacks=[observer])
graph.add_node("analyze", analyze_node, callbacks=[observer])
compiled = graph.compile()
result = compiled.invoke({"query": "quantum computing"})
CrewAI
from zentaxa.integrations.crewai import CrewAIObserver
from crewai import Crew
observer = CrewAIObserver(client=client, agent_id="research-crew")
# Context manager automatically tracks execution
with observer:
result = crew.kickoff()
AutoGen
from zentaxa.integrations.autogen import AutoGenTracer
from autogen import UserProxyAgent, AssistantAgent
tracer = AutoGenTracer(client=client, agent_id="autogen-conversation")
with tracer:
user_proxy.initiate_chat(assistant, message="Research quantum computing")
LlamaIndex
from zentaxa.integrations.llamaindex import LlamaIndexObserver
from llama_index.core.agent import ReActAgent
from llama_index.core.callbacks import CallbackManager
observer = LlamaIndexObserver(client=client, agent_id="llama-agent")
callback_manager = CallbackManager([observer])
agent = ReActAgent.from_tools(
tools,
llm=llm,
callback_manager=callback_manager
)
response = agent.chat("Research quantum computing")
Configuration
Client Options
client = ZentaxaClient(
base_url="https://zentaxaapp.azurewebsites.net", # Default (live platform)
timeout=10.0, # Request timeout (seconds)
max_retries=3, # Retry attempts
async_mode=False # Enable async operations
)
Async Mode
import asyncio
async def main():
async with ZentaxaClient(async_mode=True) as client:
await client.trace_async(
run_id=None,
agent_id="async-agent",
framework="langchain",
event_type="agent_start"
)
asyncio.run(main())
Manual Telemetry
Trace Events
# Start agent run
response = client.trace(
run_id=None,
agent_id="research-agent",
framework="langchain",
event_type="agent_start",
metadata={"goal": "Research quantum computing"}
)
run_id = response["run_id"]
# End agent run
client.trace(
run_id=run_id,
agent_id="research-agent",
framework="langchain",
event_type="agent_end",
metadata={"output": "Research complete"}
)
Log Events
client.log(
run_id=run_id,
message="Executing search tool",
level="info",
context={"tool_name": "TavilySearch", "input": "quantum computing"}
)
Metric Events
client.metric(
run_id=run_id,
metric_name="llm_latency_ms",
value=2340.5,
tags={"provider": "openai", "model": "gpt-4"}
)
Agent Events
client.agent_event(
run_id=run_id,
event_type="step_complete",
step_number=1,
step_name="Plan Research",
action_type="plan",
input_data="Research quantum computing",
output_data="1. Search papers\n2. Analyze trends",
latency_ms=150,
cost_usd=0.0001
)
API Reference
ZentaxaClient
trace(run_id, agent_id, framework, event_type, metadata)- Send trace eventlog(run_id, message, level, context)- Send log eventmetric(run_id, metric_name, value, tags)- Send metric eventagent_event(run_id, event_type, ...)- Send agent eventbulk(events)- Send multiple events in batch
Callback Handlers
ZentaxaCallbackHandler- LangChain/LangGraph callbackLangGraphObserver- LangGraph-specific observerCrewAIObserver- CrewAI wrapperAutoGenTracer- AutoGen conversation trackerLlamaIndexObserver- LlamaIndex event handler
Development
# Clone repository
git clone https://github.com/zentaxa/zentaxa-sdk
# Install dependencies
cd zentaxa-sdk/sdk/python
pip install -e ".[dev]"
# Run tests
pytest
# Format code
black zentaxa/
License
MIT License - see LICENSE file for details.
Support
- Live Platform: https://zentaxaapp.azurewebsites.net
- Dashboard: https://zentaxaapp.azurewebsites.net/dashboard
- Pipeline Explorer: https://zentaxaapp.azurewebsites.net/pipeline
- Issues: https://github.com/CoderInWaves/Startup-MVP/issues
- Email: support@zentaxa.me
Version
1.0.1
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
zentaxa-1.1.2.tar.gz
(23.6 kB
view details)
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
zentaxa-1.1.2-py3-none-any.whl
(26.4 kB
view details)
File details
Details for the file zentaxa-1.1.2.tar.gz.
File metadata
- Download URL: zentaxa-1.1.2.tar.gz
- Upload date:
- Size: 23.6 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.10.12
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
f6847ddcf4aaa70a5a02be363ab61d26908d5d7443418067df82d522cf8a4820
|
|
| MD5 |
cfab3de753fa5349b8bbb2772d0c9da0
|
|
| BLAKE2b-256 |
acd066cd212fce0dcdaa4eadd7305771508768fd9e0f75babe06f9a5f3ded277
|
File details
Details for the file zentaxa-1.1.2-py3-none-any.whl.
File metadata
- Download URL: zentaxa-1.1.2-py3-none-any.whl
- Upload date:
- Size: 26.4 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.10.12
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
1ef278305e458dcc44b2d84554354b118e79b736798a0269a66b04f83606fe76
|
|
| MD5 |
19f595ff7aa2b1c0a032cc2567075324
|
|
| BLAKE2b-256 |
e78fb59696edada0ff8fdb198a3c50e4f1deb5cf6da3e16354f2c05bf6bedf32
|