Agentic Workflows: LLMs Chạy Multi-step Pipelines – Credit Assignment, Failure Recovery, Human Oversight

Chào anh em dev, mình là Hải đây. Hơn 12 năm lăn lộn với code, từ PHP thuần túy năm 2012 đến build microservices handle hàng triệu CCU, mình thấy AI giờ không còn là thứ “xịn sò” để khoe nữa. Nó đang lọt vào workflow hàng ngày, đặc biệt là với Large Language Models (LLMs – Mô hình ngôn ngữ lớn, những con AI như GPT hay Llama có khả năng xử lý text tự nhiên). Hôm nay, mình sẽ deep dive vào Agentic Workflows – một cách tiếp cận để LLMs không chỉ trả lời câu hỏi đơn lẻ mà chạy cả pipeline đa bước, tự quyết định hành động như một “agent” thông minh.

Mình chọn góc nhìn này vì chủ đề đòi hỏi đào sâu under the hood: cơ chế hoạt động của các agent, cách chúng phân bổ trách nhiệm, xử lý lỗi, và giữ chỗ cho con người can thiệp. Không phải lý thuyết suông, mình sẽ lôi code mẫu, số liệu thực tế, và so sánh để anh em thấy rõ cái gì đang xảy ra bên dưới. Nếu anh em đang build hệ thống automation với AI, bài này sẽ giúp tránh mấy cái bẫy phổ biến như pipeline treo cứng hay credit attribution sai lệch.

Agentic Workflows Là Gì? Deep Dive Vào Bản Chất

Trước hết, giải thích Agentic Workflows (Luồng công việc dựa trên agent). Đây không phải là script tự động hóa đơn giản kiểu cron job. Thay vào đó, nó là hệ thống nơi LLMs đóng vai trò “agent” – một thực thể tự chủ, có khả năng nhận nhiệm vụ, phân tích, chọn tool (công cụ bên ngoài như API search, database query), và thực thi multi-step (đa bước). Khác với prompt engineering cơ bản, agentic workflows cho phép LLMs tự loop qua các bước dựa trên output của bước trước, giống như một dev junior đang debug code nhưng với tốc độ AI.

Under the hood, cơ chế cốt lõi là ReAct pattern (Reasoning + Acting): LLM suy nghĩ (reason) về bước tiếp theo, hành động (act) bằng cách gọi tool, quan sát kết quả (observe), rồi lặp lại. Ví dụ, nhiệm vụ “tìm thông tin về stock giá Apple” không chỉ generate text mà có thể: (1) search web, (2) parse data từ API, (3) tính toán trend, (4) generate report. Điều này dựa trên transformer architecture của LLMs, nơi attention mechanism giúp model “nhớ” context qua các bước.

Theo docs chính thức của LangChain (một framework phổ biến cho agentic systems), agentic workflows giải quyết vấn đề hallucination (ảo tưởng – LLM bịa thông tin) bằng cách ground output vào real-world tools. GitHub repo của LangChain có hơn 80k stars (tính đến 2024), chứng tỏ cộng đồng đang đổ xô vào đây. Nhưng deep hơn, vấn đề lớn là scalability: với pipeline dài 10-20 bước, context window của LLM (ví dụ GPT-4o có 128k tokens) dễ overflow, dẫn đến latency tăng từ 200ms/step lên 2s/step nếu không optimize.

⚡ Lưu ý: Trong Python 3.12, dùng async/await để handle multi-agent parallel execution, tránh blocking I/O khi gọi external APIs.

Multi-step Pipelines Với LLMs: Xây Dựng Và Thách Thức

Bây giờ đi sâu vào multi-step pipelines. Một pipeline điển hình gồm: input parsing → planning → execution → validation. LLMs ở đây không chỉ generate mà còn orchestrate (điều phối) – ví dụ dùng LLM router để quyết định agent nào handle bước nào.

Hãy tưởng tượng use case kỹ thuật: Xử lý dữ liệu Big Data 50GB từ log files, nơi cần (1) extract patterns bằng regex, (2) summarize bằng LLM, (3) anomaly detection qua ML model, (4) alert nếu phát hiện issue. Không agentic, bạn phải hard-code sequence – dễ fail nếu data thay đổi. Với agentic, LLM tự adapt: nếu bước 2 fail do rate limit API, nó retry hoặc switch tool.

Dưới đây là code mẫu đơn giản dùng LangChain (version 0.1.0+) trong Python 3.12. Giả sử build một pipeline để query weather data rồi generate travel advice.

from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain.tools import Tool
from langchain.prompts import PromptTemplate
import asyncio
import openai  # Phiên bản 1.12.0

