从 0 到 1 开发一个 AI Agent 智能体实战项目：工作流自动化引擎（附完整源码与踩坑记录）

本文记录了一个 AI Agent 智能体实战项目的完整开发过程：从架构设计、核心模块实现，到调试过程中踩过的 6 个真实坑位。
项目内置 MockLLM，无需任何 API Key 即可离线跑通；设置环境变量即可一键切换到真实大模型。
文末附完整源码资源包获取方式与后续扩展方向。

---

一、为什么做这个项目

2026 年，Agent 与工作流自动化已经成为 AI 落地的主旋律。但市面上的教程大多停留在"调用一个 API 让 LLM 回复一句话"的层面，缺少一个能真正跑起来、能看懂原理、能动手改造的实战项目。

于是我想写一个"麻雀虽小、五脏俱全"的项目，覆盖 Agent 应用开发的完整技术栈：

• 用 YAML 声明式定义 一条可编排的工作流（任务、条件分支、并行、循环）
• 实现一个 工作流编排引擎，驱动节点逐步执行并跨节点传递上下文
• 构建一个 具备工具调用（Function Calling）能力的 Agent，让它自主规划并调用工具
• 设计 插件式工具注册表，一行代码注册自定义工具
• 接入真实 LLM 的同时内置 MockLLM，保证离线也能完整演示

一句话总结目标：开箱即跑 + 教学完整 + 可二次开发。

---

二、总体架构设计

┌─────────────────────────────────────────────────────┐
│                     上层应用/示例                     │
│   examples/run_workflow.py    examples/build_agent   │
└──────────────────────┬──────────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────────┐
│                    Agent 层（src/agent）             │
│   Agent 核心(ReAct循环)  LLM抽象(Mock/OpenAI)  记忆   │
└──────────────────────┬──────────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────────┐
│                 工作流引擎（src/workflow）            │
│   Engine(调度)  Executor(执行)  Loader(加载)  Nodes   │
└──────────────────────┬──────────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────────┐
│                  工具层（src/tools）                 │
│   Tool基类  ToolRegistry  file_tools/http/text_tools │
└──────────────────────┬──────────────────────────────┘
                       │
        ┌──────────────▼──────────────┐
        │  utils: 模板渲染 + 日志      │
        └─────────────────────────────┘

设计要点：

1. 分层解耦：Agent 层只依赖工具层的注册表接口，工作流引擎可以独立于 Agent 运行
2. 统一结果模型：所有工具/节点都返回 ToolResult（ok / data / error / meta），上层逻辑统一处理
3. 双模式 LLM：BaseLLM 抽象 + 工厂函数 create_llm()，按配置或环境变量自动选择实现
4. 上下文传递：所有节点共享一个变量空间，通过 {{ 变量名 }} 模板引用

---

三、核心模块实现

3.1 工具层：从一行代码开始

工具是 Agent 的"手"。设计上先定义统一的结果模型和工具基类：

# src/tools/base.py
@dataclass
class ToolResult:
    """工具执行结果：所有工具/节点统一返回该结构。"""
    ok: bool
    data: Any = None
    error: str = ""
    meta: Dict[str, Any] = field(default_factory=dict)

    @classmethod
    def success(cls, data: Any = None, **meta):
        return cls(ok=True, data=data, meta=meta)

    @classmethod
    def failure(cls, error: str, **meta):
        return cls(ok=False, error=error, meta=meta)


class Tool:
    """工具基类：子类实现 execute，并声明 name/description/parameters。"""
    name: str = ""
    description: str = ""
    parameters: List[Dict[str, Any]] = []

    def execute(self, **kwargs: Any) -> ToolResult:
        raise NotImplementedError

    def schema(self) -> Dict[str, Any]:
        """生成 OpenAI function calling 格式的 schema。"""
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": {
                    "type": "object",
                    "properties": {p["name"]: p.get("schema", {"type": "string"}) for p in self.parameters},
                    "required": [p["name"] for p in self.parameters if p.get("required")],
                },
            },
        }

配套一个线程安全的注册表，统一管理注册、发现与调用：

# src/tools/registry.py
class ToolRegistry:
    def register(self, tool: Tool) -> None: ...          # 注册
    def unregister(self, name: str) -> None: ...         # 注销
    def call(self, name: str, **kwargs) -> ToolResult:   # 调用（内部捕获异常）
    def schemas(self) -> List[Dict[str, Any]]:           # 供 LLM function calling
    def names(self) -> List[str]: ...

