LangChain Chain Composition: Hướng Dẫn Toàn Diện Cho Kỹ Sư Production

Giới Thiệu

Trong quá trình xây dựng các ứng dụng AI production tại HolySheep AI, tôi đã tiếp cận hàng chục kiến trúc LLM khác nhau. Điều tôi nhận ra sau 3 năm thực chiến: Chain Composition không chỉ là cách ghép nối các module lại với nhau — đó là nghệ thuật kiểm soát luồng dữ liệu, tối ưu chi phí, và đảm bảo độ tin cậy cao nhất có thể. Bài viết này sẽ đi sâu vào những gì sách vở không dạy: cách tôi xử lý race condition trong parallel chains, cách tiết kiệm 85% chi phí với HolySheep AI so với OpenAI, và những lỗi production mà tôi đã "đổ máu" để tìm ra giải pháp.

1. Tại Sao Chain Composition Quan Trọng

Khi bạn xây dựng một ứng dụng RAG (Retrieval-Augmented Generation) hoặc agent system, bạn không chỉ gọi một LLM duy nhất. Bạn cần:

• Chuỗi xử lý nhiều bước (multi-step processing)
• Xử lý song song các tác vụ độc lập (parallel execution)
• Điều kiện rẽ nhánh dựa trên output (conditional branching)
• Tái sử dụng kết quả trung gian (intermediate result caching)

Với HolySheep AI, bạn có thể chạy các chain phức tạp với chi phí cực kỳ thấp:

• DeepSeek V3.2: chỉ $0.42/MTok — rẻ hơn 95% so với GPT-4.1
• Gemini 2.5 Flash: $2.50/MTok — lý tưởng cho batch processing
• Độ trễ trung bình <50ms với API endpoint tại Việt Nam

2. Kiến Trúc Chain Cơ Bản

2.1 Sequential Chain - Xử Lý Tuần Tự

Đây là dạng chain đơn giản nhất: output của bước trước là input của bước sau. Tôi thường dùng pattern này cho các tác vụ pipeline đơn giản.

import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

Cấu hình HolySheep AI - KHÔNG dùng OpenAI endpoint
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"

llm = ChatOpenAI(
    model="deepseek-v3.2",
    temperature=0.7,
    api_key="YOUR_HOLYSHEEP_API_KEY"
)

Bước 1: Phân tích yêu cầu
analyze_prompt = ChatPromptTemplate.from_messages([
    ("system", "Bạn là chuyên gia phân tích yêu cầu. Trích xuất 3 từ khóa chính."),
    ("human", "{input}")
])
analyzer = analyze_prompt | llm | StrOutputParser()

Bước 2: Tạo nội dung dựa trên phân tích
generate_prompt = ChatPromptTemplate.from_messages([
    ("system", "Viết bài blog 200 từ về chủ đề: {keywords}"),
    ("human", "Yêu cầu ban đầu: {input}")
])
generator = generate_prompt | llm | StrOutputParser()

Chain tuần tự: analyzer -> generator
chain = analyzer | generator

Thực thi
result = chain.invoke({"input": "Hướng dẫn xây dựng RAG system với LangChain"})
print(result)

2.2 Parallel Chain - Xử Lý Song Song

Đây là nơi tôi thấy nhiều kỹ sư junior mắc lỗi. Khi các tác vụ độc lập, bạn phải chạy chúng song song để tận dụng throughput. LangChain cung cấp RunnableParallel cho việc này.

from langchain_core.runnables import RunnableParallel

Định nghĩa các chain độc lập
analyze_prompt = ChatPromptTemplate.from_messages([
    ("system", "Phân tích mặt mạnh và trả về JSON {{'strengths': []}}"),
    ("human", "{input}")
])

weakness_prompt = ChatPromptTemplate.from_messages([
    ("system", "Phân tích mặt yếu và trả về JSON {{'weaknesses': []}}"),
    ("human", "{input}")
])

opportunity_prompt = ChatPromptTemplate.from_messages([
    ("system", "Phân tích cơ hội và trả về JSON {{'opportunities': []}}"),
    ("human", "{input}")
])

