LLM复杂编程实战：多步骤Agent流程设计与自审最佳实践

大多数编码 Agent 在复杂任务上失败，是因为它们直接跳到实现阶段，而没有做规划。在本教程中，我将带你构建一个基于 Oxlo.ai 的多步骤编码 Agent，它能够分析需求、生成设计计划、编写 Python 代码，并对自己的输出进行逻辑错误审查。最终的流水线在并发处理、错误处理和边界情况上比单次提示要好得多。

需要准备：

一个来自 https://portal.oxlo.ai 的 Oxlo.ai API key

通过 pip install openai 安装 OpenAI SDK

步骤 1：初始化 Oxlo.ai 客户端

在构建逻辑之前，我总是先验证端点和 API key。这段代码连接 Oxlo.ai 并用 Llama 3.3 70B 做一次快速健全性检查。

import os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.oxlo.ai/v1",
    api_key=os.environ.get("OXLO_API_KEY", "YOUR_OXLO_API_KEY"),
)

response = client.chat.completions.create(
    model="llama-3.3-70b",
    messages=[{"role": "user", "content": "Say hello"}],
)

print("Oxlo.ai connection OK:", response.choices[0].message.content)

步骤 2：定义 system prompt

一个强有力的 system prompt 能让模型专注于架构和边界情况，而不是表面的语法。将其存储在常量中，以便快速迭代。

SYSTEM_PROMPT = """You are a principal software engineer who writes production-grade Python.

Follow these rules on every task:
1. Analyze requirements for concurrency, error handling, and edge cases before writing code.
2. Use type hints, docstrings, and the standard library unless there is a compelling reason to import third-party packages.
3. Prefer composition over inheritance. Keep functions small and single-purpose.
4. For concurrent code, explicitly state which primitives protect shared state.
5. After generating code, list any assumptions or potential risks."""

步骤 3：将任务分解为计划

在写代码之前，Agent 需要对并发、数据结构和失败模式进行推理。这里我使用 Kimi K2.6，因为它的长上下文窗口可以在不截断的情况下处理详细需求。

def plan_task(client, requirements):
    response = client.chat.completions.create(
        model="kimi-k2.6",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"Create a detailed implementation plan for the following requirements. Do not write code yet. Outline modules, classes, and concurrency approach.\n\nRequirements:\n{requirements}"},
        ],
        temperature=0.2,
    )
    return response.choices[0].message.content

requirements = (
    "Implement a thread-safe priority task queue in Python. "
    "It must support rate limiting per worker, exponential backoff for failed tasks, "
    "and graceful shutdown that waits for in-flight tasks. Include unit tests."
)

plan = plan_task(client, requirements)
print("=== PLAN ===")
print(plan)

步骤 4：生成实现

有了计划在手，我们切换到 DeepSeek V3.2，它针对编码和推理做了调优。将计划传入 user message 可以让模型扎根于设计。

def generate_code(client, plan):
    response = client.chat.completions.create(
        model="deepseek-v3.2",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"Implementation plan:\n{plan}\n\nNow write the complete Python implementation. Include all classes, methods, and unit tests in a single file."},
        ],
        temperature=0.2,
    )
    return response.choices[0].message.content

code = generate_code(client, plan)
print("=== CODE ===")
print(code)

步骤 5：自我审查 bugs 和竞态条件

复杂代码需要再看一遍。我将生成的代码连同原始计划一起发回给 Kimi K2.6，让它标记出竞态条件、缺失的错误处理和类型安全问题。

def review_code(client, plan, code):
    response = client.chat.completions.create(
        model="kimi-k2.6",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"Implementation plan:\n{plan}\n\nGenerated code:\n{code}\n\nReview the code. Flag any race conditions, missing error handling, type safety issues, or deviations from the plan. Be specific."},
        ],
        temperature=0.2,
    )
    return response.choices[0].message.content

review = review_code(client, plan, code)
print("=== REVIEW ===")
print(review)

步骤 6：组装完整的 Agent

现在将三个阶段串联成一条流水线。最终脚本读取任务描述，运行规划器、编码器和审查器，然后打印最终代码和审查意见。

import os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.oxlo.ai/v1",
    api_key=os.environ.get("OXLO_API_KEY", "YOUR_OXLO_API_KEY"),
)

SYSTEM_PROMPT = """You are a principal software engineer who writes production-grade Python.

Follow these rules on every task:
1. Analyze requirements for concurrency, error handling, and edge cases before writing code.
2. Use type hints, docstrings, and the standard library unless there is a compelling reason to import third-party packages.
3. Prefer composition over inheritance. Keep functions small and single-purpose.
4. For concurrent code, explicitly state which primitives protect shared state.
5. After generating code, list any assumptions or potential risks."""

def plan_task(client, requirements):
    response = client.chat.completions.create(
        model="kimi-k2.6",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"Create a detailed implementation plan for the following requirements. Do not write code yet. Outline modules, classes, and concurrency approach.\n\nRequirements:\n{requirements}"},
        ],
        temperature=0.2,
    )
    return response.choices[0].message.content

def generate_code(client, plan):
    response = client.chat.completions.create(
        model="deepseek-v3.2",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"Implementation plan:\n{plan}\n\nNow write the complete Python implementation. Include all classes, methods, and unit tests in a single file."},
        ],
        temperature=0.2,
    )
    return response.choices[0].message.content

