Tôi vẫn nhớ rõ ngày hôm đó — dự án xử lý đơn hàng tự động của tôi đang chạy ngon trên Coze khi bỗng nhiên nhận được notification: ConnectionError: timeout exceeded 30s. Toàn bộ pipeline bị treo, 200 đơn hàng chờ xử lý. Sau 3 tiếng debug, tôi phát hiện vấn đề nằm ở chỗ: node LLM gọi tool mà không có timeout handler, và condition branch bị sai logic dẫn đến loop vô hạn.
Bài viết hôm nay sẽ chia sẻ kinh nghiệm thực chiến về cách cấu hình Coze workflow nodes — đặc biệt là LLM tool calling và condition branches — để bạn tránh những cái bẫy mà tôi đã gặp.
Tổng quan về Kiến trúc Coze Workflow
Coze cho phép xây dựng workflow bằng cách kết nối các node với nhau. Với HolySheep AI — nền tảng API AI có độ trễ dưới 50ms, tỷ giá chỉ ¥1=$1 (tiết kiệm 85%+ so với OpenAI) — bạn có thể tích hợp LLM mạnh mẽ vào workflow một cách đáng tin cậy.
Cấu hình LLM Node với Tool Calling
Tool calling cho phép LLM chủ động gọi function để thực hiện tác vụ cụ thể. Dưới đây là cách cấu hình đúng cách.
Bước 1: Định nghĩa Tool Schema
{
"name": "get_order_status",
"description": "Truy vấn trạng thái đơn hàng theo order_id",
"parameters": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "Mã đơn hàng cần tra cứu"
},
"include_history": {
"type": "boolean",
"description": "Bao gồm lịch sử thay đổi trạng thái",
"default": false
}
},
"required": ["order_id"]
}
}Bước 2: Kết nối với HolySheep API
#!/usr/bin/env python3
import httpx
import json
from typing import Optional
class CozeLLMConnector:
"""Kết nối Coze workflow với HolySheep AI qua tool calling"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.client = httpx.Client(
timeout=30.0, # Timeout 30s - quan trọng để tránh treo
limits=httpx.Limits(max_keepalive_connections=20)
)
def call_with_tools(
self,
model: str,
messages: list,
tools: list,
tool_choice: str = "auto"
) -> dict:
"""
Gọi LLM với tool calling qua HolySheep API
Args:
model: Tên model (gpt-4.1, claude-sonnet-4.5, deepseek-v3.2)
messages: Lịch sử hội thoại
tools: Danh sách tool schema
tool_choice: "auto", "none", hoặc {"type": "function", "function": {...}}
Returns:
Response với tool_calls nếu có
"""
endpoint = f"{self.BASE_URL}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"tools": tools,
"tool_choice": tool_choice,
"stream": False
}
try:
response = self.client.post(endpoint, headers=headers, json=payload)
response.raise_for_status()
return response.json()
except httpx.TimeoutException:
# Xử lý timeout - trả về fallback response
return {
"error": "timeout",
"message": "LLM response exceeded 30s timeout",
"fallback_action": "retry_with_simpler_prompt"
}
except httpx.HTTPStatusError as e:
# Xử lý HTTP error codes
if e.response.status_code == 401:
return {"error": "unauthorized", "message": "Invalid API key"}
elif e.response.status_code == 429:
return {"error": "rate_limit", "message": "Rate limit exceeded", "retry_after": 60}
raise
Sử dụng với DeepSeek V3.2 - model rẻ nhất, chỉ $0.42/MTok
connector = CozeLLMConnector(api_key="YOUR_HOLYSHEEP_API_KEY")
result = connector.call_with_tools(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": "Bạn là trợ lý xử lý đơn hàng"},
{"role": "user", "content": "Kiểm tra trạng thái đơn hàng ORD-2024-8847"}
],
tools=[{
"type": "function",
"function": {
"name": "get_order_status",
"description": "Truy vấn trạng thái đơn hàng",
"parameters": {
"type": "object",
"properties": {
"order_id": {"type": "string"}
},
"required": ["order_id"]
}
}
}]
)
print(f"Kết quả: {json.dumps(result, indent=2, ensure_ascii=False)}")Cấu hình Condition Branch — Tránh Loop Vô Hạn
Đây là phần dễ gây ra bug nhất trong workflow. Condition branch cần được thiết kế cẩn thận với exit condition rõ ràng.
#!/usr/bin/env python3
"""
Coze Workflow: Condition Branch Logic Handler
Xử lý các nhánh điều kiện với timeout protection
"""
from enum import Enum
from dataclasses import dataclass
from typing import Callable, Optional
import time
class WorkflowBranch(Enum):
"""Các nhánh có thể có trong workflow"""
SUCCESS = "success"
RETRY = "retry"
FALLBACK = "fallback"
ERROR = "error"
TIMEOUT = "timeout"
@dataclass
class BranchCondition:
"""Định nghĩa một điều kiện rẽ nhánh"""
name: str
check: Callable[[dict], bool]
max_retries: int = 3
timeout_seconds: float = 30.0
class ConditionBranchHandler:
"""
Xử lý rẽ nhánh có điều kiện trong Coze workflow
Điểm quan trọng:
1. Luôn có exit condition
2. Retry với exponential backoff
3. Timeout protection cho mỗi nhánh
"""
def __init__(self):
self.conditions: list[BranchCondition] = []
self.execution_log: list[dict] = []
def add_condition(
self,
name: str,
check_fn: Callable[[dict], bool],
max_retries: int = 3
):
"""Thêm một điều kiện rẽ nhánh"""
self.conditions.append(
BranchCondition(name=name, check=check_fn, max_retries=max_retries)
)
def evaluate(
self,
context: dict,
llm_response: dict,
available_tools: list[str]
) -> WorkflowBranch:
"""
Đánh giá response và quyết định nhánh thực thi
Returns:
WorkflowBranch tương ứng với điều kiện thỏa mãn
"""
# Kiểm tra lỗi HTTP
if "error" in llm_response:
error_type = llm_response["error"]
if error_type == "timeout":
self.execution_log.append({
"branch": "timeout",
"timestamp": time.time(),
"context": context.get("order_id", "unknown")
})
return WorkflowBranch.RETRY
elif error_type == "unauthorized":
self.execution_log.append({
"branch": "error",
"error": "API key invalid",
"timestamp": time.time()
})
return WorkflowBranch.ERROR
elif error_type == "rate_limit":
self.execution_log.append({
"branch": "retry",
"reason": "rate_limit",
"retry_after": llm_response.get("retry_after", 60),
"timestamp": time.time()
})
return WorkflowBranch.FALLBACK
# Kiểm tra tool_calls từ LLM response
tool_calls = llm_response.get("choices", [{}])[0].get("message", {}).get("tool_calls", [])
if tool_calls:
# LLM muốn gọi tool
for call in tool_calls:
tool_name = call.get("function", {}).get("name", "")
if tool_name in available_tools:
self.execution_log.append({
"branch": "success",
"tool_called": tool_name,
"timestamp": time.time()
})
return WorkflowBranch.SUCCESS
else:
# Tool không có sẵn -> fallback
self.execution_log.append({
"branch": "fallback",
"reason": f"Tool {tool_name} not available",
"timestamp": time.time()
})
return WorkflowBranch.FALLBACK
# Kiểm tra điều kiện tùy chỉnh
for condition in self.conditions:
if condition.check(context):
return WorkflowBranch.SUCCESS
# Mặc định: fallback
return WorkflowBranch.FALLBACK
def get_retry_strategy(self, attempt: int) -> float:
"""
Exponential backoff cho retry
Args:
attempt: Số lần thử (bắt đầu từ 1)
Returns:
Thời gian chờ tính bằng giây
"""
base_delay = 1.0 # 1 giây
max_delay = 60.0 # Tối đa 60 giây
delay = min(base_delay * (2 ** (attempt - 1)), max_delay)
# Thêm jitter để tránh thundering herd
import random
jitter = random.uniform(0, 0.3 * delay)
return delay + jitter
Demo sử dụng
handler = ConditionBranchHandler()
Thêm điều kiện tùy chỉnh
handler.add_condition(
name="high_value_order",
check_fn=lambda ctx: ctx.get("order_value", 0) > 10000,
max_retries=5
)
Test với mock response
test_context = {"order_id": "ORD-2024-8847", "order_value": 15000}
Trường hợp 1: Timeout
timeout_response = {
"error": "timeout",
"message": "LLM response exceeded 30s"
}
branch = handler.evaluate(test_context, timeout_response, [])
print(f"Branch cho timeout: {branch.value}") # -> retry
Trường hợp 2: Tool call thành công
success_response = {
"choices": [{
"message": {
"tool_calls": [{
"function": {"name": "get_order_status"}
}]
}
}]
}
branch = handler.evaluate(test_context, success_response, ["get_order_status"])
print(f"Branch cho tool call: {branch.value}") # -> success
print(f"\nExecution log: {json.dumps(handler.execution_log, indent=2)}")Tích hợp Hoàn chỉnh vào Coze Workflow
Dưới đây là workflow hoàn chỉnh kết hợp LLM tool calling với condition branches:
#!/usr/bin/env python3
"""
Coze Workflow: Complete Order Processing Pipeline
Pipeline hoàn chỉnh xử lý đơn hàng với error handling
"""
import asyncio
import httpx
from typing import Optional
from dataclasses import dataclass
import json
@dataclass
class OrderContext:
"""Context cho workflow xử lý đơn hàng"""
order_id: str
customer_id: str
items: list[dict]
total_amount: float
priority: str = "normal"
class OrderProcessingWorkflow:
"""
Workflow xử lý đơn hàng hoàn chỉnh
Nodes trong workflow:
1. LLM Analyze (phân tích đơn hàng, quyết định action)
2. Condition Router (định tuyến theo loại action)
3. Tool Executor (gọi tool tương ứng)
4. Result Aggregator (tổng hợp kết quả)
"""
BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_KEY = "YOUR_HOLYSHEEP_API_KEY"
def __init__(self):
self.http_client = httpx.AsyncClient(
timeout=httpx.Timeout(30.0, connect=5.0),
limits=httpx.Limits(max_connections=100, max_keepalive_connections=20)
)
self.available_tools = [
"check_inventory",
"calculate_shipping",
"apply_discount",
"confirm_payment",
"create_shipment"
]
async def llm_analyze_node(self, order: OrderContext) -> dict:
"""
Node 1: LLM phân tích đơn hàng
Prompt gửi sang HolySheep API (model: gpt-4.1 hoặc deepseek-v3.2)
"""
system_prompt = """Bạn là AI phân tích đơn hàng tự động.
Với mỗi đơn hàng, quyết định action cần thực hiện:
- "validate": Xác thực thông tin đơn hàng
- "discount": Áp dụng giảm giá nếu đủ điều kiện
- "fulfill": Xử lý fulfillment trực tiếp
- "review": Cần human review
Luôn trả về JSON với fields: action, reasoning, confidence"""
user_prompt = f"""Phân tích đơn hàng:
- Mã đơn: {order.order_id}
- Khách hàng: {order.customer_id}
- Sản phẩm: {json.dumps(order.items, ensure_ascii=False)}
- Tổng tiền: ${order.total_amount}
- Ưu tiên: {order.priority}"""
headers = {
"Authorization": f"Bearer {self.HOLYSHEEP_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": "deepseek-v3.2", # Model rẻ nhất, $0.42/MTok
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
"tools": [{
"type": "function",
"function": {
"name": tool,
"description": f"Tool để {tool.replace('_', ' ')}"
}
} for tool in self.available_tools],
"tool_choice": "auto",
"response_format": {"type": "json_object"}
}
try:
response = await self.http_client.post(
f"{self.BASE_URL}/chat/completions",
headers=headers,
json=payload
)
response.raise_for_status()
result = response.json()
return {
"success": True,
"action": result.get("choices", [{}])[0].get("message", {}).get("content", "{}"),
"tool_calls": result.get("choices", [{}])[0].get("message", {}).get("tool_calls", [])
}
except httpx.TimeoutException:
return {
"success": False,
"error": "timeout",
"action": "retry",
"retry_count": 1
}
except httpx.HTTPStatusError as e:
return {
"success": False,
"error": f"HTTP_{e.response.status_code}",
"action": "fallback"
}
def condition_router(self, llm_result: dict) -> str:
"""
Node 2: Condition Router - định tuyến theo kết quả LLM
Flow:
- success + tool_calls -> execute_tools
- success + no tool -> direct_fulfillment
- error = timeout -> retry
- error = other -> human_review
"""
if not llm_result.get("success"):
error_type = llm_result.get("error", "")
if error_type == "timeout":
return "retry"
return "human_review"
tool_calls = llm_result.get("tool_calls", [])
if tool_calls:
return "execute_tools"
action = llm_result.get("action", "")
try:
action_obj = json.loads(action) if isinstance(action, str) else action
if action_obj.get("action") == "review":
return "human_review"
except:
pass
return "direct_fulfillment"
async def execute_tools_node(self, tool_calls: list) -> dict:
"""
Node 3: Tool Executor - thực thi các tool được gọi
"""
results = []
for call in tool_calls:
tool_name = call.get("function", {}).get("name", "")
tool_args = json.loads(call.get("function", {}).get("arguments", "{}"))
# Thực thi tool (mock implementation)
result = await self._execute_single_tool(tool_name, tool_args)
results.append({
"tool": tool_name,
"args": tool_args,
"result": result
})
return {"tool_results": results, "all_success": all(r.get("success") for r in results)}
async def _execute_single_tool(self, tool_name: str, args: dict) -> dict:
"""Execute một tool đơn lẻ"""
# Mock implementation - thay bằng logic thực tế
return {
"success": True,
"tool": tool_name,
"output": f"Executed {tool_name} with args {args}"
}
async def run(self, order: OrderContext) -> dict:
"""
Chạy complete workflow
Returns:
Dict chứa kết quả và execution trace
"""
trace = []
max_retries = 3
for attempt in range(1, max_retries + 1):
# Node 1: LLM Analyze
llm_result = await self.llm_analyze_node(order)
trace.append({
"node": "llm_analyze",
"attempt": attempt,
"result": llm_result
})
# Node 2: Condition Router
next_step = self.condition_router(llm_result)
trace.append({
"node": "condition_router",
"decision": next_step
})
if next_step == "retry":
if attempt < max_retries:
await asyncio.sleep(2 ** attempt) # Exponential backoff
continue
else:
return {"status": "failed", "reason": "max_retries_exceeded", "trace": trace}
# Node 3: Execute Tools
if next_step