最有价值的设计：call() 内部把工具异常统一包装成 ToolResult.failure，Agent 即使调用工具出错也不会崩溃，而是把错误信息回填给 LLM 继续决策——这是 Agent 健壮性的关键。

内置工具按域划分：file_tools（读/写/列表）、http_tools（GET/POST）、text_tools（split/join/统计）。每个模块提供 build_xxx_tools() 工厂函数。

3.2 工作流引擎：四种节点 + 调度器

工作流是 DAG 的声明式表达。节点模型支持四种类型：

类型	作用	关键字段
task	调用工具或让 LLM 生成	tool / prompt + llm: true
condition	条件分支	condition、on_true、on_false
parallel	并行执行子节点	branches
loop	循环处理列表	over、item_var、body

以一条"每日工作报告"工作流为例（workflows/daily_report.yaml）：

name: daily_report
version: 1.0.0
description: "根据每日任务清单，自动生成一份结构化的工作报告"

inputs:
  - name: date
    type: string
    default: "2026-08-14"
  - name: tasks_file
    type: string
    default: "./data/tasks.txt"

nodes:
  # 节点 1：读取任务清单（工具节点）
  - id: load_tasks
    type: task
    tool: file.read
    params:
      path: "{{ tasks_file }}"
    on_success:
      store_to: raw_tasks

  # 节点 2：LLM 生成日报（LLM 节点）
  - id: generate_report
    type: task
    prompt: |
      你是一名高效的行政助理。请根据下面提供的每日任务原始数据，生成一份专业的中文日报。
      要求：包含标题、日期、完成情况统计、明日计划三个部分，使用 Markdown 列表。
      原始数据：{{ raw_tasks }}
    llm: true
    on_success:
      store_to: report_text

  # 节点 3：质量检查（条件分支）
  - id: quality_check
    type: condition
    condition: "len({{ report_text }}) > 30"
    on_true: [save_report]
    on_false: [regenerate_report]

  # 节点 4：兜底重新生成
  - id: regenerate_report
    type: task
    prompt: "请为以下数据补充生成一份简短日报（至少 50 字）：{{ raw_tasks }}"
    llm: true
    on_success:
      store_to: report_text
      then: [save_report]

  # 节点 5：保存到文件
  - id: save_report
    type: task
    tool: file.write
    params:
      path: "{{ output_dir }}/daily_report_{{ date }}.md"
      content: "{{ report_text }}"
    on_success:
      store_to: saved_path

引擎的调度核心是一个 双队列策略：

# src/workflow/engine.py —— 主循环（节选）
sequence = deque(n.id for n in workflow.nodes)  # 顺序队列：按 YAML 声明顺序
self._pending = deque()                          # 优先队列：then/条件跳转的目标

while sequence or self._pending:
    # 优先执行跳转目标，否则取顺序队列下一个
    node_id = self._pending.popleft() if self._pending else sequence.popleft()
    if node_id in self._done:
        continue

    result = executor.execute(node, ctx)
    self._done.add(node_id)

    # 条件分支：未选中的分支目标节点标记为跳过
    if node.type == "condition" and result.ok:
        branch = (result.data or {}).get("branch")
        skipped = node.on_false if branch == "true" else node.on_true
        for sid in skipped:
            if sid not in self._done:
                self._done.add(sid)  # 防止被顺序队列再次执行

节点执行器负责具体的执行逻辑，包括模板渲染、条件求值、错误重试：

# src/workflow/executor.py —— 节点分发（节选）
def execute(self, node: Node, ctx: Dict[str, Any]) -> ToolResult:
    retries = node.retries or int(self.settings.get("max_retries", 0))
    for attempt in range(retries + 1):
        if node.type == "task":
            last = self._run_task(node, ctx)
        elif node.type == "condition":
            last = self._run_condition(node, ctx)
        elif node.type == "parallel":
            last = self._run_parallel(node, ctx)   # ThreadPoolExecutor 并发
        elif node.type == "loop":
            last = self._run_loop(node, ctx)
        ...
        if last.ok:
            break
    if last.ok:
        self._handle_success(node, ctx, last)      # store_to / append_to / then
    return last

_handle_success 支持三种动作：

• store_to：结果存入上下文变量
• append_to：追加到列表变量
• then：执行完后跳转到指定节点（通过回调把目标节点插入优先队列）

3.3 Agent 层：ReAct 循环 + 工具调用

Agent 采用经典的 ReAct 风格循环：

