Pypestream Weave Onboarding Guide

os.environ["WANDB_BASE_URL"] = "https://api.wandb.ai"
os.environ["WANDB_API_KEY"]  = "your-wandb-api-key"

pip install weave

wandb login --host <YOUR W&B HOST URL> <YOUR API TOKEN>

wandb.login(host=os.getenv("WANDB_BASE_URL"), key=os.getenv("WANDB_API_KEY"))

import weave
weave.init('<entity-name>/<project-name>')
# this ensures the project is created in the relevant team. 
# If you have WANDB_ENTITY and WANDB_PROJECT set as env vars then you won't need to specify these.
# You may need to create the team. 

import weave
from openai import OpenAI

client = OpenAI()

# Weave will track the inputs, outputs and code of this function
@weave.op()
def extract_dinos(sentence: str) -> dict:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": """In JSON format extract a list of `dinosaurs`, with their `name`,
their `common_name`, and whether its `diet` is a herbivore or carnivore"""
            },
            {
                "role": "user",
                "content": sentence
            }
            ],
            response_format={ "type": "json_object" }
        )
    return response.choices[0].message.content


# Initialize the weave project
weave.init('<team-name>/jurassic-park')

sentence = """I watched as a Tyrannosaurus rex (T. rex) chased after a Triceratops (Trike), \
both carnivore and herbivore locked in an ancient dance. Meanwhile, a gentle giant \
Brachiosaurus (Brachi) calmly munched on treetops, blissfully unaware of the chaos below."""

result = extract_dinos(sentence)
print(result)

from openai import OpenAI

import weave

weave.init(PROJECT)

client = OpenAI()
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {
            "role": "system",
            "content": "You are a grammar checker, correct the following user input.",
        },
        {"role": "user", "content": "That was so easy, it was a piece of pie!"},
    ],
    temperature=0,
)
generation = response.choices[0].message.content
print(generation)

import weave
from weave import Evaluation

# Define any custom scoring function
@weave.op()
def exact_match(expected: str, output: dict) -> dict:
    # Here is where you'd define the logic to score the model output
    return {"match": expected == output}


# Score your examples using scoring functions
evaluation = Evaluation(
    dataset=dataset,  # can be a list of dictionaries or a weave.Dataset object
    scorers=[exact_match],  # can be a list of scoring functions
)

# Start tracking the evaluation
weave.init(PROJECT)
# Run the evaluation
print(asyncio.run(evaluation.evaluate(corrector)))

# if you're in a Jupyter Notebook, run:
# await evaluation.evaluate(corrector)

import weave
weave.init('intro-example')

system_prompt = weave.StringPrompt("You are a pirate")
weave.publish(system_prompt, name="pirate_prompt")

from openai import OpenAI
client = OpenAI()

response = client.chat.completions.create(
  model="gpt-4o",
  messages=[
    {
      "role": "system",
      "content": system_prompt.format()
    },
    {
      "role": "user",
      "content": "Explain general relativity in one paragraph."
    }
  ],
)


import weave
weave.init('intro-example')

system_prompt = weave.StringPrompt("Talk like a pirate. I need to know I'm listening to a pirate.")
weave.publish(system_prompt, name="pirate_prompt")

from openai import OpenAI
client = OpenAI()

response = client.chat.completions.create(
  model="gpt-4o",
  messages=[
    {
      "role": "system",
      "content": system_prompt.format()
    },
    {
      "role": "user",
      "content": "Explain general relativity in one paragraph."
    }
  ],
)

import weave
weave.init('intro-example')

prompt = weave.MessagesPrompt([
    {
        "role": "system",
        "content": "You are a stegosaurus, but don't be too obvious about it."
    },
    {
        "role": "user",
        "content": "What's good to eat around here?"
    }
])
weave.publish(prompt, name="dino_prompt")

from openai import OpenAI
client = OpenAI()

response = client.chat.completions.create(
  model="gpt-4o",
  messages=prompt.format(),
)

import weave
weave.init('intro-example')

prompt = weave.StringPrompt("Solve the equation {equation}")
weave.publish(prompt, name="calculator_prompt")

from openai import OpenAI
client = OpenAI()

response = client.chat.completions.create(
  model="gpt-4o",
  messages=[
    {
      "role": "user",
      "content": prompt.format(equation="1 + 1 = ?")
    }
  ],
)

from openai import OpenAI

import weave

weave.init(PROJECT)


class OpenAIGrammarCorrector(weave.Model):
    # Properties are entirely user-defined
    openai_model_name: str
    system_message: str

    @weave.op()
    def predict(self, user_input):
        client = OpenAI()
        response = client.chat.completions.create(
            model=self.openai_model_name,
            messages=[
                {"role": "system", "content": self.system_message},
                {"role": "user", "content": user_input},
            ],
            temperature=0,
        )
        return response.choices[0].message.content


corrector = OpenAIGrammarCorrector(
    openai_model_name="gpt-4o-mini",
    system_message="You are a grammar checker, correct the following user input.",
)

result = corrector.predict("That was so easy, it was a piece of pie!")
print(result)

import weave

weave.init(PROJECT)

# Note: this url is available from the UI after publishing the object!
ref_url = f"weave:///{ref.entity}/{PROJECT}/object/{ref.name}:{ref.digest}"
fetched_collector = weave.ref(ref_url).get()

# Notice: this object was loaded from remote location!
result = fetched_collector.predict("That was so easy, it was a piece of pie!")

print(result)

import weave
client = weave.init('intro-example')

# Get all feedback in a project
all_feedback = client.get_feedback()

# Fetch a specific feedback object by id.
# The API returns a collection, which is expected to contain at most one item.
one_feedback = client.get_feedback("<feedback_uuid>")[0]

# Find all feedback objects with a specific reaction. You can specify offset and limit.
thumbs_up = client.get_feedback(reaction="👍", limit=10)

# After retrieval, view the details of individual feedback objects.
for f in client.get_feedback():
    print(f.id)
    print(f.created_at)
    print(f.feedback_type)
    print(f.payload)

import weave
client = weave.init('intro-example')

call = client.get_call("<call_uuid>")

# Adding an emoji reaction
call.feedback.add_reaction("👍")

# Adding a note
call.feedback.add_note("this is a note")

# Adding custom key/value pairs.
# The first argument is a user-defined "type" string.
# Feedback must be JSON serializable and less than 1 KB when serialized.
call.feedback.add("correctness", { "value": 5 })

import weave
from weave.flow.annotation_spec import AnnotationSpec

client = weave.init("feedback-example")

spec1 = AnnotationSpec(
  name="Temperature",
  description="The perceived temperature of the llm call",
  field_schema={
    "type": "number",
    "minimum": -1,
    "maximum": 1,
  }
)
spec2 = AnnotationSpec(
  name="Tone",
  description="The tone of the llm response",
  field_schema={
    "type": "string",
    "enum": ["Aggressive", "Neutral", "Polite", "N/A"],
  },
)
weave.publish(spec1, "temperature-scorer")
weave.publish(spec2, "tone-scorer")

import weave

client = weave.init("feedback-example")

call = client.get_call("<call_id>")
annotation_spec = weave.ref("<annotation_spec_ref_uri>")

call.feedback.add(
  feedback_type="wandb.annotation." + annotation_spec.name,
  payload={"value": 1},
  annotation_ref=annotation_spec.uri(),
)