Tạo các chain riêng lẻ
analyze_chain = analyze_prompt | llm | StrOutputParser()
weakness_chain = weakness_prompt | llm | StrOutputParser()
opportunity_chain = opportunity_prompt | llm | StrOutputParser()

Chạy song song với RunnableParallel
parallel_analysis = RunnableParallel(
    strengths=analyze_chain,
    weaknesses=weakness_chain,
    opportunities=opportunity_chain
)

Thực thi song song - tất cả 3 LLM call chạy cùng lúc
result = parallel_analysis.invoke({"input": "Phân tích startup AI tại Việt Nam"})

Benchmark: Sequential vs Parallel
import time

start = time.time()
Sequential (sai cách)
seq_result = (analyze_chain | weakness_chain | opportunity_chain).invoke({"input": "test"})
seq_time = time.time() - start

start = time.time()
Parallel (đúng cách)
par_result = parallel_analysis.invoke({"input": "test"})
par_time = time.time() - start

print(f"Sequential: {seq_time:.2f}s")
print(f"Parallel: {par_time:.2f}s")
print(f"Speedup: {seq_time/par_time:.1f}x")

2.3 Router Chain - Điều Hướng Thông Minh

Trong thực tế, tôi cần route user query đến chain phù hợp. Đây là pattern mạnh mẽ cho multi-intent systems.

from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel

Định nghĩa schema cho intent classification
class IntentClassification(BaseModel):
    intent: str  # "technical_support", "billing", "general"
    confidence: float

router_prompt = ChatPromptTemplate.from_messages([
    ("system", """Phân loại intent của user và trả về JSON:
    - "technical_support": hỏi về code, bug, API
    - "billing": hỏi về thanh toán, giá cả
    - "general": câu hỏi chung khác"""),
    ("human", "{input}")
])

router = router_prompt | llm.with_structured_output(IntentClassification)

Định nghĩa các chain xử lý riêng
technical_chain = ChatPromptTemplate.from_messages([
    ("system", "Bạn là kỹ sư hỗ trợ kỹ thuật. Giải đáp: {input}")
]) | llm | StrOutputParser()

billing_chain = ChatPromptTemplate.from_messages([
    ("system", """Bạn là chuyên gia tư vấn. Giá HolySheep AI 2026:
    - GPT-4.1: $8/MTok
    - Claude Sonnet 4.5: $15/MTok  
    - DeepSeek V3.2: $0.42/MTok (TIẾT KIỆM 95%)
    Trả lời: {input}""")
]) | llm | StrOutputParser()

general_chain = ChatPromptTemplate.from_messages([
    ("system", "Bạn là trợ lý thân thiện. Trả lời: {input}")
]) | llm | StrOutputParser()

Conditional routing logic
def route_intent(state: dict) -> str:
    """Route based on classified intent"""
    if state["intent"] == "technical_support":
        return "technical"
    elif state["intent"] == "billing":
        return "billing"
    return "general"

Router chain với conditional routing
from operator import itemgetter

router_chain = router | itemgetter("intent") | {
    "technical": technical_chain,
    "billing": billing_chain,
    "general": general_chain
}

Sử dụng
result = router_chain.invoke({
    "input": "Giá của DeepSeek V3.2 là bao nhiêu?"
})
print(result)

3. Xử Lý Lỗi và Retry Logic

Production code không thể thiếu retry mechanism. Trong quá trình vận hành HolySheep AI, tôi gặp nhiều trường hợp timeout, rate limit, và network error. Đây là cách tôi xử lý:

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from openai import RateLimitError, APIError
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

Retry decorator với exponential backoff
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type((RateLimitError, APIError, TimeoutError)),
    before_sleep=lambda retry_state: logger.warning(
        f"Retry {retry_state.attempt_number} sau {retry_state.next_action.sleep}s"
    )
)
def call_llm_with_retry(prompt: str, model: str = "deepseek-v3.2") -> str:
    """Gọi LLM với retry tự động"""
    try:
        response = llm.invoke(prompt)
        return response.content
    except RateLimitError as e:
        logger.error(f"Rate limit exceeded: {e}")
        raise  # Trigger retry
    except APIError as e:
        logger.error(f"API Error: {e}")
        raise  # Trigger retry

