Multi-Agent Systems
Monitor complex multi-agent collaborations and workflows with comprehensive observability.
Overview
Multi-agent systems involve multiple AI agents working together to solve complex problems. AgenticAnts provides complete visibility into:
- Agent Communication - Track messages between agents
- Workflow Orchestration - Monitor complex workflows
- Collaboration Patterns - Understand agent interactions
- Performance Bottlenecks - Identify slow agents
- Cost Attribution - Track costs per agent
Architecture Patterns
1. Sequential Workflow
Agents work in a pipeline, each handing its result to the next. Model each pipeline stage as its own agent observation (always passing agent_name=), and nest the LLM call inside each agent as a generation so cost and latency roll up per agent.
from ants_platform import AntsPlatform
ants = AntsPlatform(
public_key=os.getenv("ANTS_PLATFORM_PUBLIC_KEY"),
secret_key=os.getenv("ANTS_PLATFORM_SECRET_KEY"),
host="https://api.agenticants.ai", # always set host explicitly
)
def sequential_workflow(input_text: str):
with ants.start_as_current_observation(
name="sequential-workflow", as_type="agent", agent_name="supervisor"
):
ants.update_current_trace(
input=input_text,
metadata={"workflow_type": "sequential", "total_agents": 3},
)
# Agent 1: Text Preprocessor
with ants.start_as_current_observation(
name="text-preprocessor", as_type="agent", agent_name="preprocessor"
) as agent:
with ants.start_as_current_observation(
name="clean", as_type="generation", model="gpt-4o-mini", input=input_text
) as gen:
preprocessed = preprocess_agent.process(input_text)
gen.update(
output=preprocessed,
usage_details={"input": 180, "output": 96, "total": 276},
)
agent.update(output=preprocessed)
# Agent 2: Content Analyzer
with ants.start_as_current_observation(
name="content-analyzer", as_type="agent", agent_name="analyzer"
) as agent:
with ants.start_as_current_observation(
name="analyze", as_type="generation", model="gpt-4o", input=preprocessed
) as gen:
analysis = analyzer_agent.process(preprocessed)
gen.update(
output=analysis.summary,
usage_details={"input": 96, "output": 142, "total": 238},
)
agent.update(output={"analysis_type": analysis.type})
# Agent 3: Response Generator
with ants.start_as_current_observation(
name="response-generator", as_type="agent", agent_name="generator"
) as agent:
with ants.start_as_current_observation(
name="generate", as_type="generation", model="gpt-4o", input=analysis.summary
) as gen:
response = generator_agent.process(analysis)
gen.update(
output=response,
usage_details={"input": 238, "output": 210, "total": 448},
)
agent.update(output=response)
ants.update_current_trace(
output=response,
metadata={"total_agents": 3, "success": True},
)
ants.flush()
return response
In Java (17+), open the supervisor with Ants.observe(..., ObservationType.AGENT, ...) and nest each pipeline stage in its own Ants.observe call — nested Ants.observe calls auto-nest under the active span. Unlike Python, Java does not need an agent name; for ObservationType.AGENT the SDK derives it from the observation name.
// Configure once at startup (registers the global OpenTelemetry tracer):
// AntsPlatformOtel.configure(AntsPlatformOtel.options()
// .publicKey(System.getenv("ANTS_PLATFORM_PUBLIC_KEY")) // "pk-ap-..."
// .secretKey(System.getenv("ANTS_PLATFORM_SECRET_KEY"))); // "sk-ap-..."
// // host defaults to https://api.agenticants.ai
String sequentialWorkflow(String inputText) {
String response = Ants.observe("sequential-workflow", ObservationType.AGENT, supervisor -> {
supervisor.updateTrace(TraceAttributes.builder()
.input(inputText)
.metadata(Map.of("workflow_type", "sequential", "total_agents", 3))
.build());
// Agent 1: Text Preprocessor
String preprocessed = Ants.observe("text-preprocessor", ObservationType.AGENT, agent -> {
AntsGeneration gen = Ants.startGeneration("clean")
.model("gpt-4o-mini").input(inputText);
String out = preprocessAgent.process(inputText);
gen.output(out).usage(180, 96).end();
agent.output(out);
return out;
});
// Agent 2: Content Analyzer
Analysis analysis = Ants.observe("content-analyzer", ObservationType.AGENT, agent -> {
AntsGeneration gen = Ants.startGeneration("analyze")
.model("gpt-4o").input(preprocessed);
Analysis a = analyzerAgent.process(preprocessed);
gen.output(a.summary).usage(96, 142).end();
agent.output(Map.of("analysis_type", a.type));
return a;
});
// Agent 3: Response Generator
String result = Ants.observe("response-generator", ObservationType.AGENT, agent -> {
AntsGeneration gen = Ants.startGeneration("generate")
.model("gpt-4o").input(analysis.summary);
String r = generatorAgent.process(analysis);
gen.output(r).usage(238, 210).end();
agent.output(r);
return r;
});
supervisor.updateTrace(TraceAttributes.builder()
.output(result)
.metadata(Map.of("total_agents", 3, "success", true))
.build());
return result;
});
AntsPlatformOtel.flush();
return response;
}
2. Parallel Processing
Multiple agents work simultaneously on different tasks under a single supervisor. Use startActiveObservation with asType: 'agent' to open the supervisor context, then start one agent observation per fan-out worker. Each worker carries its own generation (with model + usageDetails) so the trace tree shows three sibling agents under the supervisor.
startActiveObservation,
startObservation,
updateActiveTrace,
getActiveTraceId,
} from 'ants-platform'
// Tracing must be initialized once at startup (see instrumentation.ts):
// import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node'
// import { AntsPlatformSpanProcessor } from '@antsplatform/otel'
// import { setAntsPlatformTracerProvider } from 'ants-platform'
// const provider = new NodeTracerProvider({
// spanProcessors: [
// new AntsPlatformSpanProcessor({
// publicKey: process.env.ANTS_PLATFORM_PUBLIC_KEY,
// secretKey: process.env.ANTS_PLATFORM_SECRET_KEY,
// }),
// ],
// })
// provider.register()
// setAntsPlatformTracerProvider(provider)
async function parallelWorkflow(inputData: { text: string }) {
return startActiveObservation(
'parallel-supervisor',
async () => {
updateActiveTrace({
name: 'parallel-workflow',
input: inputData,
metadata: { workflowType: 'parallel', totalAgents: 3 },
})
// Run worker agents in parallel; each nests its own generation.
const [sentiment, entities, summary] = await Promise.all([
startActiveObservation(
'sentiment-analyzer',
async () => {
const gen = startObservation(
'classify-sentiment',
{ model: 'gpt-4o-mini', input: inputData.text, modelParameters: { temperature: 0 } },
{ asType: 'generation' },
)
const result = await sentimentAgent.analyze(inputData.text)
gen.update({ output: result, usageDetails: { input: 120, output: 8, total: 128 } })
gen.end()
return result
},
{ asType: 'agent' },
),
startActiveObservation(
'entity-extractor',
async () => {
const gen = startObservation(
'extract-entities',
{ model: 'gpt-4o-mini', input: inputData.text },
{ asType: 'generation' },
)
const result = await entityAgent.extract(inputData.text)
gen.update({
output: result.entities,
usageDetails: { input: 120, output: 64, total: 184 },
})
gen.end()
return result
},
{ asType: 'agent' },
),
startActiveObservation(
'text-summarizer',
async () => {
const gen = startObservation(
'summarize',
{ model: 'gpt-4o', input: inputData.text },
{ asType: 'generation' },
)
const result = await summarizerAgent.summarize(inputData.text)
gen.update({
output: result.text,
usageDetails: { input: 120, output: 88, total: 208 },
})
gen.end()
return result
},
{ asType: 'agent' },
),
])
const combinedResult = {
sentiment: sentiment.score,
entities: entities.entities,
summary: summary.text,
}
updateActiveTrace({ output: combinedResult, metadata: { totalAgents: 3, success: true } })
return { traceId: getActiveTraceId(), combinedResult }
},
{ asType: 'agent' },
)
}
In Java the trace tree is built from OpenTelemetry context, and OTel context does not auto-propagate across threads. If you fan these workers out onto an executor, the child spans will not nest under the supervisor unless you propagate the context manually (e.g. wrap each task with io.opentelemetry.context.Context.current().wrap(runnable)). The simplest correct pattern is to open the three worker agents as siblings under one supervisor on the calling thread — each carries its own generation, and the tree still shows three agents under the supervisor. Run the actual model work concurrently and only open the spans where you have the results, so nesting stays correct.
WorkflowResult parallelWorkflow(InputData input) {
return Ants.observe("parallel-supervisor", ObservationType.AGENT, supervisor -> {
supervisor.updateTrace(TraceAttributes.builder()
.name("parallel-workflow")
.input(input.text)
.metadata(Map.of("workflowType", "parallel", "totalAgents", 3))
.build());
// Run the model work concurrently off the active span...
CompletableFuture<SentimentResult> fSentiment =
CompletableFuture.supplyAsync(() -> sentimentAgent.analyze(input.text));
CompletableFuture<EntityResult> fEntities =
CompletableFuture.supplyAsync(() -> entityAgent.extract(input.text));
CompletableFuture<SummaryResult> fSummary =
CompletableFuture.supplyAsync(() -> summarizerAgent.summarize(input.text));
SentimentResult sentiment = fSentiment.join();
EntityResult entities = fEntities.join();
SummaryResult summary = fSummary.join();
// ...then record each worker as a sibling agent on the active (supervisor) thread,
// so all three nest under the supervisor without cross-thread context propagation.
Ants.startActiveObservation("sentiment-analyzer", ObservationType.AGENT, agent -> {
AntsGeneration gen = Ants.startGeneration("classify-sentiment")
.model("gpt-4o-mini").modelParameters(Map.of("temperature", 0))
.input(input.text);
gen.output(sentiment).usage(120, 8).end();
});
Ants.startActiveObservation("entity-extractor", ObservationType.AGENT, agent -> {
AntsGeneration gen = Ants.startGeneration("extract-entities")
.model("gpt-4o-mini").input(input.text);
gen.output(entities.entities).usage(120, 64).end();
});
Ants.startActiveObservation("text-summarizer", ObservationType.AGENT, agent -> {
AntsGeneration gen = Ants.startGeneration("summarize")
.model("gpt-4o").input(input.text);
gen.output(summary.text).usage(120, 88).end();
});
Map<String, Object> combined = Map.of(
"sentiment", sentiment.score,
"entities", entities.entities,
"summary", summary.text);
supervisor.updateTrace(TraceAttributes.builder()
.output(combined)
.metadata(Map.of("totalAgents", 3, "success", true))
.build());
return new WorkflowResult(combined);
});
}
3. Hierarchical Coordination
A supervisor agent plans the task, hands each step off to a specialized sub-agent, and synthesizes the results. The supervisor is the top-level agent observation; every sub-agent it dispatches is a nested agent observation, and each sub-agent runs its own generation, tool, or retriever children. The resulting trace tree is supervisor -> -> . In Python every agent observation must pass agent_name=.
from ants_platform import get_client
ants = get_client() # configured once at startup with host="https://api.agenticants.ai"
class HierarchicalMultiAgent:
def __init__(self):
self.coordinator = CoordinatorAgent()
self.workers = {
"researcher": ResearcherAgent(),
"writer": WriterAgent(),
"reviewer": ReviewerAgent(),
}
def process_task(self, task: str):
# Top-level supervisor agent
with ants.start_as_current_observation(
name="hierarchical-workflow", as_type="agent", agent_name="supervisor"
):
ants.update_current_trace(
input=task,
metadata={
"workflow_type": "hierarchical",
"coordinator": "supervisor",
"workers": list(self.workers.keys()),
},
)
# Supervisor plans the task (its own generation)
with ants.start_as_current_observation(
name="plan-task", as_type="generation", model="gpt-4o", input=task
) as gen:
plan = self.coordinator.analyze_task(task)
gen.update(
output=[s.name for s in plan.steps],
usage_details={"input": 140, "output": 60, "total": 200},
)
# Hand off each step to the assigned sub-agent
results = {}
for step in plan.steps:
worker_name = step.assigned_worker # "researcher" | "writer" | "reviewer"
with ants.start_as_current_observation(
name=f"handoff-{worker_name}",
as_type="agent",
agent_name=worker_name,
input={"step": step.name, "from": "supervisor"},
) as sub_agent:
if worker_name == "researcher":
# Sub-agent uses a retriever tool
with ants.start_as_current_observation(
name="search-corpus", as_type="retriever", input=step.query
) as retr:
docs = self.workers[worker_name].search(step.query)
retr.update(output={"hits": len(docs)})
result = self.workers[worker_name].execute(step, docs)
elif worker_name == "writer":
# Sub-agent runs a generation
with ants.start_as_current_observation(
name="draft", as_type="generation", model="gpt-4o", input=step.brief
) as gen:
result = self.workers[worker_name].execute(step)
gen.update(
output=result,
usage_details={"input": 320, "output": 480, "total": 800},
)
else: # reviewer uses a tool call
with ants.start_as_current_observation(
name="lint-content", as_type="tool", input={"draft": step.draft}
) as tool:
result = self.workers[worker_name].execute(step)
tool.update(output={"issues": result.issues})
results[worker_name] = result
sub_agent.update(output={"result_size": len(str(result))})
# Supervisor synthesizes the sub-agent outputs
with ants.start_as_current_observation(
name="synthesize", as_type="generation", model="gpt-4o", input=results
) as gen:
final_result = self.coordinator.synthesize(results)
gen.update(
output=final_result,
usage_details={"input": 520, "output": 300, "total": 820},
)
ants.update_current_trace(
output=final_result,
metadata={"total_steps": len(plan.steps), "success": True},
)
ants.flush()
return final_result
In Java the supervisor is the top-level AGENT observation; each dispatched sub-agent is a nested Ants.observe(..., ObservationType.AGENT, ...), and each runs its own generation, tool, or retriever child via startObservation / startGeneration. The tree is supervisor -> -> .
String processTask(String task) {
String finalResult = Ants.observe("hierarchical-workflow", ObservationType.AGENT, supervisor -> {
supervisor.updateTrace(TraceAttributes.builder()
.input(task)
.metadata(Map.of(
"workflow_type", "hierarchical",
"coordinator", "supervisor",
"workers", List.of("researcher", "writer", "reviewer")))
.build());
// Supervisor plans the task (its own generation)
AntsGeneration planGen = Ants.startGeneration("plan-task").model("gpt-4o").input(task);
Plan plan = coordinator.analyzeTask(task);
planGen.output(plan.stepNames()).usage(140, 60).end();
// Hand off each step to the assigned sub-agent
Map<String, Object> results = new LinkedHashMap<>();
for (Step step : plan.steps) {
String workerName = step.assignedWorker; // "researcher" | "writer" | "reviewer"
Object result = Ants.observe("handoff-" + workerName, ObservationType.AGENT, subAgent -> {
subAgent.input(Map.of("step", step.name, "from", "supervisor"));
Object r;
if (workerName.equals("researcher")) {
// Sub-agent uses a retriever
AntsObservation retr = Ants.startObservation("search-corpus", ObservationType.RETRIEVER);
retr.input(step.query);
List<Doc> docs = workers.get(workerName).search(step.query);
retr.output(Map.of("hits", docs.size())).end();
r = workers.get(workerName).execute(step, docs);
} else if (workerName.equals("writer")) {
// Sub-agent runs a generation
AntsGeneration gen = Ants.startGeneration("draft").model("gpt-4o").input(step.brief);
r = workers.get(workerName).execute(step);
gen.output(r).usage(320, 480).end();
} else { // reviewer uses a tool call
AntsObservation tool = Ants.startObservation("lint-content", ObservationType.TOOL);
tool.input(Map.of("draft", step.draft));
ReviewResult rr = workers.get(workerName).execute(step);
tool.output(Map.of("issues", rr.issues)).end();
r = rr;
}
subAgent.output(Map.of("result_size", String.valueOf(r).length()));
return r;
});
results.put(workerName, result);
}
// Supervisor synthesizes the sub-agent outputs
AntsGeneration synth = Ants.startGeneration("synthesize").model("gpt-4o").input(results);
String result = coordinator.synthesize(results);
synth.output(result).usage(520, 300).end();
supervisor.updateTrace(TraceAttributes.builder()
.output(result)
.metadata(Map.of("total_steps", plan.steps.size(), "success", true))
.build());
return result;
});
AntsPlatformOtel.flush();
return finalResult;
}
Agent Communication Monitoring
Message Passing Tracking
A handoff from one agent to another is modeled as a sender agent observation whose child is the receiver agent observation. The nesting captures who delegated to whom.
from ants_platform import get_client
ants = get_client() # configured once at startup with host="https://api.agenticants.ai"
class CommunicatingAgents:
def __init__(self):
self.agents = {
"agent_a": AgentA(),
"agent_b": AgentB(),
"agent_c": AgentC(),
}
def send_message(self, from_agent: str, to_agent: str, message: dict):
# Sender agent context
with ants.start_as_current_observation(
name=f"{from_agent}-send", as_type="agent", agent_name=from_agent
):
ants.update_current_trace(
input=message,
metadata={
"from_agent": from_agent,
"to_agent": to_agent,
"message_type": message.get("type"),
"communication_id": message.get("id"),
},
)
# Handoff: receiver agent nests under the sender
with ants.start_as_current_observation(
name=f"{to_agent}-receive",
as_type="agent",
agent_name=to_agent,
input=message,
) as receiver:
response = self.agents[to_agent].receive_message(message)
receiver.update(output=response)
ants.update_current_trace(output=response, metadata={"success": True})
return response
In Java the handoff is the same shape: open the sender as an AGENT observation, and nest the receiver AGENT observation inside it so the tree captures who delegated to whom.
Object sendMessage(String fromAgent, String toAgent, Map<String, Object> message) {
return Ants.observe(fromAgent + "-send", ObservationType.AGENT, sender -> {
sender.updateTrace(TraceAttributes.builder()
.input(message)
.metadata(Map.of(
"from_agent", fromAgent,
"to_agent", toAgent,
"message_type", message.get("type"),
"communication_id", message.get("id")))
.build());
// Handoff: receiver agent nests under the sender
Object response = Ants.observe(toAgent + "-receive", ObservationType.AGENT, receiver -> {
receiver.input(message);
Object resp = agents.get(toAgent).receiveMessage(message);
receiver.output(resp);
return resp;
});
sender.updateTrace(TraceAttributes.builder()
.output(response).metadata(Map.of("success", true)).build());
return response;
});
}
Conversation Flow Tracking
Each turn in a conversation is an agent observation that wraps the participant's generation, so the trace tree shows one agent per message under the conversation supervisor.
class ConversationFlow {
private agents: Map<string, any> = new Map()
async processConversation(conversation: Conversation) {
return startActiveObservation(
'conversation-flow',
async () => {
updateActiveTrace({
name: 'conversation-flow',
input: conversation,
metadata: {
conversationId: conversation.id,
participantCount: conversation.participants.length,
messageCount: conversation.messages.length,
},
})
for (const message of conversation.messages) {
const agent = this.agents.get(message.agentId)
await startActiveObservation(
`turn-${message.agentId}`,
async () => {
const gen = startObservation(
`respond-${message.id}`,
{ model: 'gpt-4o-mini', input: message.content },
{ asType: 'generation' },
)
const response = await agent.processMessage(message)
gen.update({
output: response,
usageDetails: { input: 64, output: 96, total: 160 },
})
gen.end()
},
{ asType: 'agent' },
)
}
updateActiveTrace({ output: conversation.responses, metadata: { success: true } })
},
{ asType: 'agent' },
)
}
}
In Java, wrap the conversation in an AGENT observation and open one nested AGENT per turn, each carrying the participant's generation.
void processConversation(Conversation conversation) {
Ants.startActiveObservation("conversation-flow", ObservationType.AGENT, flow -> {
flow.updateTrace(TraceAttributes.builder()
.name("conversation-flow")
.input(conversation)
.metadata(Map.of(
"conversationId", conversation.id,
"participantCount", conversation.participants.size(),
"messageCount", conversation.messages.size()))
.build());
for (Message message : conversation.messages) {
Agent agent = agents.get(message.agentId);
Ants.startActiveObservation("turn-" + message.agentId, ObservationType.AGENT, turn -> {
AntsGeneration gen = Ants.startGeneration("respond-" + message.id)
.model("gpt-4o-mini").input(message.content);
Object response = agent.processMessage(message);
gen.output(response).usage(64, 96).end();
});
}
flow.updateTrace(TraceAttributes.builder()
.output(conversation.responses).metadata(Map.of("success", true)).build());
});
}
Workflow Orchestration
Complex Workflow with Decision Points
A router agent decides which downstream agent should handle the request, then hands off to the chosen path. The selected execution agent nests under the router as its own agent observation.
from ants_platform import get_client
ants = get_client() # configured once at startup with host="https://api.agenticants.ai"
class ComplexWorkflow:
def __init__(self):
self.decision_agent = DecisionAgent()
self.execution_agents = {
"path_a": PathAAgent(),
"path_b": PathBAgent(),
"path_c": PathCAgent(),
}
def execute_workflow(self, input_data: dict):
with ants.start_as_current_observation(
name="complex-workflow", as_type="agent", agent_name="router"
):
ants.update_current_trace(
input=input_data,
metadata={"workflow_type": "complex", "has_decision_points": True},
)
# Step 1: Router decides which path to take (its own generation)
with ants.start_as_current_observation(
name="route", as_type="generation", model="gpt-4o-mini", input=input_data
) as gen:
decision = self.decision_agent.decide(input_data)
gen.update(
output={"path": decision.path, "confidence": decision.confidence},
usage_details={"input": 90, "output": 12, "total": 102},
)
# Step 2: Hand off to the chosen execution agent
path = decision.path # "path_a" | "path_b" | "path_c"
with ants.start_as_current_observation(
name=f"handoff-{path}",
as_type="agent",
agent_name=f"{path}-agent",
input={"from": "router", "path": path},
) as sub_agent:
result = self.execution_agents[path].execute(input_data)
sub_agent.update(output=result)
ants.update_current_trace(
output=result,
metadata={"decision_path": decision.path, "success": True},
)
ants.flush()
return result
In Java the router is the top-level AGENT observation with its routing generation; the chosen execution agent nests under it as its own AGENT observation.
Object executeWorkflow(Map<String, Object> inputData) {
Object result = Ants.observe("complex-workflow", ObservationType.AGENT, router -> {
router.updateTrace(TraceAttributes.builder()
.input(inputData)
.metadata(Map.of("workflow_type", "complex", "has_decision_points", true))
.build());
// Step 1: Router decides which path to take (its own generation)
AntsGeneration gen = Ants.startGeneration("route").model("gpt-4o-mini").input(inputData);
Decision decision = decisionAgent.decide(inputData);
gen.output(Map.of("path", decision.path, "confidence", decision.confidence))
.usage(90, 12).end();
// Step 2: Hand off to the chosen execution agent
String path = decision.path; // "path_a" | "path_b" | "path_c"
Object r = Ants.observe("handoff-" + path, ObservationType.AGENT, subAgent -> {
subAgent.input(Map.of("from", "router", "path", path));
Object out = executionAgents.get(path).execute(inputData);
subAgent.output(out);
return out;
});
router.updateTrace(TraceAttributes.builder()
.output(r).metadata(Map.of("decision_path", decision.path, "success", true)).build());
return r;
});
AntsPlatformOtel.flush();
return result;
}
Per-agent performance metrics (average latency, success rate, request counts, and cost) are not exposed through the SDK. View and compare agent performance across your traces in the AgenticAnts dashboard at https://app.agenticants.ai, where you can filter by agent name and time range, and set thresholds for latency, success rate, and cost.
Workflow Bottleneck Analysis
To find the slowest and most error-prone steps in a workflow, open the trace and span explorer in the dashboard at https://app.agenticants.ai. Sorting spans by duration and error rate surfaces bottlenecks without any code. Make sure each agent step is wrapped in its own span (as shown above) so it appears as a distinct, measurable step.
Cost Attribution
Track Costs Per Agent
Cost and token-usage rollups per agent are a dashboard feature, not an SDK query. Visit https://app.agenticants.ai to see cost distribution across agents over a chosen period. To make costs attributable, tag each agent's span with a consistent agent value in its metadata (as shown in the patterns above); the dashboard then groups cost by that attribute.
Optimize Expensive Agents
Optimization recommendations for high-cost agents (model right-sizing, prompt reduction, caching) are surfaced in the LLM Optimizer view of the dashboard at https://app.agenticants.ai. There is no SDK call for retrieving these recommendations.
Best Practices
1. Design for Observability
from ants_platform import AntsPlatform
ants = AntsPlatform(
public_key=os.getenv("ANTS_PLATFORM_PUBLIC_KEY"),
secret_key=os.getenv("ANTS_PLATFORM_SECRET_KEY"),
host="https://api.agenticants.ai", # always set host explicitly
)
class ObservableMultiAgent:
def execute_workflow(self, input_data):
# Top-level supervisor agent (agent_name is required for as_type="agent")
with ants.start_as_current_observation(
name="observable-workflow", as_type="agent", agent_name="supervisor"
):
ants.update_current_trace(
input=input_data,
metadata={"workflow_version": "1.0", "environment": "production"},
)
# One agent observation per sub-agent so cost/latency roll up per agent
with ants.start_as_current_observation(
name="worker", as_type="agent", agent_name="worker", input=input_data
) as agent:
with ants.start_as_current_observation(
name="reason", as_type="generation", model="gpt-4o", input=input_data
) as gen:
result = self.process_with_agents(input_data)
gen.update(
output=result,
usage_details={"input": 200, "output": 150, "total": 350},
)
agent.update(output={"result_size": len(str(result))})
ants.update_current_trace(output=result)
ants.flush()
return result
The Java equivalent — top-level supervisor agent, one nested agent per sub-agent so cost and latency roll up per agent:
Object executeWorkflow(Object inputData) {
Object result = Ants.observe("observable-workflow", ObservationType.AGENT, supervisor -> {
supervisor.updateTrace(TraceAttributes.builder()
.input(inputData)
.metadata(Map.of("workflow_version", "1.0", "environment", "production"))
.build());
// One agent observation per sub-agent so cost/latency roll up per agent
Object out = Ants.observe("worker", ObservationType.AGENT, agent -> {
agent.input(inputData);
AntsGeneration gen = Ants.startGeneration("reason").model("gpt-4o").input(inputData);
Object r = processWithAgents(inputData);
gen.output(r).usage(200, 150).end();
agent.output(Map.of("result_size", String.valueOf(r).length()));
return r;
});
supervisor.updateTrace(TraceAttributes.builder().output(out).build());
return out;
});
AntsPlatformOtel.flush();
return result;
}
2. Implement Circuit Breakers
class ResilientMultiAgent {
private circuitBreakers: Map<string, CircuitBreaker> =
new Map()
async executeWithCircuitBreaker(agentName: string, operation: () => Promise<any>) {
const breaker =
this.circuitBreakers.get(agentName)
if (breaker && breaker.isOpen()) {
throw new Error(
`Circuit breaker open for ${agentName}`)
}
try {
const result =
await operation()
breaker?.recordSuccess()
return result
}
catch (error) {
breaker?.recordFailure()
throw error
}
}
}
3. Monitor Agent Health
Agent health status, uptime, and last-check timestamps are surfaced in the AgenticAnts dashboard at https://app.agenticants.ai. Configure alerts there to be notified when an agent becomes unhealthy. There is no SDK health-check API; emit a span per agent invocation (as shown above) so the dashboard can compute availability from your traces.
Troubleshooting
Common Issues
Issue: Agents not communicating properly
Inspect failed agent-communication traces in the dashboard at https://app.agenticants.ai. Filter by trace name and an error status over your chosen time range to see the failing communications, including the from_agent / to_agent metadata and the input message you attached with update_current_trace. There is no SDK trace-query API; ingestion is one-way.
Issue: Workflow taking too long
Use the trace explorer in the dashboard at https://app.agenticants.ai to find slow workflows. Filter by duration (for example, longer than 30 seconds) over the last 24 hours, then open a trace to see its span waterfall and identify the slowest spans. Because each agent step is wrapped in its own span (as shown in the patterns above), bottlenecks are immediately visible without any querying code.
Next Steps
Example Projects
Congratulations! You now have comprehensive monitoring for your multi-agent systems with complete visibility into agent interactions, performance, and costs.