# Define tools
def get_weather(city: str) -> str:
    # Simulate API call to OpenWeatherMap
    return f"Weather in {city}: Sunny, 25°C"  # Thực tế gọi API async

weather_tool = Tool(
    name="WeatherChecker",
    description="Get current weather for a city",
    func=get_weather
)

# LLM setup (use your API key)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.1, api_key="your-key")

# ReAct prompt template
prompt = PromptTemplate.from_template("""
You are a travel agent. Use tools to answer: {input}
{agent_scratchpad}
""")

# Create agent
agent = create_react_agent(llm, [weather_tool], prompt)
agent_executor = AgentExecutor(agent=agent, tools=[weather_tool], verbose=True, max_iterations=5)

# Async execution for multi-step
async def run_pipeline(query: str):
    result = await asyncio.to_thread(agent_executor.invoke, {"input": query})
    return result['output']

# Usage
async def main():
    advice = await run_pipeline("What's the weather in Hanoi and suggest outfit?")
    print(advice)  # Output: "Weather in Hanoi: Sunny, 25°C. Suggest light shirt and shorts."

asyncio.run(main())

Code này chạy multi-step: Agent reason “I need weather first”, act bằng tool, observe, rồi generate advice. Trong thực tế, với 10k queries/giây, dùng Redis 7.2 để cache tool outputs, giảm latency từ 500ms xuống 120ms (dựa trên benchmark từ LangChain docs).

Thách thức under the hood: Token consumption. Mỗi step thêm context, có thể burn 10k tokens/pipeline. Giải pháp: Use summarization chains để compress history, giữ context dưới 4k tokens.

Credit Assignment: Gán Tín Dụng Cho Các Bước Đúng

Credit assignment (Phân bổ tín dụng) là vấn đề cốt lõi trong agentic workflows – giống như trong RL (Reinforcement Learning – Học tăng cường), nơi bạn reward action nào dẫn đến success. Với LLMs, không có explicit reward signal, nên phải infer từ outcome.

Deep dive: Sử dụng trajectory logging (ghi log đường đi của pipeline). Mỗi step được tag với metadata: input tokens, output quality score (dùng LLM-as-judge, ví dụ rate response từ 1-10). Sau khi pipeline hoàn thành, backpropagate credit: Nếu final output tốt (ví dụ accuracy >90% trên benchmark), assign +1 credit cho các step contributing (dựa trên dependency graph).

Ví dụ use case kỹ thuật: Khi hệ thống đạt 10.000 user/giây phân tích sentiment từ social media streams. Nếu pipeline fail ở bước 3/5 (parsing JSON error), credit assignment giúp identify: Bước 1 (data fetch) chỉ đáng 20% credit nếu nó clean data kém, dẫn đến downstream failure.

Code minh họa credit tracking với custom callback in LangChain:

from langchain.callbacks import BaseCallbackHandler
from typing import Dict, Any

class CreditTracker(BaseCallbackHandler):
    def __init__(self):
        self.steps: Dict[str, float] = {}
        self.total_credit = 0.0

    def on_agent_action(self, action, **kwargs):
        step_id = f"step_{len(self.steps)}"
        self.steps[step_id] = {"action": action.log, "credit": 0.0}
        print(f"Tracking step: {step_id}")

    def on_chain_end(self, outputs, **kwargs):
        # Final evaluation
        quality = self.evaluate_output(outputs["output"])  # Custom scorer
        self.assign_credit(quality)

    def evaluate_output(self, text: str) -> float:
        # Use another LLM to score, e.g., via OpenAI
        response = openai.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"Score this from 0-10: {text}"}]
        )
        return float(response.choices[0].message.content)

    def assign_credit(self, quality: float):
        self.total_credit = quality / 10  # Normalize
        num_steps = len(self.steps)
        base_credit = self.total_credit / num_steps
        for step in self.steps.values():
            step["credit"] = base_credit  # Simple uniform; real: use graph algo

# Integrate
tracker = CreditTracker()
agent_executor = AgentExecutor(..., callbacks=[tracker])

Từ StackOverflow Survey 2024, 45% dev AI gặp khó với attribution in multi-agent systems. Giải pháp advanced: Sử dụng Bayesian inference để model probability contribution, như trong paper “Credit Assignment in Sequential Decision Making” từ NeurIPS 2023.

Best Practice: Luôn log trajectories vào PostgreSQL 16 với TimescaleDB extension cho time-series data. Query credit như: SELECT step_id, AVG(credit) FROM credits WHERE pipeline_id = 'uuid' GROUP BY step_id;. Giảm storage 30% so với flat JSON logs.

