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.1.tar.gz
(23.3 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.1-py3-none-any.whl
(26.0 kB
view details)
File details
Details for the file zentaxa-1.1.1.tar.gz.
File metadata
- Download URL: zentaxa-1.1.1.tar.gz
- Upload date:
- Size: 23.3 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.10.12
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
432007b13403b0ea336c76d52f4e52208829fc615227cfd0cd3034ba9c412ee9
|
|
| MD5 |
bbd9d8f428b2ea1c1dea27d44008ef83
|
|
| BLAKE2b-256 |
7a0b94b88240bbd23446c73bd7f473a943c1fb21810fa9b5330887d3a9be7fbb
|
File details
Details for the file zentaxa-1.1.1-py3-none-any.whl.
File metadata
- Download URL: zentaxa-1.1.1-py3-none-any.whl
- Upload date:
- Size: 26.0 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 |
1ceb9e58ac4fe0149031d2f209765b2ed4df2f52044b5549bfcdc9896477602f
|
|
| MD5 |
d3fc5e934445305436a9cd9e32187ce1
|
|
| BLAKE2b-256 |
3f170a4f1ec0a30357f20bba7810640723431072339faea7dabc10b25b2ae55e
|