DSPy

1
from respan import Respan
2
from respan_instrumentation_dspy import DSPyInstrumentor
3
4
respan = Respan(
5
    app_name="dspy-api",
6
    instrumentations=[DSPyInstrumentor()],
7
    customer_identifier="user_123",
8
    metadata={"service": "dspy-api", "version": "1.0.0"},
9
)

1
import dspy
2
from respan import Respan
3
from respan_instrumentation_dspy import DSPyInstrumentor
4
5
respan = Respan(instrumentations=[DSPyInstrumentor()])
6
7
class QA(dspy.Signature):
8
    question: str = dspy.InputField()
9
    answer: str = dspy.OutputField()
10
11
predict = dspy.ChainOfThought(QA)
12
13
def handle_request(user_id: str, question: str):
14
    with respan.propagate_attributes(
15
        customer_identifier=user_id,
16
        thread_identifier="conv_abc_123",
17
        trace_group_identifier="dspy-support-request",
18
        metadata={"plan": "pro"},
19
    ):
20
        result = predict(question=question)
21
        print(result.answer)

1
import json
2
3
import dspy
4
from opentelemetry.semconv_ai import SpanAttributes
5
from respan import Respan
6
from respan_instrumentation_dspy import DSPyInstrumentor
7
8
respan = Respan(
9
    app_name="dspy-support",
10
    instrumentations=[DSPyInstrumentor()],
11
)
12
13
class QA(dspy.Signature):
14
    question: str = dspy.InputField()
15
    answer: str = dspy.OutputField()
16
17
answerer = dspy.ChainOfThought(QA)
18
19
def support_workflow(question: str):
20
    client = respan.telemetry.get_client()
21
22
    with respan.propagate_attributes(
23
        trace_group_identifier="support-workflow-001",
24
        metadata={"workflow": "support_qa"},
25
    ):
26
        with client.start_span("dspy_support.workflow", kind="workflow") as span:
27
            span.set_attribute(
28
                SpanAttributes.TRACELOOP_ENTITY_INPUT,
29
                json.dumps({"question": question}),
30
            )
31
            prediction = answerer(question=question)
32
            span.set_attribute(
33
                SpanAttributes.TRACELOOP_ENTITY_OUTPUT,
34
                json.dumps({"answer": prediction.answer}),
35
            )
36
            return prediction.answer
37
38
print(support_workflow("Why is tracing useful for DSPy programs?"))
39
respan.flush()

1
import dspy
2
3
def lookup_order_status(order_id: str) -> str:
4
    statuses = {
5
        "ord-1001": "ord-1001 is shipped and arriving tomorrow.",
6
        "ord-1002": "ord-1002 is waiting for carrier pickup.",
7
    }
8
    return statuses.get(order_id, f"No status found for {order_id}.")
9
10
tool = dspy.Tool(
11
    lookup_order_status,
12
    name="lookup_order_status",
13
    desc="Look up the shipping status for an order id.",
14
)
15
16
status = tool(order_id="ord-1001")
17
print(status)

1
import dspy
2
3
class CityQuestion(dspy.Signature):
4
    """Answer the user's city question with one sentence."""
5
6
    question: str = dspy.InputField()
7
    answer: str = dspy.OutputField()
8
9
def lookup_city_fact(city: str) -> str:
10
    facts = {
11
        "tokyo": "Tokyo has one of the world's busiest rail networks.",
12
        "paris": "Paris is known for the Louvre and the Eiffel Tower.",
13
    }
14
    return facts.get(city.lower(), f"No stored fact for {city}.")
15
16
agent = dspy.ReAct(CityQuestion, tools=[lookup_city_fact], max_iters=3)
17
prediction = agent(
18
    question=(
19
        "Use lookup_city_fact for Tokyo, then answer with the fact in "
20
        "one sentence."
21
    )
22
)
23
print(prediction.answer)