Integrating Signals with Cost Tracking

1
from paid import Paid
2
from paid.tracing import paid_autoinstrument, initialize_tracing
3
from openai import OpenAI
4
5
# Initialize Paid SDK
6
client = Paid(token="PAID_API_KEY")
7
initialize_tracing()
8
9
paid_autoinstrument()  # instruments all available: anthropic, gemini, openai, openai-agents, bedrock, langchain
10
11
# Now all OpenAI calls will be automatically traced
12
openai_client = OpenAI(api_key="<OPENAI_API_KEY>")
13
14
@paid_tracing("your_external_customer_id", "your_external_agent_id")
15
def chat_with_gpt():
16
    response = openai_client.chat.completions.create(
17
        model="gpt-4",
18
        messages=[{"role": "user", "content": "Hello!"}]
19
    )
20
    return response
21
22
chat_with_gpt()  # Costs are automatically tracked!

1
from paid.tracing import paid_autoinstrument
2
3
# Instrument only Anthropic and OpenAI
4
paid_autoinstrument(libraries=["anthropic", "openai"])

1
from paid.tracing import paid_tracing, signal, initialize_tracing
2
from paid.tracing.wrappers import PaidOpenAI
3
from openai import OpenAI
4
5
initialize_tracing()
6
7
# Wrap your AI provider client
8
openai_client = PaidOpenAI(OpenAI(api_key="<OPENAI_API_KEY>"))
9
10
@paid_tracing("customer_123", "agent_123")
11
def process_event(event):
12
    """
13
    Process an agent event - wrapped in tracing context
14
    """
15
    # Step 1: Make AI calls (automatically traced)
16
    response = openai_client.chat.completions.create(
17
        model="gpt-4",
18
        messages=[
19
            {"role": "system", "content": "You are an AI SDR assistant"},
20
            {"role": "user", "content": event.content}
21
        ]
22
    )
23
24
    # Step 2: Your business logic
25
    send_response(response.choices[0].message.content)
26
27
    # Step 3: Emit signal with cost tracing enabled
28
    signal(
29
        event_name="email_sent",
30
        data={"response_length": len(response.choices[0].message.content)},
31
        enable_cost_tracing=True  # Link costs to this signal
32
    )
33
34
# Call your event processor
35
process_event(incoming_event)

1
from paid.tracing import paid_tracing, signal, paid_autoinstrument, initialize_tracing
2
from openai import OpenAI
3
4
initialize_tracing()
5
6
# Instrument OpenAI before creating OpenAI client
7
paid_autoinstrument()
8
openai_client = OpenAI(api_key="<OPENAI_API_KEY>")
9
# or alternatively can use openai wrapper without paid autoinstrumentation
10
# from paid.tracing.wrappers import PaidOpenAI
11
# openai_client = PaidOpenAI(OpenAI(api_key="<OPENAI_API_KEY>"))
12
13
@paid_tracing("acme_account_123", "ai_sdr_agent")
14
def handle_email_reply(email):
15
    """
16
    Handle incoming email reply from prospect
17
    """
18
    # AI call #1: Analyze sentiment
19
    sentiment_response = openai_client.chat.completions.create(
20
        model="gpt-4",
21
        messages=[
22
            {"role": "system", "content": "Analyze email sentiment"},
23
            {"role": "user", "content": email.body}
24
        ]
25
    )
26
27
    sentiment = sentiment_response.choices[0].message.content
28
29
    # AI call #2: Generate response
30
    reply_response = openai_client.chat.completions.create(
31
        model="gpt-4",
32
        messages=[
33
            {"role": "system", "content": f"Generate reply. Sentiment: {sentiment}"},
34
            {"role": "user", "content": email.body}
35
        ]
36
    )
37
38
    reply_text = reply_response.choices[0].message.content
39
40
    # Business logic: Send the email
41
    send_email(
42
        to=email.sender,
43
        subject=f"Re: {email.subject}",
44
        body=reply_text
45
    )
46
47
    # Emit signal - costs from BOTH AI calls will be attributed here
48
    signal(
49
        event_name="email_reply_processed",
50
        data={
51
            "sentiment": sentiment,
52
            "reply_length": len(reply_text)
53
        },
54
        enable_cost_tracing=True
55
    )
56
57
    # Optional: Emit additional signal WITHOUT cost tracing
58
    # (to track the event without duplicating costs)
59
    signal(
60
        event_name="email_sentiment_analyzed",
61
        data={"sentiment": sentiment},
62
        enable_cost_tracing=False  # Don't link costs to this one
63
    )
64
65
# Process incoming email
66
handle_email_reply(incoming_email)

1
from paid.tracing import paid_tracing, signal, initialize_tracing
2
from paid.tracing.wrappers import PaidOpenAI
3
from openai import OpenAI
4
5
initialize_tracing()
6
7
# Initialize once
8
openai_client = PaidOpenAI(OpenAI(api_key="<OPENAI_API_KEY>"))
9
10
@paid_tracing("customer_123", "agent_123")
11
def process_agent_event(event):
12
    """Process a single agent event with cost tracking"""
