Multi-Step Workflows
Learn how to implement Proxy in real-world scenarios, from simple single API calls to complex multi-step workflows with automatic trace and span generation.Understanding Traces and Spans
- Trace: A complete request journey through your application
- Span: Individual operations within a trace (e.g., each AI API call)
- Automatic Generation: Proxy creates these automatically without manual instrumentation
Headers Reference
Proxy heavily relies on headers to mutate traces and spans created in Adaline. Refer to the Headers Reference for more details beyond the examples below.Example 1: Simple RAG Pipeline
A Retrieval-Augmented Generation (RAG) system that combines embedding generation, vector search, and chat completion in a single trace.import os
import uuid
from openai import OpenAI
class RAGPipeline:
def __init__(self):
self.client = OpenAI(
api_key=os.getenv("OPENAI_API_KEY"),
base_url="https://gateway.adaline.ai/v1/openai/"
)
self.base_headers = {
"adaline-api-key": os.getenv("ADALINE_API_KEY"),
"adaline-project-id": os.getenv("ADALINE_PROJECT_ID"),
"adaline-prompt-id": os.getenv("ADALINE_PROMPT_ID")
}
def query(self, user_question: str, user_id: str) -> str:
# Generate unique trace ID for this RAG workflow
trace_id = str(uuid.uuid4())
# Step 1: Generate embedding for user question
query_embedding = self._generate_embedding(
user_question, trace_id, user_id, "query-embedding"
)
# Step 2: Simulate vector search
relevant_docs = self._vector_search(
query_embedding, trace_id, user_id, "vector-search"
)
# Step 3: Generate final response using retrieved context
response = self._generate_response(
user_question, relevant_docs, trace_id, user_id, "response-generation"
)
return response
def _generate_embedding(self, text: str, trace_id: str, user_id: str, span_name: str) -> list:
headers = self.base_headers.copy()
headers.update({
"adaline-trace-status": "pending",
"adaline-trace-reference-id": trace_id,
"adaline-trace-name": "rag-pipeline",
"adaline-span-name": span_name,
"adaline-trace-tags": f'[{"operation": "create", "tag": "production-v1.3"}]',
"adaline-trace-attributes": f'[{{"operation": "create", "key": "user_id", "value": "{user_id}"}}]',
"adaline-span-tags": f'["production-v1.3"]'
})
response = self.client.embeddings.create(
model="text-embedding-3-small",
input=text,
extra_headers=headers
)
return response.data[0].embedding
def _vector_search(self, embedding: list, trace_id: str, user_id: str, span_name: str) -> str:
# Simulate vector search - in real implementation, query your vector database
headers = self.base_headers.copy()
headers.update({
"adaline-trace-status": "pending",
"adaline-trace-reference-id": trace_id,
"adaline-trace-name": "rag-pipeline",
"adaline-span-name": span_name,
"adaline-trace-attributes": f'[{{"operation": "create", "key": "user_id", "value": "{user_id}"}}]',
"adaline-span-tags": f'["production-v1.3"]'
})
# Mock relevant documents (replace with actual vector search)
return "Relevant context: AI models are trained on large datasets and use neural networks..."
def _generate_response(self, question: str, context: str, trace_id: str, user_id: str, span_name: str) -> str:
headers = self.base_headers.copy()
headers.update({
"adaline-trace-status": "success",
"adaline-trace-reference-id": trace_id,
"adaline-trace-name": "rag-pipeline",
"adaline-span-name": span_name,
"adaline-trace-attributes": f'[{{"operation": "create", "key": "user_id", "value": "{user_id}"}}]',
"adaline-span-tags": f'["production-v1.3"]',
"adaline-span-variables": f'{{"user_id": {{"modality": "text", "value": "{user_id}"}}}}'
})
response = self.client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": f"Answer the question using this context: {context}"},
{"role": "user", "content": question}
],
extra_headers=headers
)
return response.choices[0].message.content
# Usage
rag = RAGPipeline()
answer = rag.query("How do neural networks learn?", user_id="user123")
print(answer)
Example 2: Multi-Step Content Generation
A content generation workflow that uses multiple embedding calls and chat completions with span attributes and variables tracking.import os
import uuid
from openai import OpenAI
from anthropic import Anthropic
class ContentGenerator:
def __init__(self):
self.openai_client = OpenAI(
api_key=os.getenv("OPENAI_API_KEY"),
base_url="https://gateway.adaline.ai/v1/openai/"
)
self.anthropic_client = Anthropic(
api_key=os.getenv("ANTHROPIC_API_KEY"),
base_url="https://gateway.adaline.ai/v1/anthropic/"
)
self.base_headers = {
"adaline-api-key": os.getenv("ADALINE_API_KEY"),
"adaline-project-id": os.getenv("ADALINE_PROJECT_ID"),
"adaline-prompt-id": os.getenv("ADALINE_PROMPT_ID")
}
def generate_article(self, topic: str, target_audience: str, user_id: str) -> dict:
trace_id = str(uuid.uuid4())
workflow_data = {
"topic": topic,
"target_audience": target_audience,
"user_id": user_id,
"workflow_id": f"content-gen-{trace_id[:8]}"
}
# Step 1: Research phase - multiple embedding calls for topic analysis
research_data = self._research_topic(topic, trace_id, workflow_data)
# Step 2: Content synthesis using research
content = self._synthesize_content(research_data, trace_id, workflow_data)
# Step 3: Review and enhancement
final_content = self._review_content(content, trace_id, workflow_data)
return {
"topic": topic,
"research": research_data,
"content": content,
"final_content": final_content,
"workflow_id": workflow_data["workflow_id"]
}
def _research_topic(self, topic: str, trace_id: str, workflow_data: dict) -> dict:
research_queries = [
f"What are the key concepts in {topic}?",
f"What are recent developments in {topic}?",
f"What challenges exist in {topic}?"
]
embeddings = []
for i, query in enumerate(research_queries):
headers = self.base_headers.copy()
headers.update({
"adaline-trace-status": "pending",
"adaline-trace-reference-id": trace_id,
"adaline-trace-name": "content-generation",
"adaline-trace-tags": f'["production-v1.3"]',
"adaline-trace-attributes": f'[{{"operation": "create", "key": "query_type", "value": "research"}, {{"operation": "create", "key": "query_index", "value": "{i+1}"}, {{"operation": "create", "key": "total_queries", "value": "{len(research_queries)}"}}]',
"adaline-span-name": f"research-embedding-{i+1}",
})
response = self.openai_client.embeddings.create(
model="text-embedding-3-small",
input=query,
extra_headers=headers
)
embeddings.append({
"query": query,
"embedding": response.data[0].embedding
})
return {"queries": research_queries, "embeddings": embeddings}
def _synthesize_content(self, research_data: dict, trace_id: str, workflow_data: dict) -> str:
headers = self.base_headers.copy()
headers.update({
"adaline-trace-status": "pending",
"adaline-trace-reference-id": trace_id,
"adaline-trace-name": "content-generation",
"adaline-span-name": "content-synthesis",
"adaline-trace-attributes": f'[{"operation": "create", "key": "phase", "value": "synthesis"}, {"operation": "create", "key": "research_queries_count", "value": "{len(research_data["queries"])}"}]',
"adaline-span-variables": f'{{"topic": {{"modality": "text", "value": "{workflow_data["topic"]}"}}, "target_audience": {{"modality": "text", "value": "{workflow_data["target_audience"]}"}}}}'
})
research_context = "\n".join([f"- {query}" for query in research_data["queries"]])
response = self.anthropic_client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=2000,
messages=[
{"role": "user", "content": f"Write a comprehensive article about {workflow_data['topic']} for {workflow_data['target_audience']}. Base it on these research questions:\n{research_context}"}
],
extra_headers=headers
)
return response.content[0].text
def _review_content(self, content: str, trace_id: str, workflow_data: dict) -> str:
headers = self.base_headers.copy()
headers.update({
"adaline-trace-status": "success",
"adaline-trace-reference-id": trace_id,
"adaline-trace-name": "content-generation",
"adaline-span-name": "content-review",
"adaline-trace-attributes": f'[{{"operation": "create", "key": "phase", "value": "review"}, {{"operation": "create", "key": "content_length", "value": "{len(content)}"}, {{"operation": "create", "key": "review_type", "value": "enhancement"}}]',
"adaline-span-variables": f'{{"content": {{"modality": "text", "value": "{content}"}}}}'
})
response = self.openai_client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "Review and enhance the content for clarity, engagement, and accuracy."},
{"role": "user", "content": content}
],
extra_headers=headers
)
return response.choices[0].message.content
# Usage
generator = ContentGenerator()
result = generator.generate_article(
"Machine Learning in Healthcare",
"healthcare professionals",
user_id="user456"
)
print(f"Generated article: {result['workflow_id']}")
Example 3: Conversational Agent with Tools
A conversational agent that uses tool calling with trace status updates and comprehensive error handling.import os
import uuid
import json
from openai import OpenAI
class ConversationalAgent:
def __init__(self):
self.client = OpenAI(
api_key=os.getenv("OPENAI_API_KEY"),
base_url="https://gateway.adaline.ai/v1/openai/"
)
self.base_headers = {
"adaline-api-key": os.getenv("ADALINE_API_KEY"),
"adaline-project-id": os.getenv("ADALINE_PROJECT_ID"),
"adaline-prompt-id": os.getenv("ADALINE_PROMPT_ID")
}
self.tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get current weather for a location",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string", "description": "City name"}
},
"required": ["location"]
}
}
},
{
"type": "function",
"function": {
"name": "calculate",
"description": "Perform mathematical calculations",
"parameters": {
"type": "object",
"properties": {
"expression": {"type": "string", "description": "Math expression to evaluate"}
},
"required": ["expression"]
}
}
}
]
def chat(self, user_message: str, user_id: str) -> str:
trace_id = str(uuid.uuid4())
try:
# Step 1: Initial query processing with tool selection
response = self._process_query(user_message, trace_id, user_id)
# Step 2: Handle tool calls if needed
if response.choices[0].message.tool_calls:
tool_results = self._execute_tools(response.choices[0].message.tool_calls, trace_id, user_id)
# Step 3: Generate final response with tool results
final_response = self._generate_final_response(
user_message, response.choices[0].message, tool_results, trace_id, user_id
)
# Update trace status to completed
self._update_trace_status(trace_id, user_id, "completed", "success")
return final_response
else:
# Direct response without tools
self._update_trace_status(trace_id, user_id, "completed", "success")
return response.choices[0].message.content
except Exception as e:
# Update trace status to failed
self._update_trace_status(trace_id, user_id, "failed", f"error: {str(e)}")
return f"I encountered an error: {str(e)}"
def _process_query(self, user_message: str, trace_id: str, user_id: str):
headers = self.base_headers.copy()
headers.update({
"adaline-trace-reference-id": trace_id,
"adaline-trace-name": "conversational-agent",
"adaline-trace-status": "in_progress",
"adaline-span-name": "query-processing",
"adaline-span-tags": '["tool_selection", "intent_analysis"]',
"adaline-span-attributes": f'{{"user_id": "{user_id}", "message_length": {len(user_message)}, "tools_available": {len(self.tools)}}}'
})
response = self.client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "You are a helpful assistant with access to tools. Use them when appropriate."},
{"role": "user", "content": user_message}
],
tools=self.tools,
tool_choice="auto",
extra_headers=headers
)
return response
def _execute_tools(self, tool_calls, trace_id: str, user_id: str) -> list:
results = []
for i, tool_call in enumerate(tool_calls):
headers = self.base_headers.copy()
headers.update({
"adaline-trace-reference-id": trace_id,
"adaline-trace-name": "conversational-agent",
"adaline-span-name": f"tool-execution-{i+1}",
"adaline-span-tags": f'["tool_call", "{tool_call.function.name}"]',
"adaline-span-attributes": f'{{"tool_name": "{tool_call.function.name}", "tool_call_id": "{tool_call.id}", "execution_order": {i+1}}}'
})
# Simulate tool execution
if tool_call.function.name == "get_weather":
args = json.loads(tool_call.function.arguments)
result = f"Weather in {args['location']}: 72°F, sunny"
elif tool_call.function.name == "calculate":
args = json.loads(tool_call.function.arguments)
try:
result = str(eval(args['expression'])) # Note: Use safe evaluation in production
except:
result = "Error in calculation"
else:
result = "Tool not implemented"
results.append({
"tool_call_id": tool_call.id,
"result": result
})
return results
def _generate_final_response(self, original_message: str, assistant_message, tool_results: list, trace_id: str, user_id: str) -> str:
headers = self.base_headers.copy()
headers.update({
"adaline-trace-reference-id": trace_id,
"adaline-trace-name": "conversational-agent",
"adaline-span-name": "response-generation",
"adaline-span-tags": '["final_response", "tool_integration"]',
"adaline-span-attributes": f'{{"tools_used": {len(tool_results)}, "response_type": "tool_assisted"}}'
})
# Build messages with tool results
messages = [
{"role": "system", "content": "You are a helpful assistant. Provide a natural response using the tool results."},
{"role": "user", "content": original_message},
assistant_message,
]
# Add tool results
for tool_result in tool_results:
messages.append({
"role": "tool",
"tool_call_id": tool_result["tool_call_id"],
"content": tool_result["result"]
})
response = self.client.chat.completions.create(
model="gpt-4",
messages=messages,
extra_headers=headers
)
return response.choices[0].message.content
def _update_trace_status(self, trace_id: str, user_id: str, status: str, details: str):
headers = self.base_headers.copy()
headers.update({
"adaline-trace-reference-id": trace_id,
"adaline-trace-name": "conversational-agent",
"adaline-trace-status": status,
"adaline-span-name": "status-update",
"adaline-span-tags": f'["status_update", "{status}"]',
"adaline-span-attributes": f'{{"final_status": "{status}", "details": "{details}"}}'
})
# Make a minimal API call to update trace status
self.client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "Status update"}],
max_tokens=1,
extra_headers=headers
)
# Usage
agent = ConversationalAgent()
response = agent.chat("What's the weather in New York and what's 15 * 24?", user_id="user789")
print(response)
Best Practices for Complex Workflows
1. Consistent Trace Management
Use a single trace ID across all related operations:trace_id = str(uuid.uuid4())
# Use the same trace_id for all spans in the workflow
2. Meaningful Span Names
Use descriptive span names that indicate the operation:"adaline-span-name": "query-embedding"
"adaline-span-name": "vector-search"
"adaline-span-name": "response-generation"
3. Error Handling with Trace Status
Update trace status to reflect workflow state:"adaline-trace-status": "in_progress" # During execution
"adaline-trace-status": "completed" # On success
"adaline-trace-status": "failed" # On error