用户任务 → LLM 决策(是否调用工具) → 执行工具 → 结果回填 → 再次决策
          ↑_____________________________________________________|
                    直到 LLM 给出最终答复或达到最大迭代次数

核心实现：

# src/agent/core.py —— Agent 主循环（节选）
def run(self, task: str) -> AgentResult:
    self.memory.clear()
    current_task = task

    for iteration in range(1, self.config.max_iterations + 1):
        decision = self._decide(current_task)
        calls = decision.get("tool_calls") or []

        if not calls:
            # 无工具调用 → 视为最终答复
            answer = decision.get("content") or self._fallback_answer(current_task)
            return AgentResult(success=True, answer=answer, iterations=iteration, ...)

        # 依次执行工具，结果写入记忆
        for call in calls:
            name, args = call["name"], call.get("arguments", {})
            result = self.registry.call(name, **args)
            self.memory.add_tool(name, args, json.dumps(result.to_dict(), ensure_ascii=False, default=str))

        # 把工具结果拼进下一轮决策上下文
        current_task = self._build_followup(current_task, tool_calls)

_build_followup 会把工具执行结果原样回填给下一轮决策：

def _build_followup(self, original, tool_calls):
    parts = [f"原始任务：{original}", "\n以下是工具执行结果："]
    for tc in tool_calls:
        parts.append(f"- 工具 {tc['tool']}：成功={tc['ok']}，结果={json.dumps(tc.get('data'), ensure_ascii=False, default=str)[:200]}")
    parts.append("\n请基于上述结果给出最终答案（不要再次调用工具）。")
    return "\n".join(parts)

LLM 抽象层设计了两套实现：

# src/agent/llm.py（节选）
class BaseLLM:
    def chat(self, prompt, system="") -> str: ...                  # 单轮文本
    def chat_with_tools(self, prompt, tool_schemas, system=""): ... # 带工具声明

class MockLLM(BaseLLM):
    """内置模拟 LLM：无需 API Key，离线演示与单元测试用。"""

class OpenAILLM(BaseLLM):
    """真实 LLM：兼容 OpenAI Chat Completions（含各类兼容网关）。"""

def create_llm(cfg):
    mode = os.environ.get("AGENT_LLM_MODE") or cfg.get("mode", "mock")
    if mode == "openai":
        return OpenAILLM(...)
    return MockLLM()

关键设计：MockLLM 不是简单返回固定字符串，而是实现了一个启发式决策器——根据任务文本中的关键词（"统计/多少个"→ 调 file.list，"天气"→ 调 weather.query，等等）模拟 LLM 的工具调用决策。这让离线演示也具备完整的"规划 → 调用 → 回填 → 答复"链路，教学价值极高。

3.4 模板渲染：让数据在节点间流动

所有节点的参数都支持 {{ 变量 }} 模板引用。渲染器支持嵌套对象深度渲染：

# src/utils/template.py
def render(text: str, context: Dict[str, Any]) -> Any:
    """渲染单条文本模板，{{ key }} 引用上下文变量。"""

def render_deep(obj: Any, context: Dict[str, Any]) -> Any:
    """递归渲染 dict / list / 字符串，用于节点参数。"""

def render_expr(expr: str, context: Dict[str, Any]) -> str:
    """表达式场景渲染：字符串值会被 repr 加引号，便于 eval 安全求值。"""

---

四、实战演示

4.1 离线跑通工作流自动化

python examples/run_workflow.py --demo

运行效果（节选）：

开始执行工作流: daily_report v1.0.0
>> 节点 [load_tasks] (task) 读取任务清单
>> 节点 [generate_report] (task) 生成工作报告
>> 节点 [quality_check] (condition) 报告质量检查
  条件 len('...日报内容...') > 30 => True
>> 节点 [save_report] (task) 保存日报文件
[OK] 工作流 daily_report 执行成功，耗时 0.52s

执行后自动生成 output/daily_report_2026-08-14.md，完整链路：读取数据 → LLM 生成 → 质量检查 → 保存报告。

4.2 Agent 自主调用工具

python examples/build_custom_agent.py

>> Agent 接收任务: 统计当前项目 src 目录下 Python 文件的数量
─ 决策轮 1/10
  [tool] 调用工具 file.list {"path": "src"}
─ 决策轮 2/10
[OK] Agent 完成（第 2 轮）

========== Agent 执行结果 ==========
成功: True  迭代: 2  工具调用: 1
  - file.list -> ok=True, data=[...]