13
14
    # Make AI calls as needed
15
    if event.type == "email_received":
16
        response = openai_client.chat.completions.create(
17
            model="gpt-4",
18
            messages=[{"role": "user", "content": event.content}]
19
        )
20
        handle_email_response(response)
21
22
        # Emit signal with cost tracing
23
        signal(
24
            event_name="email_processed",
25
            enable_cost_tracing=True
26
        )
27
28
    elif event.type == "call_completed":
29
        # Different AI calls for call processing
30
        summary = openai_client.chat.completions.create(
31
            model="gpt-4",
32
            messages=[{"role": "user", "content": f"Summarize: {event.transcript}"}]
33
        )
34
35
        signal(
36
            event_name="call_summarized",
37
            enable_cost_tracing=True
38
        )
39
40
# Event loop
41
while True:
42
    event = get_next_event()  # Your event queue
43
    if event:
44
        process_agent_event(event)  # Each call creates a new trace

1
from paid.tracing import paid_tracing, signal, initialize_tracing
2
from paid.tracing.wrappers import PaidOpenAI
3
from openai import OpenAI
4
5
initialize_tracing()
6
7
openai_client = PaidOpenAI(OpenAI(api_key="<OPENAI_API_KEY>"))
8
9
@paid_tracing(
10
    "customer_123",
11
    "agent_123",
12
    metadata={
13
        "campaign_id": "campaign_456",
14
        "environment": "production",
15
        "user_tier": "enterprise"
16
    }
17
)
18
def process_event(event):
19
    """Process event with custom metadata"""
20
    response = openai_client.chat.completions.create(
21
        model="gpt-4",
22
        messages=[{"role": "user", "content": event.content}]
23
    )
24
25
    signal("event_processed", enable_cost_tracing=True)
26
    return response
27
28
process_event(incoming_event)

1
from paid.tracing import paid_tracing, signal, generate_tracing_token, initialize_tracing
2
from paid.tracing.wrappers.openai import PaidOpenAI
3
from openai import OpenAI
4
5
initialize_tracing()
6
7
openai_client = PaidOpenAI(OpenAI(api_key="<OPENAI_API_KEY>"))
8
9
# Generate a tracing token
10
tracing_token = generate_tracing_token()
11
print(f"Tracing token: {tracing_token}")
12
13
# Store this token somewhere accessible to other processes
14
# (database, message queue, Redis, etc.)
15
save_to_redis("workflow_123", tracing_token)
16
17
@paid_tracing("customer_123", tracing_token=tracing_token, external_agent_id="agent_123")
18
def initial_processing(email):
19
    """First part of workflow - email analysis"""
20
    analysis = openai_client.chat.completions.create(
21
        model="gpt-4",
22
        messages=[{"role": "user", "content": f"Analyze: {email}"}]
23
    )
24
25
    # Emit signal WITHOUT cost tracing (costs will be linked later)
26
    signal(
27
        event_name="email_analyzed",
28
        enable_cost_tracing=False
29
    )
30
31
    return analysis
32
33
# Process the email
34
result = initial_processing(incoming_email)
35
36
# Send result to another process for response generation
37
enqueue_task("generate_response", {"analysis": result, "workflow_id": "workflow_123"})

1
from paid.tracing import paid_tracing, signal
2
3
# ✅ Good: One trace = One outcome
4
@paid_tracing("customer_123", "agent_123")
5
def book_meeting(prospect):
6
    # Multiple AI calls OK
7
    analyze_calendar(prospect)
8
    generate_invite(prospect)
9
    send_confirmation(prospect)
10
11
    # Single signal with cost tracing
12
    signal("meeting_booked", enable_cost_tracing=True)

1
# ❌ Bad: Multiple cost-enabled signals in one trace
2
@paid_tracing("customer_123", "agent_123")
3
def process_workflow(prospect):
4
    analyze_calendar(prospect)
5
    signal("calendar_analyzed", enable_cost_tracing=True)  # Costs counted
6
7
    generate_invite(prospect)
8
    signal("invite_generated", enable_cost_tracing=True)  # Costs counted AGAIN!

1
from paid.tracing import paid_tracing, signal
2
3
# ✅ Good: Everything in one trace
4
@paid_tracing("customer_123", "agent_123")
5
def process_event(event):
6
    ai_call_1()
7
    ai_call_2()
8
    signal("event_processed", enable_cost_tracing=True)
9
10
# ❌ Bad: AI calls outside trace
11
def process_event(event):
12
    ai_call_1()  # Cost not tracked!
13
14
    @paid_tracing("customer_123", "agent_123")
15
    def emit_signal():
16
        signal("event_processed", enable_cost_tracing=True)
17
18
    emit_signal()