def review_code(client, plan, code):
    response = client.chat.completions.create(
        model="kimi-k2.6",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"Implementation plan:\n{plan}\n\nGenerated code:\n{code}\n\nReview the code. Flag any race conditions, missing error handling, type safety issues, or deviations from the plan. Be specific."},
        ],
        temperature=0.2,
    )
    return response.choices[0].message.content

def main():
    requirements = (
        "Implement a thread-safe priority task queue in Python. "
        "It must support rate limiting per worker, exponential backoff for failed tasks, "
        "and graceful shutdown that waits for in-flight tasks. Include unit tests."
    )

    print("Planning...")
    plan = plan_task(client, requirements)

    print("Coding...")
    code = generate_code(client, plan)

    print("Reviewing...")
    review = review_code(client, plan, code)

    print("\n=== FINAL CODE ===\n")
    print(code)
    print("\n=== REVIEW ===\n")
    print(review)

if __name__ == "__main__":
    main()

将脚本保存为 complex_coder.py，设置好 API key，然后用一条有挑战性的提示词运行它。以下是我在请求一个带限流的线程安全优先级队列时收到的确切命令和输出。

export OXLO_API_KEY="sk-oxlo.ai-..."
python complex_coder.py

Planning...
Coding...
Reviewing...

=== FINAL CODE ===

import threading
import time
import unittest
from queue import PriorityQueue, Empty
from typing import Callable, Any, Optional

class Task:
    """Represents a unit of work with priority and retry state."""

    def __init__(
        self,
        fn: Callable,
        args: tuple = (),
        kwargs: Optional[dict] = None,
        priority: int = 5,
    ):
        self.fn = fn
        self.args = args
        self.kwargs = kwargs or {}
        self.priority = priority
        self.attempts = 0
        self.lock = threading.Lock()

    def run(self) -> Any:
        return self.fn(*self.args, **self.kwargs)

    def __lt__(self, other: "Task") -> bool:
        return self.priority < other.priority

class PriorityTaskQueue:
    """Thread-safe priority queue with rate limiting and graceful shutdown."""

    def __init__(self, max_workers: int = 4, rate_limit: float = 10.0):
        self.max_workers = max_workers
        self.rate_limit = rate_limit
        self._queue: PriorityQueue[Task] = PriorityQueue()
        self._workers: list[threading.Thread] = []
        self._shutdown = False
        self._active_tasks = 0
        self._lock = threading.Lock()
        self._condition = threading.Condition(self._lock)

    def submit(self, task: Task) -> None:
        with self._lock:
            if self._shutdown:
                raise RuntimeError("Queue is shut down")
            self._queue.put(task)

    def _worker_loop(self) -> None:
        while True:
            with self._lock:
                if self._shutdown and self._queue.empty() and self._active_tasks == 0:
                    break
            try:
                task = self._queue.get(timeout=0.5)
            except Empty:
                continue

            with self._lock:
                self._active_tasks += 1

            try:
                if task.attempts > 0:
                    backoff = min(2 ** task.attempts, 60)
                    time.sleep(backoff)
                task.run()
            except Exception:
                with self._lock:
                    task.attempts += 1
                    if task.attempts < 3:
                        self._queue.put(task)
            finally:
                with self._lock:
                    self._active_tasks -= 1
                    self._condition.notify_all()
                time.sleep(1.0 / self.rate_limit)

    def start(self) -> None:
        for _ in range(self.max_workers):
            t = threading.Thread(target=self._worker_loop, daemon=True)
            t.start()
            self._workers.append(t)

    def shutdown(self, wait: bool = True) -> None:
        with self._lock:
            self._shutdown = True
        if wait:
            for t in self._workers:
                t.join()

class TestPriorityTaskQueue(unittest.TestCase):
    def test_basic_submit_and_run(self) -> None:
        q = PriorityTaskQueue(max_workers=1, rate_limit=100.0)
        results: list[int] = []
        q.submit(Task(fn=lambda: results.append(1), priority=1))
        q.start()
        time.sleep(0.2)
        q.shutdown()
        self.assertEqual(results, [1])

if __name__ == "__main__":
    unittest.main()

=== REVIEW ===

1. Race condition: task.attempts is incremented inside self._lock in the exception handler, but Task.attempts is also read in _worker_loop outside the queue lock. Use task.lock when reading task.attempts in the worker.
2. Graceful shutdown: The shutdown logic waits for the queue to empty, but if a task is requeued after failure, shutdown may hang until backoff expires. Consider a shutdown timeout.
3. Type safety: PriorityQueue requires a total ordering. If priorities are equal, Python falls back to comparing the Task objects themselves. If two tasks have the same priority and different callables, this can raise a TypeError. Add a tie-breaker such as a monotonic sequence number.

两个具体的后续步骤。首先，添加一个文件写入器，让 Agent 将代码输出到磁盘，然后对输出运行 pytest，将任何失败信息反馈给 Agent 进行修复轮次。其次，尝试在规划步骤中换用 qwen-3-32b，或在审查步骤中换用 llama-3.3-70b。因为 Oxlo.ai 使用基于请求的定价，你可以发送很长的 system prompt 和大量上下文，而不必担心输入 token 推高成本。详见 https://oxlo.ai/pricing