多步骤推理与代理式工作流：构建能够规划和执行的 AI

Source: Dev.to

快速参考：你将遇到的术语

技术缩写

• DAG：有向无环图——一种没有循环依赖的工作流结构
• FSM：有限状态机——具有定义好的状态和转移的系统
• CoT：思路链（Chain of Thought）——用于逐步推理的提示技术
• ReAct：推理 + 行动（Reasoning + Acting）——将思考与工具使用相结合的模式
• LLM：大语言模型（Large Language Model）——基于 Transformer 的文本生成系统

统计与数学术语

• State（状态）：工作流中所有变量的当前快照
• Transition（转移）：基于条件从一个状态移动到另一个状态的过程
• Topological Sort（拓扑排序）：对 DAG 节点进行排序，使依赖关系先出现
• Idempotent（幂等）：多次执行仍产生相同结果的操作

引言：从单一提示到编排工作流

想象你在计划一次跨州自驾旅行。你不会只说“开车去加州”然后直接上路。你会：

• 分解：把行程拆成若干段（芝加哥 → 丹佛 → 拉斯维加斯 → 洛杉矶）
• 规划：确定加油站、酒店、景点等
• 执行：逐段驾驶，并根据交通和天气进行调整
• 跟踪状态：知道自己所在位置、剩余油量、已完成的任务

单次 LLM 调用就像在问“我该怎么去加州？”得到的是路线指引，却没有执行过程。Agentic 工作流则真正完成旅程——处理绕行、爆胎、道路封闭等突发情况。

类比 1： 单一提示相当于函数；Agentic 工作流相当于程序。
函数只计算一件事。程序则编排多个函数，管理状态，处理错误，产生复杂结果。

类比 2： 传统 LLM 使用像是计算器；Agentic AI 像是电子表格。
计算器回答单个问题。电子表格维护状态，单元格之间相互依赖，输入变化时自动更新。

任务分解：把问题拆开

from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
import json

class TaskStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    BLOCKED = "blocked"

@dataclass
class Task:
    """Represents a single task in a workflow."""
    id: str
    description: str
    dependencies: List[str] = field(default_factory=list)
    status: TaskStatus = TaskStatus.PENDING
    result: Optional[Any] = None
    error: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)

@dataclass
class TaskPlan:
    """A decomposed plan with multiple tasks."""
    goal: str
    tasks: List[Task]
    created_at: str = ""

    def get_ready_tasks(self) -> List[Task]:
        """Get tasks whose dependencies are all completed."""
        completed_ids = {t.id for t in self.tasks if t.status == TaskStatus.COMPLETED}
        ready = []
        for task in self.tasks:
            if task.status == TaskStatus.PENDING:
                if all(dep in completed_ids for dep in task.dependencies):
                    ready.append(task)
        return ready

    def is_complete(self) -> bool:
        """Check if all tasks are completed."""
        return all(t.status == TaskStatus.COMPLETED for t in self.tasks)

    def has_failed(self) -> bool:
        """Check if any task has failed."""
        return any(t.status == TaskStatus.FAILED for t in self.tasks)

class TaskDecomposer:
    """
    Decompose complex goals into executable task plans.

    Two approaches:
    1. LLM‑based: Let the model break down the task
    2. Template‑based: Use predefined patterns for known task types
    """

    DECOMPOSITION_PROMPT = """Break down this goal into specific, executable tasks.

Goal: {goal}

Context: {context}

Rules:
1. Each task should be atomic (one clear action)
2. Identify dependencies between tasks
3. Tasks should be ordered logically
4. Include validation/verification tasks where appropriate

Output JSON format:
{
    "tasks": [
        {"id": "task_1", "description": "...", "dependencies": []},
        {"id": "task_2", "description": "...", "dependencies": ["task_1"]}
    ]
}

Output only valid JSON."""

    def __init__(self, llm_client):
        self.llm = llm_client

    def decompose(self, goal: str, context: str = "") -> TaskPlan:
        """Decompose a goal into tasks using LLM."""
        prompt = self.DECOMPOSITION_PROMPT.format(goal=goal, context=context)

        response = self.llm.generate(prompt)

        # Parse response
        try:
            # Handle markdown code blocks
            if "```" in response:
                response = response.split("```")[1]
                if response.startswith("json"):
                    response = response[4:]

            data = json.loads(response.strip())

            tasks = [
                Task(
                    id=t["id"],
                    description=t["description"],
                    dependencies=t.get("dependencies", [])
                )
                for t in data["tasks"]
            ]

            return TaskPlan(goal=goal, tasks=tasks)

        except (json.JSONDecodeError, KeyError) as e:
            # Fallback: single task
            return TaskPlan(
                goal=goal,
                tasks=[Task(id="task_1", description=goal)]
            )

    def decompose_with_template(
        self,
        goal: str,
        template: str
    ) -> TaskPlan:
        """Use predefined templates for common task patterns."""

        templates = {
            "research": [
                Task(id="search", description="Search for relevant sources", dependencies=[]),
                Task(id="extract", description="Extract key information", dependencies=["search"]),
                Task(id="synthesize", description="Synthesize findings", dependencies=["extract"]),
                Task(id="validate", description="Validate accuracy", dependencies=["synthesize"])
            ],
            "data_pipeline": [
                Task(id="extract", description="Extract data from source", dependencies=[]),
                Task(id="validate_input", description="Validate input data", dependencies=["extract"]),
                Task(id="transform", description="Transform data", dependencies=["validate_input"]),
                Task(id="validate_output", description="Validate output data", dependencies=["transform"]),
                Task(id="load", description="Load to destination", dependencies=["validate_output"])
            ],
            "analysis": [
                Task(id="gather", description="Gather relevant data", dependencies=[]),
                Task(id="clean", description="Clean and prepare data", dependencies=["gather"]),
                Task(id="analyze", description="Perform analysis", dependencies=["clean"]),
                Task(id="interpret", description="Interpret results", dependencies=["analyze"]),
                Task(id="report", description="Generate report", dependencies=["interpret"])
            ]
        }

        if template not in templates:
            raise ValueError(f"Unknown template: {template}")

        # Customize task descriptions with goal
        tasks = []
        for t in templates[template]:
            tasks.append(Task(
                id=t.id,
                description=f"{t.description} for: {goal}",
                dependencies=t.dependencies.copy()
            ))

        return TaskPlan(goal=goal, tasks=tasks)

# Simple LLM client interface
class LLMClient:
    """Provider‑agnostic LLM client."""

    def __init__(self, provider: str = "openai", model: str = None):
        self.provider = provider
        self.model = model or self._default_model()

    def _default_model(self) -> str:
        return {
            "openai": "gpt-4o-mini",
            "anthropic": "claude-3-haiku-20240307"
        }.get(self.provider, "gpt-4o-mini")

    def generate(self, prompt: str, temperature: float = 0) -> str:
        if self.provider == "openai":
            # Implementation would call OpenAI API
            pass
        elif self.provider == "anthropic":
            # Implementation would call Anthropic API
            pass
        else:
            raise NotImplementedError(f"Provider {self.provider} not supported")