从 0 到 1 开发一个 AI Agent 智能体实战项目：工作流自动化引擎

从 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: 模板渲染 + 日志 │
└─────────────────────────────┘

设计要点：

三、核心模块实现

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 的声明式表达。节点模型支持四种类型：

以一条"每日工作报告"工作流为例（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 文件数量"任务，在同一工具上反复调用，陷入死循环。

排查：两个原因叠加：

修复：两处一起改：

• 路径提取正则只匹配 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

八、后续扩展方向

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

九、结语

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

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

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

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

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