Framework Integrations

import thinkhive
from thinkhive.integrations.langchain import ThinkHiveCallbackHandler
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
thinkhive.init(service_name="langchain-app")
 
# Create callback handler
callback = ThinkHiveCallbackHandler(
    run_name="support-chain",
    metadata={"environment": "production"}
)
 
# Build a chain
model = ChatOpenAI(model="gpt-4")
prompt = ChatPromptTemplate.from_template("Answer this question: {question}")
chain = prompt | model | StrOutputParser()
 
# Invoke with tracing
result = chain.invoke(
    {"question": "What is ThinkHive?"},
    config={"callbacks": [callback]}
)

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_core.runnables import RunnablePassthrough
 
thinkhive.init(service_name="rag-app")
callback = ThinkHiveCallbackHandler(run_name="rag-query")
 
# Build RAG chain
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_texts(documents, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
 
model = ChatOpenAI(model="gpt-4")
prompt = ChatPromptTemplate.from_template(
    "Answer based on this context:\n{context}\n\nQuestion: {question}"
)
 
chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)
 
# Full RAG pipeline traced: retrieval → prompt → LLM → output
result = chain.invoke("How does ThinkHive work?", config={"callbacks": [callback]})

from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain_community.tools import DuckDuckGoSearchRun
 
tools = [DuckDuckGoSearchRun()]
 
agent = create_openai_functions_agent(
    llm=ChatOpenAI(model="gpt-4"),
    tools=tools,
    prompt=agent_prompt,
)
 
executor = AgentExecutor(agent=agent, tools=tools)
 
# Each agent step is traced: Decision → Tool Call → Decision → Final Answer
result = executor.invoke(
    {"input": "What is the latest news about AI?"},
    config={"callbacks": [callback]}
)

import thinkhive
from crewai import Agent, Task, Crew
 
thinkhive.init(service_name="crewai-app")
 
# Define agents
researcher = Agent(
    role="Research Analyst",
    goal="Find relevant information",
    backstory="Expert researcher with deep analytical skills",
    llm="gpt-4",
)
 
writer = Agent(
    role="Content Writer",
    goal="Write clear, accurate content",
    backstory="Technical writer specializing in AI documentation",
    llm="gpt-4",
)
 
# Define tasks
research_task = Task(
    description="Research the topic: {topic}",
    expected_output="A summary of key findings",
    agent=researcher,
)
 
writing_task = Task(
    description="Write a report based on the research findings",
    expected_output="A polished report",
    agent=writer,
)
 
# Create and run crew with tracing
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
)
 
# Wrap in ThinkHive trace context
tracer = thinkhive.get_tracer()
with tracer.start_as_current_span("crew-execution") as span:
    span.set_attribute("crew.agents", 2)
    span.set_attribute("crew.tasks", 2)
    result = crew.kickoff(inputs={"topic": "AI observability"})

import thinkhive
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
 
thinkhive.init(service_name="llamaindex-app")
 
# Load documents and create index
documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents)
 
# Query with tracing
tracer = thinkhive.get_tracer()
with tracer.start_as_current_span("llamaindex-query") as span:
    query_engine = index.as_query_engine()
    response = query_engine.query("What is ThinkHive?")
 
    span.set_attribute("query", "What is ThinkHive?")
    span.set_attribute("response.length", len(str(response)))

from thinkhive.integrations.llamaindex import ThinkHiveCallbackManager
 
# Set up callback manager
callback_manager = ThinkHiveCallbackManager(run_name="index-query")
 
# Attach to service context
index = VectorStoreIndex.from_documents(
    documents,
    callback_manager=callback_manager,
)
 
query_engine = index.as_query_engine(callback_manager=callback_manager)
response = query_engine.query("Explain AI observability")
# Retrieval, reranking, and LLM calls all traced

import thinkhive
from openai import OpenAI
 
thinkhive.init(service_name="openai-app")
client = OpenAI()
 
@thinkhive.trace_llm(model_name="gpt-4", provider="openai")
def generate_response(prompt: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}],
    )
    return response.choices[0].message.content
 
# Automatically traced with model, tokens, latency
answer = generate_response("What is ThinkHive?")

@thinkhive.trace_llm(model_name="gpt-4", provider="openai")
def stream_response(prompt: str) -> str:
    stream = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )
 
    full_response = ""
    for chunk in stream:
        content = chunk.choices[0].delta.content or ""
        full_response += content
        print(content, end="", flush=True)
 
    return full_response

import thinkhive
import anthropic
 
thinkhive.init(service_name="anthropic-app")
client = anthropic.Anthropic()
 
@thinkhive.trace_llm(model_name="claude-3-5-sonnet-20241022", provider="anthropic")
def claude_chat(message: str) -> str:
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        messages=[{"role": "user", "content": message}],
    )
    return response.content[0].text
 
answer = claude_chat("Explain AI observability")

import thinkhive
 
thinkhive.init(service_name="custom-app")
 
class MyRAGPipeline:
    def __init__(self, retriever, llm):
        self.retriever = retriever
        self.llm = llm
 
    @thinkhive.trace_retrieval()
    def retrieve(self, query: str) -> list:
        return self.retriever.search(query, top_k=5)
 
    @thinkhive.trace_llm(model_name="custom-model", provider="custom")
    def generate(self, query: str, context: str) -> str:
        return self.llm.generate(query=query, context=context)
 
    def query(self, question: str) -> str:
        tracer = thinkhive.get_tracer()
        with tracer.start_as_current_span("rag-pipeline") as span:
            span.set_attribute("question", question)
 
            docs = self.retrieve(question)
            context = "\n".join(docs)
            answer = self.generate(question, context)
 
            return answer

from opentelemetry.context import attach, detach, set_value
 
# Disable tracing for a block
token = attach(set_value("suppress_instrumentation", True))
try:
    # This call will not be traced
    response = client.chat.completions.create(...)
finally:
    detach(token)