Building intelligent automation workflows requires mastering two critical concepts: how Large Language Models interact with external tools, and how conditional logic routes data through your pipeline. Whether you're handling thousands of customer inquiries during flash sales, processing enterprise documents through a RAG system, or building a weekend project that scales beyond your expectations, understanding these Coze workflow nodes will dramatically improve your system's efficiency and reliability.
Introduction: Why Tool Calling and Branching Matter
When I first built a customer service bot for a friend's e-commerce store, I made the classic mistake of treating every user query the same. The bot would launch into lengthy explanations about return policies when users simply wanted to check order status. After spending three days refining prompts with minimal improvement, I realized the solution wasn't better prompts—it was proper workflow architecture with conditional branching that routes queries to specialized handlers based on intent detection.
Coze workflows excel at this pattern because they separate concerns: the LLM node handles understanding and generation, while conditional nodes make routing decisions. Combined with tool calling, your workflow can query databases, call APIs, or manipulate files in response to detected user needs. This tutorial walks through the complete implementation using HolySheep AI as your LLM backbone, delivering sub-50ms latency at a fraction of enterprise costs.
Understanding Coze Workflow Architecture
A Coze workflow consists of interconnected nodes that pass data between each stage. The three essential node types you'll work with are:
- LLM Nodes — Process natural language input, extract structured data, and generate responses
- Tool Nodes — Execute external operations like API calls, database queries, or file operations
- Conditional Branch Nodes — Route workflow execution based on variable values or expressions
The power comes from combining these nodes: your LLM can decide when to call tools, and conditional branches can re-route based on tool outputs. This creates dynamic, responsive automation that adapts to real-world inputs.
Setting Up Your HolySheep AI Integration
Before building workflows, configure your LLM provider. Sign up here for HolySheep AI, which provides API-compatible access to major models at dramatically reduced pricing: DeepSeek V3.2 costs just $0.42 per million tokens versus the $7.30+ you'd pay elsewhere, with support for WeChat and Alipay payments.
The 2026 model pricing landscape shows significant variation:
- GPT-4.1 — $8.00 per million output tokens
- Claude Sonnet 4.5 — $15.00 per million output tokens
- Gemini 2.5 Flash — $2.50 per million output tokens
- DeepSeek V3.2 — $0.42 per million output tokens
For high-volume production workflows, DeepSeek V3.2 delivers 95% cost savings while maintaining excellent quality. For complex reasoning tasks requiring frontier models, Claude Sonnet 4.5 remains the top choice despite premium pricing.
Building the Workflow: Step-by-Step Implementation
Step 1: Create the Workflow Structure
Start by defining your workflow in Coze's visual editor. For our e-commerce customer service example, we'll build a flow that classifies incoming queries, calls appropriate tools, and routes responses based on customer intent.
Step 2: Configure the Intent Classification LLM Node
The first LLM node analyzes the user's message and extracts structured intent data. This node uses a system prompt to enforce consistent classification:
{
"nodes": [
{
"id": "intent_classifier",
"type": "llm",
"config": {
"model": "deepseek-v3.2",
"provider": "holysheep",
"system_prompt": "You are an intent classification system for an e-commerce customer service bot. Analyze the user's message and classify their primary intent into one of these categories: ORDER_STATUS, RETURN_REQUEST, PRODUCT_INQUIRY, BILLING_ISSUE, GENERAL_QUESTION, or ESCALATION_NEEDED. Return your response as a JSON object with 'intent' and 'confidence' fields.",
"temperature": 0.1,
"max_tokens": 150
}
}
]
}Step 3: Implement Tool Calling for Order Status
Tool calling allows your LLM to request external actions. When the intent classifier detects ORDER_STATUS, your workflow should call an order lookup tool. Here's the complete configuration for a tool-calling workflow:
import requests
import json
class HolySheepWorkflow:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
def call_llm_with_tools(self, user_message: str, intent: str, tools: list):
"""Execute LLM with tool definitions for dynamic function calling"""
endpoint = f"{self.base_url}/chat/completions"
# Define available tools for this workflow
tool_schemas = {
"get_order_status": {
"name": "get_order_status",
"description": "Retrieves current status and tracking information for a customer order",
"parameters": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "The unique order identifier"
}
},
"required": ["order_id"]
}
},
"lookup_product": {
"name": "lookup_product",
"description": "Searches product catalog for availability and pricing",
"parameters": {
"type": "object",
"properties": {
"product_name": {"type": "string"},
"category": {"type": "string"}
},
"required": ["product_name"]
}
},
"process_return": {
"name": "process_return",
"description": "Initiates a return request and generates return shipping label",
"parameters": {
"type": "object",
"properties": {
"order_id": {"type": "string"},
"reason": {"type": "string"},
"requested_action": {"type": "string", "enum": ["refund", "exchange"]}
},
"required": ["order_id", "reason"]
}
}
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
# Build messages with tool context
messages = [
{
"role": "system",
"content": f"""You are a helpful e-commerce customer service assistant.
The customer has sent: '{user_message}'
Detected intent: {intent}
When you need to look up information, use the available tools.
Always extract order IDs from customer messages when provided.
Format your responses clearly and concisely."""
},
{
"role": "user",
"content": user_message
}
]
payload = {
"model": "deepseek-v3.2",
"messages": messages,
"tools": [tool_schemas[t] for t in tools if t in tool_schemas],
"tool_choice": "auto",
"temperature": 0.3,
"max_tokens": 500
}
response = requests.post(endpoint, headers=headers, json=payload)
return response.json()
def execute_tool(self, tool_name: str, arguments: dict):
"""Execute tool calls and return structured results"""
# Simulated tool implementations
tool_handlers = {
"get_order_status": self._get_order_status,
"lookup_product": self._lookup_product,
"process_return": self._process_return
}
if tool_name in tool_handlers:
return tool_handlers[tool_name](**arguments)
return {"error": f"Unknown tool: {tool_name}"}
def _get_order_status(self, order_id: str):
"""Simulated order lookup with realistic response"""
return {
"order_id": order_id,
"status": "shipped",
"tracking_number": "1Z999AA10123456784",
"estimated_delivery": "2026-01-20",
"carrier": "UPS",
"last_update": "Package arrived at local distribution center"
}
def _lookup_product(self, product_name: str, category: str = None):
"""Simulated product lookup"""
return {
"product_name": product_name,
"in_stock": True,
"quantity_available": 47,
"price": 29.99,
"category": category or "general"
}
def _process_return(self, order_id: str, reason: str, requested_action: str = "refund"):
"""Simulated return processing"""
return {
"return_id": f"RTN-{order_id[-6:]}",
"order_id": order_id,
"reason": reason,
"requested_action": requested_action,
"return_label_url": f"https://returns.example.com/{order_id}",
"status": "initiated",
"estimated_refund": "3-5 business days"
}
Usage example
workflow = HolySheepWorkflow(api_key="YOUR_HOLYSHEEP_API_KEY")
Classify intent and call appropriate tools
user_message = "I want to return my order #12345, it doesn't fit"
result = workflow.call_llm_with_tools(
user_message=user_message,
intent="RETURN_REQUEST",
tools=["get_order_status", "lookup_product", "process_return"]
)
Execute any tool calls returned by the LLM
if "tool_calls" in result.get("choices", [{}])[0]:
for call in result["choices"][0]["tool_calls"]:
tool_result = workflow.execute_tool(call["function"]["name"],
json.loads(call["function"]["arguments"]))
print(f"Tool: {call['function']['name']}")
print(f"Result: {json.dumps(tool_result, indent=2)}")Step 4: Configure Conditional Branching Logic
Conditional branches route workflow execution based on node outputs. After your LLM classifies intent, use a switch node to direct traffic:
{
"conditional_branch": {
"id": "route_by_intent",
"type": "condition",
"input": "{{intent_classifier.output}}",
"conditions": [
{
"case": "ORDER_STATUS",
"target_node": "order_status_handler",
"description": "Route to order lookup flow"
},
{
"case": "RETURN_REQUEST",
"target_node": "return_handler",
"description": "Route to return processing flow"
},
{
"case": "PRODUCT_INQUIRY",
"target_node": "product_search",
"description": "Route to catalog search"
},
{
"case": "ESCALATION_NEEDED",
"target_node": "human_escalation",
"description": "Route to human agent queue"
}
],
"default": "general_response_handler"
},
"condition_expression_logic": {
"description": "For complex conditions, use expression syntax",
"examples": [
{
"expression": "{{confidence}} > 0.8 AND {{intent}} == 'ORDER_STATUS'",
"action": "proceed_directly"
},
{
"expression": "{{confidence}} < 0.6",
"action": "request_clarification"
},
{
"expression": "{{has_order_id}} == true AND {{intent}} == 'RETURN_REQUEST'",
"action": "auto_process_return"
}
]
}
}Building a Complete Workflow: Real-World Example
Let me walk through how I implemented a production RAG workflow for document processing. The system needed to accept user queries, search relevant documents, generate answers with citations, and handle cases where no relevant content exists.
import requests
import json
from typing import List, Dict, Optional
class CozeRAGWorkflow:
"""Production RAG workflow with tool calling and conditional branching"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.workflow_state = {}
def execute_complete_workflow(self, user_query: str, context_limit: int = 5) -> Dict:
"""Execute the full RAG workflow with tool calling and branching"""
print(f"Starting RAG workflow for query: {user_query}")
# Step 1: Intent Detection
intent_result = self._classify_intent(user_query)
self.workflow_state["intent"] = intent_result["intent"]
self.workflow_state["requires_context"] = intent_result.get("requires_context", True)
print(f"Intent classified as: {intent_result['intent']}")
# Step 2: Conditional Branch - Check if this query needs document context
if self.workflow_state["requires_context"]:
# Step 3: Semantic Search
search_results = self._semantic_search(user_query, top_k=context_limit)
self.workflow_state["search_results"] = search_results
# Step 4: Check if relevant documents found
if not search_results or all(r["relevance_score"] < 0.5 for r in search_results):
# Step 5a: No relevant docs - branch to fallback
return self._handle_no_results(user_query)
else:
# Step 6: Generate answer with context
return self._generate_answer_with_context(user_query, search_results)
else:
# Step 5b: Direct response - branch to general handler
return self._generate_general_response(user_query)
def _classify_intent(self, query: str) -> Dict:
"""Use LLM to classify query intent"""
endpoint = f"{self.base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
messages = [
{
"role": "system",
"content": """Classify this user query into one of these categories:
- KNOWLEDGE_QUERY: User wants information that might be in documents
- GENERAL_CHAT: Casual conversation not requiring document lookup
- TASK_EXECUTION: User wants to perform an action (create, update, delete)
- CLARIFICATION: Query is unclear or too vague
Respond with JSON containing 'intent', 'confidence' (0-1), and 'requires_context' (boolean)."""
},
{
"role": "user",
"content": query
}
]
payload = {
"model": "deepseek-v3.2",
"messages": messages,
"temperature": 0.1,
"max_tokens": 200
}
response = requests.post(endpoint, headers=headers, json=payload)
result = response.json()
try:
return json.loads(result["choices"][0]["message"]["content"])
except:
return {"intent": "KNOWLEDGE_QUERY", "confidence": 0.5, "requires_context": True}
def _semantic_search(self, query: str, top_k: int = 5) -> List[Dict]:
"""Simulated semantic search with relevance scoring"""
# In production, this would call your vector database (Pinecone, Weaviate, etc.)
# For demonstration, returning simulated results
document_corpus = [
{"id": "doc_001", "content": "Product return policy allows returns within 30 days with original packaging.", "source": "return_policy.md"},
{"id": "doc_002", "content": "Shipping times vary by location: domestic 3-5 days, international 7-14 days.", "source": "shipping_info.md"},
{"id": "doc_003", "content": "Customer support is available 24/7 via chat, email, and phone.", "source": "support_hours.md"},
{"id": "doc_004", "content": "Payment methods include credit card, PayPal, WeChat Pay, and Alipay.", "source": "payment_options.md"},
{"id": "doc_005", "content": "Order tracking is available through our mobile app or website dashboard.", "source": "tracking_guide.md"}
]
# Simulated relevance scoring (in production, use embeddings)
query_lower = query.lower()
scored_docs = []
keywords_map = {
"return": ["doc_001"],
"ship": ["doc_002", "doc_005"],
"support": ["doc_003"],
"payment": ["doc_004"],
"track": ["doc_005"],
"order": ["doc_002", "doc_005"]
}
for keyword, doc_ids in keywords_map.items():
if keyword in query_lower:
for doc_id in doc_ids:
if doc_id not in [d["id"] for d in scored_docs]:
doc = next(d for d in document_corpus if d["id"] == doc_id)
scored_docs.append({
**doc,
"relevance_score": 0.7
})
# Sort by relevance and return top_k
scored_docs.sort(key=lambda x: x["relevance_score"], reverse=True)
return scored_docs[:top_k]
def _generate_answer_with_context(self, query: str, context_docs: List[Dict]) -> Dict:
"""Generate answer using retrieved context with citations"""
endpoint = f"{self.base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
# Build context string with citations
context_with_citations = "\n\n".join([
f"[Source {i+1}: {doc['source']}]\n{doc['content']}"
for i, doc in enumerate(context_docs)
])
messages = [
{
"role": "system",
"content": f"""You are a helpful assistant answering questions based on provided context.
Always cite your sources using [Source N] notation.
If the context doesn't fully answer the question, say so clearly.
Be concise but complete."""
},
{
"role": "user",
"content": f"Question: {query}\n\nContext:\n{context_with_citations}"
}
]
payload = {
"model": "deepseek-v3.2",
"messages": messages,
"temperature": 0.3,
"max_tokens": 600
}
response = requests.post(endpoint, headers=headers, json=payload)
result = response.json()
return {
"answer": result["choices"][0]["message"]["content"],
"sources": [doc["source"] for doc in context_docs],
"workflow_branch": "context_rag",
"context_used": len(context_docs)
}
def _handle_no_results(self, query: str) -> Dict:
"""Branch handler when no relevant documents found"""
return {
"answer": f"I couldn't find specific information about that in our documents. Could you rephrase your question or provide more details?",
"sources": [],
"workflow_branch": "no_results_fallback",
"suggestions": [
"Try using different keywords",
"Ask about a specific product or policy",
"Contact support for personalized help"
]
}
def _generate_general_response(self, query: str) -> Dict:
"""Branch handler for general queries not requiring document context"""
endpoint = f"{self.base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
messages = [
{"role": "system", "content": "You are a friendly and helpful assistant."},
{"role": "user", "content": query}
]
payload = {
"model": "deepseek-v3.2",
"messages": messages,
"temperature": 0.7,
"max_tokens": 400
}
response = requests.post(endpoint, headers=headers, json=payload)
result = response.json()
return {
"answer": result["choices"][0]["message"]["content"],
"sources": [],
"workflow_branch": "general_chat"
}
Execute the complete workflow
workflow = CozeRAGWorkflow(api_key="YOUR_HOLYSHEEP_API_KEY")
Test queries demonstrating different branches
test_queries = [
"What's your return policy?",
"Hello, how are you today?",
"Can you tell me about shipping to Canada?"
]
for query in