最终答复:
src 目录下共有 19 个 .py 文件。

Agent 自主完成了：理解任务 → 决定调用 file.list → 基于工具结果给出最终答复，全程无需人工干预。

4.3 一行代码注册自定义工具

from src.tools.base import Tool, ToolResult

class WeatherTool(Tool):
    name = "weather.query"
    description = "查询指定城市的天气情况"
    parameters = [{"name": "city", "type": "string", "required": True, "description": "城市名"}]

    def execute(self, city: str = "北京") -> ToolResult:
        # 此处可替换为真实天气 API
        return ToolResult.success({"city": city, "weather": "晴", "temperature": 26})

registry.register(WeatherTool())

注册后 Agent 即可在任务中自动发现并调用它。

---

五、踩坑记录（真实 Debug 经历）

开发过程中踩了不少坑，每一个都很有代表性，分享出来帮大家少走弯路。

坑 1：Windows 控制台 GBK 编码崩溃

现象：日志里的特殊符号 ▶、✔、✘ 在 Windows 控制台直接抛 UnicodeEncodeError。

排查：Windows 默认 GBK 编码，\u25b6 等符号不在 GBK 字符集内；同时 sys.stdout 写入失败导致整个流程中断。

修复：全局日志改用 ASCII 安全字符（>>、[OK]、[FAIL]、[tool]），并支持设置 PYTHONIOENCODING=utf-8 运行。

# 修复前
log.info("▶ 节点 [%s] (%s)", node.id, node.type)
log.info("✔ 工作流执行成功")
# 修复后
log.info(">> 节点 [%s] (%s)", node.id, node.type)
log.info("[OK] 工作流执行成功")

教训：跨平台工具类项目，日志输出要避免使用平台无关的特殊符号。

坑 2：条件表达式字符串变量缺少引号导致 eval 失败

现象：condition: "len({{ report_text }}) > 30" 渲染后变成 len(2026-08-14 完成 3 项任务...) > 30，eval 直接 SyntaxError——字符串变量没有被引号包裹。

排查：render() 对字符串值做的是直接替换，没有加引号，导致拼接后的表达式语法错误。

修复：新增 render_expr()，对字符串值用 repr() 加引号：

def render_expr(expr, context):
    def _repl(match):
        value = _resolve(match.group(1), context)
        if isinstance(value, str):
            return repr(value)      # 关键：加引号
        ...
    return _PATTERN.sub(_repl, expr)
# "len({{ report_text }}) > 30" → "len('日报内容...') > 30" ✅

教训：模板引擎做"表达式渲染"和"文本渲染"是两种语义，必须分开处理。

坑 3：条件分支未选中节点被顺序队列重复执行

现象：条件为真时，兜底节点 regenerate_report 仍被执行了——条件分支只"跳过了前面的顺序"，没有阻止后续顺序队列。

排查：引擎的双队列机制里，condition 节点只负责把 on_true 目标插入优先队列，但 on_false 的目标节点还在顺序队列里等着被 pop。

修复：条件节点执行后，把未选中分支的目标节点直接标记为 _done：

if node.type == "condition" and result.ok:
    branch = (result.data or {}).get("branch")
    skipped = node.on_false if branch == "true" else node.on_true
    for sid in skipped:
        if sid not in self._done:
            self._done.add(sid)   # 跳过未选分支，防止顺序队列再次执行

教训：调度语义要区分"顺序执行"和"跳转执行"两套队列，条件分支必须显式"杀死"未选中的路径。

坑 4：MockLLM 决策器死循环

现象：Agent 对"统计 src 目录下 Python 文件数量"任务，在同一工具上反复调用，陷入死循环。

排查：两个原因叠加：

1. 正则 [\w./\-]+ 中 \w 在 Python 里默认匹配 Unicode 字符，src 被错误匹配到中文路径；
2. 工具结果回填后，决策器没有"结果阶段"的识别规则，继续命中统计规则重复调用同一工具。

修复：两处一起改：

• 路径提取正则只匹配 ASCII：[A-Za-z0-9_./\\-]+，并增加优先级（带扩展名的文件 → 带斜杠的路径 → 常见目录名）
• 决策器增加"回填阶段"规则：任务文本含"工具执行结果"时直接给最终答复，不再调用工具

# 规则 0：回填阶段 → 直接给出最终答复
if "工具执行结果" in task:
    return {"role": "assistant", "content": self.llm.chat(task, SYSTEM_PROMPT), "tool_calls": []}