Fallback chain - nếu primary model fail, dùng backup
primary_chain = ChatPromptTemplate.from_messages([
    ("system", "Bạn là chuyên gia. Trả lời chi tiết."),
    ("human", "{input}")
]) | llm.bind(model="deepseek-v3.2") | StrOutputParser()

fallback_chain = ChatPromptTemplate.from_messages([
    ("system", "Bạn là trợ lý AI. Trả lời ngắn gọn."),
    ("human", "{input}")
]) | llm.bind(model="gemini-2.5-flash") | StrOutputParser()  # Fallback sang Gemini

Chain với error handling
from langchain_core.runnables import with_fallbacks

robust_chain = primary_chain.with_fallbacks(
    fallbacks=[fallback_chain],
    exception_handler=lambda e, *args: logger.error(f"Chain failed: {e}")
)

Sử dụng với error tracking
try:
    result = robust_chain.invoke({"input": "Giải thích LangChain LCEL"})
    print(f"Success: {result[:100]}...")
except Exception as e:
    logger.critical(f"Chain failed after all fallbacks: {e}")

4. Caching và Memory Optimization

Một trong những bài học đắt giá nhất của tôi: đừng gọi LLM khi không cần thiết. Caching có thể giảm 60-80% chi phí API.

from langchain.globals import set_llm_cache
from langchain.cache import InMemoryCache
from langchain_community.cache import SQLiteCache

In-memory cache cho development
set_llm_cache(InMemoryCache())

SQLite cache cho production (persistent)
set_llm_cache(SQLiteCache(database_path=".llm_cache.db"))

Prompt caching chain - lưu intermediate results
from langchain_core.runnables import RunnablePassthrough

cache_key_template = "{input}:{model}"

LLM với caching tự động
cached_llm = llm.bind(cache=True)

Cache warming - preload common queries
warm_up_queries = [
    "Giải thích LangChain LCEL",
    "Cách sử dụng RAG",
    "Vector database là gì"
]

print("Warming up cache...")
for query in warm_up_queries:
    cached_llm.invoke(query)
    print(f"  Cached: {query}")

Semantic cache - cache dựa trên similarity
from difflib import SequenceMatcher

def semantic_similarity(a: str, b: str) -> float:
    """Tính similarity giữa 2 strings"""
    return SequenceMatcher(None, a.lower(), b.lower()).ratio()

class SemanticCache:
    """Semantic cache với similarity threshold"""
    def __init__(self, threshold: float = 0.85):
        self.cache = {}
        self.threshold = threshold
    
    def get(self, query: str):
        for cached_query, result in self.cache.items():
            if semantic_similarity(query, cached_query) >= self.threshold:
                return result
        return None
    
    def set(self, query: str, result: str):
        self.cache[query] = result

semantic_cache = SemanticCache(threshold=0.9)

def cached_chain(query: str):
    # Check cache
    cached = semantic_cache.get(query)
    if cached:
        return f"[CACHED] {cached}"
    
    # Call LLM
    result = llm.invoke(query).content
    semantic_cache.set(query, result)
    return result

Benchmark cache performance
import time

queries = ["LangChain là gì", "Giải thích LCEL", "Vector embedding hoạt động thế nào"]

Without cache
start = time.time()
for q in queries * 3:
    llm.invoke(q)
no_cache_time = time.time() - start

With semantic cache
start = time.time()
for q in queries * 3:
    cached_chain(q)
cache_time = time.time() - start

print(f"Không cache: {no_cache_time:.3f}s")
print(f"Có cache: {cache_time:.3f}s")
print(f"Tiết kiệm: {(1 - cache_time/no_cache_time)*100:.1f}%")

5. Concurrency Control - Kiểm Soát Đồng Thời

Đây là phần nâng cao mà nhiều kỹ sư bỏ qua. Khi bạn có hàng nghìn concurrent requests, không kiểm soát được concurrency sẽ dẫn đến:

• Rate limit errors (429 Too Many Requests)
• Memory exhaustion
• Unexpected timeout chains

import asyncio
from concurrent.futures import ThreadPoolExecutor
from threading import Semaphore
import asyncio

Semaphore cho concurrency control
MAX_CONCURRENT_REQUESTS = Semaphore(10)  # Tối đa 10 requests đồng thời

def limited_invoke(chain, input_dict):
    """Invoke chain với giới hạn concurrency"""
    with MAX_CONCURRENT_REQUESTS:
        return chain.invoke(input_dict)

Async chain execution với rate limiting
class RateLimitedLLM:
    """LLM wrapper với rate limiting"""
    def __init__(self, max_rpm: int = 60):
        self.semaphore = asyncio.Semaphore(max_rpm // 10)  # requests per second
        self.llm = llm
    
    async def ainvoke(self, prompt: str):
        async with self.semaphore:
            # Chuyển sync call thành async
            loop = asyncio.get_event_loop()
            return await loop.run_in_executor(None, self.llm.invoke, prompt)

Batch processing với controlled concurrency
async def process_batch_async(queries: list[str], batch_size: int = 5):
    """Process queries với batching và concurrency control"""
    rate_limited_llm = RateLimitedLLM(max_rpm=60)
    results = []
    
    for i in range(0, len(queries), batch_size):
        batch = queries[i:i + batch_size]
        print(f"Processing batch {i//batch_size + 1}: {len(batch)} queries")
        
        # Chạy batch với concurrency limit
        batch_tasks = [rate_limited_llm.ainvoke(q) for q in batch]
        batch_results = await asyncio.gather(*batch_tasks, return_exceptions=True)
        
        results.extend(batch_results)
        
        # Rate limit delay giữa batches
        await asyncio.sleep(1.0)
    
    return results

Sync version với ThreadPoolExecutor
def process_batch_sync(queries: list[str], max_workers: int = 10):
    """Process queries với ThreadPoolExecutor"""
    chain = llm | StrOutputParser()
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # Submit all tasks with semaphore
        futures = []
        for query in queries:
            future = executor.submit(limited_invoke, chain, {"input": query})
            futures.append(future)
        
        # Collect results
        results = [f.result() for f in futures]
    
    return results

Benchmark
test_queries = [f"Query {i}: Giải thích concept {i}" for i in range(50)]

print("Testing async batch processing...")
start = time.time()
async_results = asyncio.run(process_batch_async(test_queries, batch_size=10))
async_time = time.time() - start

print("Testing sync batch processing...")
start = time.time()
sync_results = process_batch_sync(test_queries, max_workers=10)
sync_time = time.time() - start

print(f"Async batch: {async_time:.2f}s ({len(test_queries)/async_time:.1f} req/s)")
print(f"Sync thread: {sync_time:.2f}s ({len(test_queries)/sync_time:.1f} req/s)")

6. Cost Optimization - Chiến Lược Tiết Kiệm

Sau khi xây dựng hệ thống cho nhiều enterprise clients, tôi đã tổng hợp các chiến lược tiết kiệm chi phí hiệu quả nhất:

6.1 Model Routing Thông Minh

# Chi phí thực tế tại HolySheep AI (2026)
MODEL_COSTS = {
    "deepseek-v3.2": 0.42,   # $0.42/MTok - Cho simple tasks
    "gemini-2.5-flash": 2.50, # $2.50/MTok - Balance speed/cost
    "claude-sonnet-4.5": 15.0, # $15/MTok - Complex reasoning
    "gpt-4.1": 8.0           # $8/MTok - General purpose
}

def select_model_by_complexity(task: str, query: str) -> str:
    """Chọn model phù hợp với độ phức tạp của task"""
    simple_keywords = ["liệt kê", "đếm", "tìm", "kiểm tra", "simple", "list"]
    complex_keywords = ["phân tích", "so sánh", "đánh giá", "reasoning", "analyze"]
    
    query_lower = query.lower()
    
    if any(kw in query_lower for kw in simple_keywords):
        return "deepseek-v3.2"  # Rẻ nhất, đủ dùng
    elif any(kw in query_lower for kw in complex_keywords):
        return "gemini-2.5-flash"  # Nhanh hơn Claude, rẻ hơn GPT
    else:
        return "deepseek-v3.2"  # Default

def calculate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    """Tính chi phí dựa trên token usage"""
    input_cost = (input_tokens / 1_000_000) * MODEL_COSTS[model]
    output_cost = (output_tokens / 1_000_000) * MODEL_COSTS[model]
    return input_cost + output_cost

class CostAwareChain:
    """Chain với automatic model selection và cost tracking"""
    def __init__(self):
        self.total_cost = 0.0
        self.total_tokens = 0
        self.models_used = {}
    
    def invoke(self, query: str):
        model = select_model_by_complexity("auto", query)
        
        # Invoke với model được chọn
        chain = ChatPromptTemplate.from_messages([
            ("system", "Trả lời ngắn gọn, chính xác."),
            ("human", "{input}")
        ]) | ChatOpenAI(model=model, api_key="YOUR_HOLYSHEEP_API_KEY")
        
        result = chain.invoke({"input": query})
        
        # Track usage (trong thực tế, lấy từ response metadata)
        estimated_tokens = len(query.split()) * 2 + 100
        cost = calculate_cost(model, estimated_tokens, 50)
        
        self.total_cost += cost
        self.total_tokens += estimated_tokens
        self.models_used[model] = self.models_used.get(model, 0) + 1
        
        return result.content
    
    def get_cost_report(self):
        return {
            "total_cost_usd": self.total_cost,
            "total_tokens": self.total_tokens,
            "cost_per_1k_tokens": (self.total_cost / self.total_tokens * 1000) if self.total_tokens > 0 else 0,
            "model_distribution": self.models_used
        }

Demo cost optimization
cost_chain = CostAwareChain()

test_queries = [
    "Liệt kê 5 tính năng của LangChain",
    "Phân tích ưu nhược điểm của RAG vs Fine-tuning",
    "Tìm file config trong project"
]

for q in test_queries:
    result = cost_chain.invoke(q)
    print(f"Query: {q[:30]}... -> Cost tracked")

report = cost_chain.get_cost_report()
print(f"\n=== Cost Report ===")
print(f"Tổng chi phí: ${report['total_cost_usd']:.4f}")
print(f"Tokens: {report['total_tokens']}")
print(f"Cost/1K tokens: ${report['cost_per_1k_tokens']:.4f}")
print(f"Model distribution: {report['model_distribution']}")

6.2 Token Budget Management

from dataclasses import dataclass
from typing import Optional

@dataclass
class TokenBudget:
    """Quản lý token budget cho chain"""
    max_tokens: int
    warning_threshold: float = 0.8  # Cảnh báo khi dùng 80%
    
    def __post_init__(self):
        self.used_tokens = 0
    
    def allocate(self, tokens: int, chain_name: str) -> Optional[int]:
        """Allocate tokens cho chain, return None nếu vượt budget"""
        if self.used_tokens + tokens > self.max_tokens:
            return None  # Budget exhausted
        
        self.used_tokens += tokens
        remaining = self.max_tokens - self.used_tokens
        
        if remaining < self.max_tokens * (1 - self.warning_threshold):
            print(f"⚠️ Warning: Budget còn {remaining} tokens ({remaining/self.max_tokens*100:.1f}%)")
        
        return tokens
    
    def reset(self):
        self.used_tokens = 0

class BudgetAwareChain:
    """Chain với token budget management"""
    def __init__(self, budget: TokenBudget):
        self.budget = budget
        self.chains = {}
    
    def add_chain(self, name: str, chain, estimated_tokens: int):
        self.chains[name] = {"chain": chain, "tokens": estimated_tokens}
    
    def invoke(self, query: str, chain_name: str = "default") -> dict:
        if chain_name not in self.chains:
            raise ValueError(f"Chain '{chain_name}' not found")
        
        chain_info = self.chains[chain_name]
        tokens = self.budget.allocate(chain_info["tokens"], chain_name)
        
        if tokens is None:
            return {
                "error": "Token budget exhausted",
                "budget_remaining": self.budget.max_tokens - self.budget.used_tokens
            }
        
        result = chain_info["chain"].invoke({"input": query})
        
        return {
            "result": result,
            "tokens_used": tokens,
            "budget_remaining": self.budget.max_tokens - self.budget.used_tokens
        }

Sử dụng
budget = TokenBudget(max_tokens=100000)

analysis_chain = ChatPromptTemplate.from_messages([
    ("system", "Phân tích chi tiết."),
    ("human", "{input}")
]) | llm

summary_chain = ChatPromptTemplate.from_messages([
    ("system", "Tóm tắt ngắn gọn."),
    ("human", "{input}")
]) | llm

budget_chain = BudgetAwareChain(budget)
budget_chain.add_chain("analysis", analysis_chain, estimated_tokens=2000)
budget_chain.add_chain("summary", summary_chain, estimated_tokens=500)

Invoke với budget control
for i in range(60):
    result = budget_chain.invoke(f"Task {i}: Phân tích dữ liệu", chain_name="analysis")
    if "error" in result:
        print(f"Task {i} failed: {result['error']}")
        break

print(f"\nFinal budget report: {budget.used_tokens}/{budget.max_tokens} tokens used")

Lỗi Thường Gặp và Cách Khắc Phục

1. Lỗi "RunnableNotCallable" - Chain Component Không Tương Thích

# ❌ SAI: Kết hợp chain với object không phải Runnable
from langchain_core.runnables import RunnableLambda

Lỗi thường gặp
try:
    chain = prompt | llm | some_function  # some_function là plain Python function
except Exception as e:
    print(f"Lỗi: {type(e).__name__}: {e}")

✅ ĐÚNG: Wrap plain function bằng RunnableLambda
def transform_output(output: str) -> str:
    """Transform output - nhưng đây là plain function"""
    return output.upper().strip()

Cách 1: Dùng RunnableLambda
chain = prompt | llm | StrOutputParser() | RunnableLambda(transform_output)

Cách 2: Dùng lambda
chain = prompt | llm | StrOutputParser() | (lambda x: x.upper())

Cách 3: Dùng itemgetter cho dict extraction
from operator import itemgetter
chain = prompt | llm | itemgetter("content") | RunnableLambda(transform_output)

Kiểm tra chain có invoke được không
print(f"Chain is runnable: {hasattr(chain, 'invoke')}")
result = chain.invoke({"input": "Hello world"})
print(f"Result: {result}")

2. Lỗi "OutputParserNotCallable" - Parser Không Nhận Input Đúng

from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel

❌ SAI: JsonOutputParser cần được configure với schema trước
try:
    chain = prompt | llm | JsonOutputParser()  # Chưa set schema
except Exception as e:
    print(f"Lỗi: {type(e).__name__}")

✅ ĐÚNG: Định nghĩa schema và bind vào parser
class ResponseModel(BaseModel):
    answer: str
    confidence: float
    sources: list[str]

parser = JsonOutputParser(pydantic_object=ResponseModel)

chain = prompt | llm | parser

Test
result = chain.invoke({"input": "Trả lời câu hỏi về AI"})
print(f"Parsed result: {result}")
print(f"Type: {type(result)}")

Alternative: Không dùng structured output, parse thủ công
def extract_json(text: str) -> dict:
    """Extract JSON từ text response"""
    import json
    import re
    # Tìm JSON trong text
    json_match = re.search(r'\{.*\}', text, re.DOTALL)
    if json_match:
        return json.loads(json_match.group())
    return {"raw": text}

chain_fallback = prompt | llm | StrOutputParser() | RunnableLambda(extract_json)
result = chain_fallback.invoke({"input": "Trả lời câu hỏi về AI"})
print(f"Fallback result: {result}")

3. Lỗi "ConcurrencyDeadlock" - Async/Sync Mixing

import asyncio
from langchain_core.runnables import RunnableLambda

❌ SAI: Gọi async chain từ sync context
async def async_chain(input: str) -> str:
    chain = ChatPromptTemplate.from_messages([
        ("human", "{input}")
    ]) | llm
    return await chain.ainvoke({"input": input})

Lỗi: Blocking trong async
try:
    # Điều này sẽ gây deadlock nếu chain có async operations
    result = asyncio.run(async_chain("test"))  # OK nhưng không an toàn
except Exception as e:
    print(f"Lỗi: {e}")

✅ ĐÚNG: Quản lý event loop đúng cách
def sync_wrapper(async_chain_func, input: str) -> str:
    """Wrapper để gọi async chain từ sync code"""
    try:
        loop = asyncio.get_event_loop()
        if loop.is_running():
            # Nếu đã có loop đang chạy, tạo task mới
            import concurrent.futures
            with concurrent.futures.ThreadPoolExecutor() as pool:
                future = pool.submit(asyncio.run, async_chain_func(input))
                return future.result()
        else:
            return asyncio.run(async_chain_func(input))
    except RuntimeError as e:
        # Fallback: tạo new event loop
        return asyncio.run(async_chain_func(input))

Best practice: Dùng async throughout
async def async_pipeline(inputs: list[str]) -> list[str]:
    """Async pipeline - cách tốt nhất"""
    chain = ChatPromptTemplate.from_messages([
        ("system", "Xử lý: {input}"),
        ("human", "{input}")
    ]) | llm
    
    # Chạy tất cả requests song song
    tasks = [chain.ainvoke({"input": inp}) for inp in inputs]
    results = await asyncio.gather(*tasks)
    return [r.content for r in results]

Test async pipeline
async def main():
    test_inputs = ["Query 1", "Query 2", "Query 3"]
    results = await async_pipeline(test_inputs)
    for i, r in enumerate(results):
        print(f"Result {i+1}: {r[:50]}...")

asyncio.run(main())

4. Lỗi "RateLimitExceeded" - Không Xử Lý Rate Limit

from time import sleep
from collections import defaultdict

❌ SAI: Không có rate limit handling
def bad_chain_batch(queries: list[str]):
    results = []
    for q in queries:
        result = llm.invoke(q)  # Không handle rate limit
        results.append(result)
    return results

✅ ĐÚNG: Exponential backoff với jitter
class RateLimitHandler:
    """Handler rate limit với exponential backoff"""
    def __init__(self, max_retries: int = 5):
        self.max_retries = max_retries
        self.retry_counts = defaultdict(int)
    
    def call_with_retry(self, func, *args, **kwargs):
        """Gọi function với retry tự động khi gặp rate limit"""
        import random
        
        for attempt in range(self.max_retries):
            try:
                result = func(*args, **kwargs)
                self.retry_counts[func.__name__] = 0  # Reset on success
                return result
            except Exception as e:
                error_str = str(e).lower()
                if "rate limit" in error_str or "429" in error_str:
                    self.retry_counts[func.__name__] += 1
                    wait_time = (2 ** attempt) + random.uniform(0, 1)
                    print(f"Rate limited. Retry {attempt+1}/{self.max_retries} sau {wait_time:.1f}s")
                    sleep(wait_time)
                else:
                    raise  # Re-raise non-rate-limit errors
        
        raise Exception(f"Max retries ({self.max_retries}) exceeded")

Sử dụng handler
handler = RateLimitHandler(max_retries=5)

def safe_llm_call(prompt: str):
    """Wrapper cho LLM call với rate limit handling"""
    return llm.invoke(prompt).
Tài nguyên liên quan
📚 Hướng dẫn AI API
💰 Xem giá
📖 Tài liệu nhà phát triển
🚀 Đăng ký miễn phí
Bài viết liên quan
2026 Tháng Tư — Bảng Xếp Hạng Độ Hoàn Thiện Tài Liệu API AI
So Sánh Chi Phí AI API Tháng 4/2026: DeepSeek Rẻ Gấp 19 Lần 
Cursor AI 多代理模式配置与使用 - Hướng Dẫn Toàn Diện Từ A-Z