> ## Documentation Index
> Fetch the complete documentation index at: https://www.adaline.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Multi-step workflows

> Real-world examples from simple single-span workflows to complex multi-span applications

# 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](/docs/reference/proxy/headers)** 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.

<CodeGroup>
  ```python Python theme={null}
  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)
  ```
</CodeGroup>

***

## Example 2: Multi-Step Content Generation

A content generation workflow that uses multiple embedding calls and chat completions with span attributes and variables tracking.

<CodeGroup>
  ```python Python theme={null}
  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']}")
  ```
</CodeGroup>

***

## Example 3: Conversational Agent with Tools

A conversational agent that uses tool calling with trace status updates and comprehensive error handling.

<CodeGroup>
  ```python Python theme={null}
  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)
  ```
</CodeGroup>

***

## Best Practices for Complex Workflows

### 1. Consistent Trace Management

Use a single trace ID across all related operations:

```python theme={null}
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:

```python theme={null}
"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:

```python theme={null}
"adaline-trace-status": "in_progress"  # During execution
"adaline-trace-status": "completed"    # On success
"adaline-trace-status": "failed"       # On error
```

***