Failure Recovery: Xử Lý Lỗi Không Để Pipeline Chết Yểu

Failure recovery (Khôi phục lỗi) là phần under the hood thú vị nhất. LLMs deterministic kém (do temperature >0), nên step có thể fail vì: API timeout (504 Gateway Time-out), invalid output (e.g., LLM generate non-JSON), hoặc deadlock ở shared resources.

Cơ chế: Retry with backoff + Fallback agents. Deep dive: Sử dụng exponential backoff (e.g., retry sau 1s, 2s, 4s) cho transient errors. Nếu permanent (e.g., OOM error ở GPU), switch sang smaller model như Llama-3-8B thay vì GPT-4.

Use case: Xử lý Big Data 50GB với pipeline extract-transform-load (ETL). Nếu bước transform fail do LLM parse sai schema (error rate 15% theo benchmark Uber Engineering Blog 2023), recovery loop: Re-prompt với error context, hoặc escalate to human.

Code mẫu cho recovery:

import time
from tenacity import retry, stop_after_attempt, wait_exponential  # tenacity 8.2.3

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10))
async def robust_llm_call(llm, prompt: str, error_context: str = "") -> str:
    try:
        response = await llm.ainvoke(prompt + error_context)
        if not is_valid_json(response.content):  # Custom validator
            raise ValueError("Invalid output format")
        return response.content
    except Exception as e:
        error_msg = f"Error: {str(e)}. Retrying..."
        print(error_msg)
        raise  # Let tenacity handle

# In pipeline
try:
    output = await robust_llm_call(llm, "Parse this JSON: {...}", "")
except Exception as final_e:
    # Fallback: Use rule-based parser
    output = fallback_parse(data)
    print(f"Recovered via fallback: {final_e}")

Benchmark từ Netflix Tech Blog (2024): Với retry logic này, recovery rate lên 92%, giảm downtime từ 5% xuống 0.8% ở production pipelines.

🐛 Warning: Tránh infinite loops bằng max_iterations (e.g., 10). Trong Node.js 20, dùng p-limit để throttle concurrent recoveries, tránh overwhelm API quotas.

Human Oversight: Giữ Con Người Trong Vòng Lập

Human oversight (Giám sát con người) đảm bảo agent không “làm bừa”. Deep dive: Guardrails như approval gates ở critical steps (e.g., trước khi delete data). Sử dụng human-in-the-loop (HITL): Pipeline pause, notify via Slack/Webhook, chờ input.

Cơ chế: Embed metadata cho traceability – mỗi decision có rationale từ LLM. Tool như Arize AI (open-source parts) trace lineage.

Use case: Hệ thống 10k user/giây approve transactions. Agent handle initial checks, nhưng oversight layer flag suspicious (score >0.8) cho human review, giảm false positives 40% (dữ liệu từ Meta AI Blog 2024).

🛡️ Best Practice: Implement RBAC (Role-Based Access Control) với Keycloak 22. Trong Python, dùng FastAPI 0.104 cho oversight endpoints: POST /review/{step_id} với JSON payload cho human feedback, feed back vào credit assignment.

So Sánh Các Framework Cho Agentic Workflows

Để build multi-step pipelines, anh em có nhiều lựa chọn. Dưới đây bảng so sánh dựa trên tiêu chí thực tế (dùng Python 3.12 ecosystem).

LangChain thắng về cộng đồng, nhưng AutoGen tốt hơn cho credit assignment nhờ built-in evaluation. Theo StackOverflow Survey 2024, 62% dev chọn LangChain cho production do integration rộng (e.g., với PostgreSQL 16 vector extensions cho semantic search).

Kết Luận: Những Gì Anh Em Nên Nhớ

Tóm lại 3 key takeaways từ deep dive này:

1. Agentic workflows biến LLMs thành orchestrators thực thụ, với ReAct pattern giúp multi-step tự adapt, nhưng phải track context để tránh token bùng nổ.
2. Credit assignment và failure recovery là chìa khóa scalability, dùng logging và retry để assign reward chính xác, đạt recovery rate >90%.
3. Human oversight giữ hệ thống an toàn, integrate HITL để balance autonomy và control, đặc biệt ở high-stakes use cases như data processing 50GB.

Anh em đã từng build agentic pipeline nào chưa? Gặp khó nhất ở phần recovery hay credit? Comment bên dưới, mình đọc và chém gió thêm.

Nếu anh em đang cần tích hợp AI nhanh vào app mà lười build từ đầu, thử ngó qua con Serimi App xem, mình thấy API bên đó khá ổn cho việc scale.