教训：启发式规则引擎必须考虑"状态机"（初始阶段 vs 回填阶段），否则同一规则会无限触发。

坑 5：相对路径依赖 CWD，一换运行方式就找不到文件

现象：示例脚本直接运行时正常，但从项目外目录运行、或 pytest 执行时，file.read 报找不到 src/、data/。

排查：所有相对路径（./data/tasks.txt、src/）都依赖进程的工作目录（CWD），运行方式一变就失效。

修复：在示例脚本和 tests/conftest.py 中统一把 CWD 切换到项目根：

PROJECT_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
os.chdir(PROJECT_ROOT)   # 保证相对路径基于项目根解析

教训：示例代码要"自适应 CWD"，用 __file__ 定位项目根，而不是假设运行目录。

坑 6：引擎只返回声明 outputs，测试断言拿不到上下文

现象：测试里构造的工作流没有声明 outputs，引擎返回的 context 是空的，断言全挂。

排查：引擎只汇总 workflow.outputs 中声明的变量，未声明时返回空字典。

修复：未声明 outputs 时返回完整上下文（过滤内部 _ 前缀变量），保持"最小意外"原则：

if workflow.outputs:
    output_ctx = {out["name"]: ctx[out["name"]] for out in workflow.outputs if out["name"] in ctx}
else:
    output_ctx = {k: v for k, v in ctx.items() if not k.startswith("_")}

教训：框架的默认行为要"宽容"，显式声明用于收窄，不声明就返回全量。

---

六、测试与质量保障

项目内置 9 个单元测试，覆盖引擎调度与 Agent 决策两个核心：

python -m pytest tests/ -v

tests/test_engine.py .........      [ 引擎：顺序执行/条件分支/并行/循环/上下文传递 ]
tests/test_agent.py  ......         [ Agent：工具调用/决策/结果回填 ]
9 passed in 0.35s

关键测试点：

• 条件分支正确跳过未选中分支
• loop 节点正确处理列表迭代与上下文回写
• parallel 节点并发执行且结果合并
• Agent 在多轮工具调用后给出最终答复

---

七、项目交付：资源包

通过 scripts/build_resource_pack.py 一键打包交付，产出 zip 资源包（约 40 个文件）：

ai-agent-workflow/
├── src/         核心源码（agent / workflow / tools / utils）
├── workflows/   3 条示例工作流（YAML 声明式）
├── examples/    CLI 运行器 + Agent/工具示例
├── docs/        实战教程 + 架构设计 + CSDN 发布说明
├── tests/       9 个单元测试
├── scripts/     打包脚本 + Windows 环境准备脚本
├── config/      全局配置（LLM 模式等）
└── data/        示例数据

快速上手：

pip install -r requirements.txt
python examples/run_workflow.py --demo            # 离线演示（无需 API Key）
python examples/build_custom_agent.py             # Agent 工具调用演示
python -m pytest tests/ -v                        # 运行测试

接入真实大模型（可选）：

$env:OPENAI_API_KEY = "sk-xxxx"
$env:OPENAI_BASE_URL = "https://api.openai.com/v1"   # 也可换成兼容网关
python examples/run_workflow.py --workflow workflows/daily_report.yaml

---

八、后续扩展方向

这个项目是刻意做"小而全"的脚手架，以下方向都可以继续生长：

1. 多 Agent 协作：引入主控 Agent 与专家 Agent，通过消息队列协作
2. 人工审批节点：工作流中增加 human 节点，关键步骤等待人工确认
3. 持久化记忆：Memory 落地到向量数据库，支持长期记忆与检索
4. 可观测性：节点级 tracing + 执行 DAG 可视化
5. 分布式执行：parallel 节点替换为 Celery / Ray 集群任务
6. 更多内置工具：数据库、爬虫、定时任务、消息推送等

---

九、结语

这个项目的价值不在于"功能多炫"，而在于让 Agent 应用开发的全链路可见、可改、可跑：

• 理解工作流 = 节点的声明式编排
• 理解 Agent = LLM + 工具 + 循环决策
• 理解工程化 = 分层、抽象、测试、打包

希望这篇开发记录对你有所帮助。如果你在部署或二次开发中遇到问题，欢迎留言交流。

源码资源包：https://download.csdn.net/download/2501_93047244/93274210
相关文章：《AI Agent 智能体实战：工作流自动化原理与架构解析》

---

如果你觉得这篇内容有帮助，欢迎点赞、收藏、关注，后续会继续输出 Agent 工程化实战系列。