Tracing

Traces are automatically logged when you setup Agiflow at the top of your application. By default, these traces are limited to backend application and are not synchronized with frontend tracking.

Libraries with Automatic Tracing

• Anthropic
• Chromadb
• Cohere
• CrewAI
• GROQ
• Langchain
• Langgraph
• Llamaindex
• Openai
• Pinecone
• Qdrant
• Weaviate

Trace association

Backend only

If you haven't integrate with the fontend sdk, you can still associate Open Telemetry trace with user and session using below method:

from agiflow import Agiflow
 
Agiflow.set_association_properties({
  "user_id": "<USER_ID>", # Optional
  "session_id": "<SESSION_ID>", # Optional
  "task_name": "<TASK_NAME>", # Optional, to label feedback task
});

Backend and frontend

If you have setup frontend tracing for Web, your backend should have access to x-agiflow-trace-id inside Http Headers.

Simply use our header to associate frontend tracing with Open Telemetry tracing as below:

from agiflow import Agiflow
from agiflow.opentelemetry import extract_association_properties_from_http_headers
 
Agiflow.set_association_properties(extract_association_properties_from_http_headers(request.headers))

Here is a quick explaination of set_association_properties:

• This helper enhance trace context by adding association properties metadata to the traces.
• With manual tracing on frontend, this will add action_id to Trace context.
• With automatic tracing on frontend, this will add action_id, task_id, and session_id to Trace context.

Trace annotation and grouping

You might want to log additional information which is important to your AI workflow or your tools are not supported by Agiflow yet. In these case, use manual tracing to add those information. These decorators support following arguments:

• name: span label
• method_name: method of class to be decorated
• description: add extra comment to make it easier for other people to review workflow and provide feedback
• prompt_settings: associate llm calls with specific version of prompt
• input_serializer: formating input to make it easier for end user to read
• output_serializer: formating output to make it easier for end user to read
• context_parser: restore trace context from distributed messages

Workflow

Trace workflow with unique name and extra information using below method:

from agiflow.opentelemetry import workflow
 
@workflow(name="<WORKFLOW_NAME>", method_name="bar")
class Foo:
    def bar(self):
        ...

Task

Trace task with unique name and extra information using below method:

from agiflow.opentelemetry import task
 
@task(name="<TASK_NAME>", method_name="bar")
class Foo:
    def bar(self):
        ...

Agent

Trace agent with unique name and extra information using below method:

from agiflow.opentelemetry import agent
 
@agent(name="<AGENT_NAME>", method_name="bar")
class Foo:
    def bar(self):
        ...

Tool

Trace tool with unique name and extra information using below method:

from agiflow.opentelemetry import tool
 
@tool(name="<TOOL_NAME>", method_name="bar")
class Foo:
    def bar(self):
        ...

Distributed tracing

If you are using event driven architecture, there is extra work to enable trace flow throughout the workflow.

On producer side, we'll need to extract current trace context and pass it to message queue system as below:

from agiflow.opentelemetry import get_carrier_from_trace_context
 
...
carrier = get_carrier_from_trace_context()
# pass carrier information to message queue
kaffaClient.produce({
...
"otlp_carrier": carrier,
})

On consumer side, we'll need to get carrier information from message and restore context from carrier information:

from agiflow.opentelemetry import get_trace_context_from_carrier, get_tracer
 
...
carrier = message.get("otpl_carrier")
ctx = get_trace_context_from_carrier(carrier)
with get_tracer() as tracer:
    with tracer.start_as_current_span('job', ctx):
        ...

The children span use the same context and parent span so you don't need to pass context around. That's it, now the trace from consumer use the same context as producer.

If you use http for your microservice architecture, OpenTelemetry will automatically pass carrier for you via traceparent headers and restore the context.

Span Update

If you are using event-driven architecture, your parent span may not have the output on completion. This would make it difficult for reviewers to understand the context of the workflow to provide feedback.

With Agiflow, you can update the span outside of the span context. To be able to do so, first associate span_id with your unique identifier (database row id, etc...) with this method:

agiflow.associate_trace(
    id, # messageId, etc... your unique id that will be linked to trace_id
    span_id # messageId, etc... your unique id that will be linked to span_id
);

Then, you can update the span using your database id with:

agiflow.update_span(
    id, # messageId, etc... your unique id that is linked to span_id
    {
    "output": "...'
    }
);