xDev Asia

Implement vòng lặp agent hoàn chỉnh từ đầu bằng Python thuần: ReAct pattern, xử lý multi-step reasoning, conversation history management, token budget, và stopping conditions.

Build AI Agents: Từ Zero đến Production

Phần 2: Function Calling & Tool Use

xdev.asia

Giới thiệu

Đây là bài quan trọng nhất của series — bạn sẽ implement agent loop hoàn chỉnh từ đầu, không dùng bất kỳ framework nào. Hiểu sâu vòng lặp này sẽ giúp bạn debug và tối ưu agent hiệu quả hơn khi dùng framework sau này.

1. The Agent Loop — Core Architecture

class SimpleAgent:
    def __init__(self, model="gpt-4o-mini", max_steps=10):
        self.client = OpenAI()
        self.model = model
        self.max_steps = max_steps
        self.tools = ToolRegistry()
        self.messages = []
        self.cost_tracker = CostTracker()
    
    def run(self, user_input: str) -> str:
        self.messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": user_input}
        ]
        
        for step in range(self.max_steps):
            response = self.client.chat.completions.create(
                model=self.model,
                messages=self.messages,
                tools=self.tools.schemas,
            )
            
            message = response.choices[0].message
            self.messages.append(message)
            self.cost_tracker.track(self.model, response.usage)
            
            # No tool calls → final answer
            if not message.tool_calls:
                return message.content
            
            # Execute tools
            for tool_call in message.tool_calls:
                result = self.tools.execute(
                    tool_call.function.name,
                    json.loads(tool_call.function.arguments)
                )
                self.messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": str(result),
                })
        
        return "Max steps reached without final answer."

2. Token Budget Management

def trim_messages(messages, max_tokens=100000):
    """Giữ messages trong budget token."""
    # Luôn giữ system prompt và user message gần nhất
    essential = [messages[0], messages[-1]]
    middle = messages[1:-1]
    
    # Xóa messages cũ nhất nếu vượt budget
    while estimate_tokens(essential + middle) > max_tokens and middle:
        middle.pop(0)
    
    return [essential[0]] + middle + [essential[1]]

3. Stopping Conditions

Agent cần biết khi nào dừng:

LLM tự quyết định dừng (không gọi thêm tool)
Max steps reached — safety net
Token budget exceeded — tránh chi phí bất ngờ
Error threshold — quá nhiều tool failures liên tiếp
User interrupt — human-in-the-loop

Tóm tắt

Agent loop = while loop gọi LLM → check tool_calls → execute → repeat
Token budget management ngăn agent "phát hoang"
Multiple stopping conditions cho safety
Logging mọi step để debug

Bài tập