AI Agent 工作流自动化教程｜附源码

在过去一年里，“AI Agent（智能体）”从一个技术概念迅速变成了很多团队正在落地的生产力工具。相比普通聊天机器人，AI Agent 最大的价值不只是“回答问题”，而是能够理解目标、拆解任务、调用工具、执行流程、反馈结果，从而完成一系列自动化工作。

例如：

• 自动读取用户需求，生成待办任务；
• 自动检索资料，整理成报告；
• 自动调用接口，完成数据查询；
• 自动发送邮件、生成文档、同步到数据库；
• 自动根据执行结果继续下一步操作。

本文将用一个完整示例，带你从零搭建一个简单但实用的 AI Agent 工作流自动化系统。文章包含架构说明、核心思路、代码实现和扩展方向，适合想要学习 AI Agent 落地实践的开发者、产品经理和自动化爱好者。

一、什么是 AI Agent 工作流自动化？

简单来说，AI Agent 工作流自动化就是让 AI 不仅能“说”，还能“做”。

传统的大模型应用一般是这样的：

用户输入问题 -> 大模型生成回答 -> 用户阅读结果

而 AI Agent 工作流则更像这样：

用户提出目标
    ↓
Agent 理解任务
    ↓
拆解执行步骤
    ↓
调用不同工具
    ↓
获得执行结果
    ↓
判断是否继续
    ↓
输出最终结果

举个例子，用户输入：

帮我整理一份关于“AI Agent 在企业办公中的应用”的文章大纲，并保存到本地文件。

普通 AI 可能只会返回一个大纲文本。

而 Agent 可以进一步完成：

1. 理解用户要写文章大纲；
2. 生成结构化大纲；
3. 调用文件写入工具；
4. 将结果保存为 Markdown 文件；
5. 返回文件保存路径。

这就是 Agent 与普通问答机器人最大的区别：它具备行动能力。

二、本文要实现什么？

本文将实现一个轻量级 AI Agent 工作流自动化系统，支持以下能力：

1. 接收用户输入的任务；
2. 使用大模型分析用户意图；
3. 自动选择合适的工具；
4. 执行工具函数；
5. 将工具执行结果返回给大模型；
6. 生成最终回答。

为了方便理解，我们将实现三个工具：

最终我们可以让 Agent 完成类似任务：

请帮我生成一份周报模板，并保存到 weekly_report.md

Agent 会自动判断需要先生成内容，再调用写文件工具，把内容保存下来。

三、技术栈说明

本文示例使用 Python 编写，整体不依赖复杂框架，适合初学者理解 Agent 的底层逻辑。

你需要准备：

• Python 3.9+
• 一个大模型 API Key
• openai Python SDK

安装依赖：

pip install openai python-dotenv

如果你使用的是其他兼容 OpenAI API 格式的大模型服务，也可以复用本文代码，只需要修改 base_url 和模型名称即可。

四、项目目录结构

建议创建如下项目结构：

ai-agent-workflow/
├── .env
├── main.py
├── tools.py
├── agent.py
└── README.md

各文件作用如下：

五、配置环境变量

在项目根目录新建 .env 文件：

OPENAI_API_KEY=你的_API_KEY
OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_MODEL=gpt-4o-mini

如果你使用其他模型服务，可以这样写：

OPENAI_API_KEY=你的_API_KEY
OPENAI_BASE_URL=https://你的服务商地址/v1
OPENAI_MODEL=你的模型名称

注意：不要把 .env 文件上传到公开仓库，否则可能导致 API Key 泄露。

六、编写工具函数：tools.py

工具是 Agent 能够“行动”的关键。
没有工具的 Agent，本质上只是一个聊天机器人；有了工具之后，它才能读取文件、写入文件、查询数据库、调用接口、发送消息。

下面我们先实现三个简单工具。

# tools.py

import os
from datetime import datetime


def get_current_time() -> str:
    """
    获取当前时间。
    """
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")


def write_file(file_path: str, content: str) -> str:
    """
    将内容写入指定文件。
    """
    try:
        # 防止目录不存在
        directory = os.path.dirname(file_path)
        if directory and not os.path.exists(directory):
            os.makedirs(directory)

        with open(file_path, "w", encoding="utf-8") as f:
            f.write(content)

        return f"文件已成功写入：{file_path}"
    except Exception as e:
        return f"写入文件失败：{str(e)}"


def read_file(file_path: str) -> str:
    """
    读取指定文件内容。
    """
    try:
        if not os.path.exists(file_path):
            return f"文件不存在：{file_path}"

        with open(file_path, "r", encoding="utf-8") as f:
            content = f.read()

        return content
    except Exception as e:
        return f"读取文件失败：{str(e)}"

这三个函数本身都非常简单，但它们代表了 Agent 工具调用的基本模式：

工具名称 + 参数 + 执行结果

大模型负责决定什么时候调用工具，而 Python 程序负责真正执行工具。

七、定义工具描述

大模型并不知道你的 Python 函数是什么，因此我们需要用 JSON Schema 的方式告诉模型：

1. 有哪些工具可以使用；
2. 每个工具的功能是什么；
3. 每个工具需要哪些参数；
4. 参数类型是什么。

在 agent.py 中先定义工具描述。

# agent.py

import json
import os
from dotenv import load_dotenv
from openai import OpenAI

from tools import get_current_time, write_file, read_file


load_dotenv()


client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL")
)

MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini")


tools_schema = [
    {
        "type": "function",
        "function": {
            "name": "get_current_time",
            "description": "获取当前系统时间。",
            "parameters": {
                "type": "object",
                "properties": {},
                "required": []
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "write_file",
            "description": "将指定内容写入本地文件。",
            "parameters": {
                "type": "object",
                "properties": {
                    "file_path": {
                        "type": "string",
                        "description": "要写入的文件路径，例如 output.md"
                    },
                    "content": {
                        "type": "string",
                        "description": "要写入文件的文本内容"
                    }
                },
                "required": ["file_path", "content"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "读取本地文件内容。",
            "parameters": {
                "type": "object",
                "properties": {
                    "file_path": {
                        "type": "string",
                        "description": "要读取的文件路径"
                    }
                },
                "required": ["file_path"]
            }
        }
    }
]

这里的 tools_schema 就是模型能看懂的工具说明书。

例如 write_file 的描述告诉模型：

• 这个工具可以写文件；
• 它需要 file_path；
• 它需要 content；
• 两个参数都是字符串。

当用户说“帮我保存到 weekly_report.md”时，模型就有机会判断需要调用 write_file 工具。

八、建立工具映射表

模型只会返回“我想调用某个工具”，但不会真正执行 Python 函数。
所以我们还需要建立工具名称到函数对象的映射关系。

继续在 agent.py 中添加：

tool_map = {
    "get_current_time": get_current_time,
    "write_file": write_file,
    "read_file": read_file,
}

后续当模型返回：

{
  "name": "write_file",
  "arguments": {
    "file_path": "weekly_report.md",
    "content": "这里是周报内容"
  }
}

程序就可以根据 tool_map["write_file"] 找到真正的 Python 函数，并执行：

write_file(file_path="weekly_report.md", content="这里是周报内容")

九、实现 Agent 核心逻辑

下面是整个 Agent 的核心代码。

# agent.py

def run_agent(user_input: str) -> str:
    """
    运行一个简单的 AI Agent。
    """

    messages = [
        {
            "role": "system",
            "content": (
                "你是一个可以调用工具完成任务的 AI Agent。"
                "当用户需要获取时间、读写文件时，你应该优先调用工具。"
                "如果工具执行完成，请根据工具结果给用户一个清晰的最终答复。"
            )
        },
        {
            "role": "user",
            "content": user_input
        }
    ]

    # 第一次请求模型：让模型判断是否需要调用工具
    response = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        tools=tools_schema,
        tool_choice="auto"
    )

    assistant_message = response.choices[0].message
    messages.append(assistant_message)

    # 如果模型决定调用工具
    if assistant_message.tool_calls:
        for tool_call in assistant_message.tool_calls:
            tool_name = tool_call.function.name
            tool_args = json.loads(tool_call.function.arguments or "{}")

            if tool_name not in tool_map:
                tool_result = f"未知工具：{tool_name}"
            else:
                tool_func = tool_map[tool_name]
                tool_result = tool_func(**tool_args)

            # 将工具执行结果加入上下文
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": str(tool_result)
            })

        # 第二次请求模型：让模型基于工具结果生成最终回复
        final_response = client.chat.completions.create(
            model=MODEL,
            messages=messages
        )

        return final_response.choices[0].message.content

    # 如果不需要调用工具，直接返回模型回答
    return assistant_message.content

这段逻辑非常关键，可以分成六步理解：

1. 构造 messages；
2. 把用户任务和工具列表发给模型；
3. 模型判断是否需要调用工具；
4. 如果需要工具调用，程序执行对应函数；
5. 将工具结果再发给模型；
6. 模型生成最终自然语言回复。

这就是一个最小可用的 Agent 工作流。

十、编写程序入口：main.py

接下来创建 main.py：

# main.py

from agent import run_agent


def main():
    print("AI Agent 工作流自动化系统已启动")
    print("输入 exit 或 quit 可退出")
    print("-" * 40)

    while True:
        user_input = input("\n请输入任务：")

        if user_input.lower() in ["exit", "quit"]:
            print("程序已退出")
            break

        if not user_input.strip():
            continue

        result = run_agent(user_input)
        print("\nAgent 回复：")
        print(result)


if __name__ == "__main__":
    main()

运行项目：

python main.py

你可以输入：

现在几点了？

Agent 会调用 get_current_time 工具，并返回类似结果：

当前时间是 2026-06-02 10:30:15。

再输入：

请帮我生成一份周报模板，并保存到 weekly_report.md

Agent 会调用 write_file 工具，将内容写入本地文件。

十一、完整源码汇总

为了方便复制，下面给出完整代码。

1. tools.py

import os
from datetime import datetime


def get_current_time() -> str:
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")


def write_file(file_path: str, content: str) -> str:
    try:
        directory = os.path.dirname(file_path)
        if directory and not os.path.exists(directory):
            os.makedirs(directory)

        with open(file_path, "w", encoding="utf-8") as f:
            f.write(content)

        return f"文件已成功写入：{file_path}"
    except Exception as e:
        return f"写入文件失败：{str(e)}"


def read_file(file_path: str) -> str:
    try:
        if not os.path.exists(file_path):
            return f"文件不存在：{file_path}"

        with open(file_path, "r", encoding="utf-8") as f:
            return f.read()
    except Exception as e:
        return f"读取文件失败：{str(e)}"

2. agent.py

import json
import os
from dotenv import load_dotenv
from openai import OpenAI

from tools import get_current_time, write_file, read_file


load_dotenv()


client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL")
)

MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini")


tools_schema = [
    {
        "type": "function",
        "function": {
            "name": "get_current_time",
            "description": "获取当前系统时间。",
            "parameters": {
                "type": "object",
                "properties": {},
                "required": []
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "write_file",
            "description": "将指定内容写入本地文件。",
            "parameters": {
                "type": "object",
                "properties": {
                    "file_path": {
                        "type": "string",
                        "description": "要写入的文件路径"
                    },
                    "content": {
                        "type": "string",
                        "description": "要写入文件的文本内容"
                    }
                },
                "required": ["file_path", "content"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "读取本地文件内容。",
            "parameters": {
                "type": "object",
                "properties": {
                    "file_path": {
                        "type": "string",
                        "description": "要读取的文件路径"
                    }
                },
                "required": ["file_path"]
            }
        }
    }
]


tool_map = {
    "get_current_time": get_current_time,
    "write_file": write_file,
    "read_file": read_file,
}


def run_agent(user_input: str) -> str:
    messages = [
        {
            "role": "system",
            "content": (
                "你是一个可以调用工具完成任务的 AI Agent。"
                "当用户需要获取时间、读写文件时，你应该优先调用工具。"
                "如果工具执行完成，请根据工具结果给用户一个清晰的最终答复。"
            )
        },
        {
            "role": "user",
            "content": user_input
        }
    ]

    response = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        tools=tools_schema,
        tool_choice="auto"
    )

    assistant_message = response.choices[0].message
    messages.append(assistant_message)

    if assistant_message.tool_calls:
        for tool_call in assistant_message.tool_calls:
            tool_name = tool_call.function.name
            tool_args = json.loads(tool_call.function.arguments or "{}")

            if tool_name not in tool_map:
                tool_result = f"未知工具：{tool_name}"
            else:
                tool_func = tool_map[tool_name]
                tool_result = tool_func(**tool_args)

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": str(tool_result)
            })

        final_response = client.chat.completions.create(
            model=MODEL,
            messages=messages
        )

        return final_response.choices[0].message.content

    return assistant_message.content

3. main.py

from agent import run_agent


def main():
    print("AI Agent 工作流自动化系统已启动")
    print("输入 exit 或 quit 可退出")
    print("-" * 40)

    while True:
        user_input = input("\n请输入任务：")

        if user_input.lower() in ["exit", "quit"]:
            print("程序已退出")
            break

        if not user_input.strip():
            continue

        result = run_agent(user_input)
        print("\nAgent 回复：")
        print(result)


if __name__ == "__main__":
    main()

十二、测试示例

示例一：获取当前时间

输入：

请告诉我当前时间

可能输出：

当前时间是 2026-06-02 14:22:09。

示例二：写入文件

输入：

请生成一份简洁的项目日报模板，并保存到 report.md

执行后，本地会生成 report.md 文件，内容可能如下：

# 项目日报

## 今日完成
- 

## 遇到的问题
- 

## 明日计划
- 

## 备注
- 

Agent 回复：

已为你生成项目日报模板，并成功保存到 report.md。

示例三：读取文件

输入：

请读取 report.md 的内容，并帮我总结一下

Agent 会调用 read_file 工具读取文件内容，然后再生成总结。

十三、如何扩展更多工具？

上面的 Agent 只是一个最小版本，但它的扩展方式非常简单。
如果你想增加新能力，只需要三步：

1. 编写 Python 工具函数；
2. 在 tools_schema 中添加工具描述；
3. 在 tool_map 中注册工具函数。

例如，我们可以增加一个“计算器工具”。

1. 添加工具函数

def calculator(expression: str) -> str:
    """
    计算数学表达式。
    注意：生产环境不要直接使用 eval。
    """
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"计算失败：{str(e)}"

2. 添加工具 Schema

{
    "type": "function",
    "function": {
        "name": "calculator",
        "description": "计算数学表达式。",
        "parameters": {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "数学表达式，例如 3 * (5 + 2)"
                }
            },
            "required": ["expression"]
        }
    }
}

3. 注册工具

tool_map = {
    "get_current_time": get_current_time,
    "write_file": write_file,
    "read_file": read_file,
    "calculator": calculator,
}

之后用户输入：

帮我计算 23 * 19 + 88

Agent 就可以自动调用计算器工具。

不过需要注意，eval 在真实业务中存在安全风险。生产环境建议使用安全的表达式解析库，比如 numexpr、asteval，或者自己实现白名单计算逻辑。

十四、让 Agent 支持多轮工具调用

当前代码只支持一轮工具调用。
也就是说，模型最多只能：

思考 -> 调用工具 -> 根据结果回答

但真实的 Agent 工作流经常需要多轮调用，例如：

读取文件 -> 分析内容 -> 写入新文件 -> 返回结果

这时可以使用循环，让模型在每轮执行后继续判断是否需要新的工具调用。

下面是一个增强版本：

def run_agent_loop(user_input: str, max_steps: int = 5) -> str:
    messages = [
        {
            "role": "system",
            "content": (
                "你是一个可以调用工具完成复杂任务的 AI Agent。"
                "你可以多轮调用工具，但必须在任务完成后给出最终答复。"
            )
        },
        {
            "role": "user",
            "content": user_input
        }
    ]

    for step in range(max_steps):
        response = client.chat.completions.create(
            model=MODEL,
            messages=messages,
            tools=tools_schema,
            tool_choice="auto"
        )

        assistant_message = response.choices[0].message
        messages.append(assistant_message)

        if not assistant_message.tool_calls:
            return assistant_message.content

        for tool_call in assistant_message.tool_calls:
            tool_name = tool_call.function.name
            tool_args = json.loads(tool_call.function.arguments or "{}")

            if tool_name not in tool_map:
                tool_result = f"未知工具：{tool_name}"
            else:
                try:
                    tool_func = tool_map[tool_name]
                    tool_result = tool_func(**tool_args)
                except Exception as e:
                    tool_result = f"工具执行异常：{str(e)}"

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": str(tool_result)
            })

    return "任务执行达到最大步骤限制，请检查任务是否过于复杂。"

这个版本引入了 max_steps，可以避免 Agent 无限循环调用工具。
实际生产环境中，多轮调用是非常重要的，因为复杂任务通常不可能一步完成。

十五、工作流自动化的实际应用场景

AI Agent 工作流自动化可以应用在很多领域。

1. 内容生产自动化

例如：

• 自动生成文章大纲；
• 自动收集参考资料；
• 自动生成初稿；
• 自动保存为 Markdown；
• 自动推送到 CMS 系统。

工作流示例：

输入主题 -> 生成大纲 -> 扩写正文 -> SEO 优化 -> 保存文件 -> 发布草稿

2. 数据分析自动化

例如：

• 读取 Excel 或 CSV；
• 分析销售数据；
• 生成图表；
• 输出分析报告；
• 发送给相关人员。

工作流示例：

读取数据 -> 清洗数据 -> 统计指标 -> 生成图表 -> 输出报告

3. 客服与运营自动化

例如：

• 自动识别用户问题；
• 查询订单状态；
• 调用退款接口；
• 生成回复话术；
• 更新工单状态。

工作流示例：

用户咨询 -> 意图识别 -> 查询订单 -> 判断处理方式 -> 回复用户

4. 企业办公自动化

例如：

• 自动生成会议纪要；
• 自动提取待办事项；
• 自动同步到飞书、钉钉或 Notion；
• 自动提醒负责人。

工作流示例：

会议录音转写 -> 提炼纪要 -> 提取任务 -> 分配负责人 -> 写入协作平台

5. 研发提效自动化

例如：

• 根据需求生成接口文档；
• 根据报错日志定位问题；
• 自动生成测试用例；
• 自动创建 GitHub Issue；
• 自动总结 Pull Request。

工作流示例：

输入需求 -> 生成代码草稿 -> 生成测试 -> 运行检查 -> 输出修改建议

十六、生产环境需要注意什么？

虽然本文示例代码很简单，但如果要真正用于生产环境，需要关注以下问题。

1. 工具权限控制

不要让 Agent 随意访问所有系统能力。
例如文件读写工具应限制目录范围，避免读取敏感文件。

可以增加路径校验：

SAFE_DIR = "workspace"

def safe_path(file_path: str) -> str:
    abs_base = os.path.abspath(SAFE_DIR)
    abs_path = os.path.abspath(os.path.join(SAFE_DIR, file_path))

    if not abs_path.startswith(abs_base):
        raise ValueError("非法文件路径")

    return abs_path

2. 参数校验

模型生成的工具参数不一定总是正确。
因此在执行工具前，应对参数进行校验。

例如：

if not isinstance(file_path, str) or not file_path.strip():
    return "file_path 参数不能为空"

3. 日志记录

Agent 的每一步操作都应该记录日志，包括：

• 用户输入；
• 模型返回；
• 工具名称；
• 工具参数；
• 工具执行结果；
• 最终输出。

这样在出错时方便排查问题。

4. 最大执行步数

多轮 Agent 必须设置最大执行步数。
否则在某些情况下，模型可能反复调用工具，造成死循环和额外成本。

5. 人工确认机制

对于高风险操作，例如：

• 删除文件；
• 发送邮件；
• 扣款退款；
• 修改数据库；
• 发布内容；

最好加入人工确认机制。

流程可以设计为：

Agent 生成操作计划 -> 用户确认 -> 执行工具 -> 返回结果

6. 成本控制

Agent 往往比普通问答消耗更多 Token，因为它需要多轮交互和工具调用。
建议：

• 控制上下文长度；
• 对历史消息做摘要；
• 限制最大调用次数；
• 对工具结果进行截断；
• 对高频任务做缓存。

十七、一个更完整的自动化任务示例

假设我们希望 Agent 完成：

请生成一份 AI Agent 学习计划，包含 7 天安排，并保存为 ai_agent_plan.md

理想执行过程如下：

用户输入任务
    ↓
模型理解：需要生成学习计划并保存文件
    ↓
模型调用 write_file 工具
    ↓
工具写入 ai_agent_plan.md
    ↓
模型返回保存成功

生成的文件可能是：

# AI Agent 7 天学习计划

## 第 1 天：理解基础概念
- 学习大模型、Prompt、Agent、Tool Calling 的区别
- 理解 Agent 的核心组成：模型、工具、记忆、规划

## 第 2 天：学习 Python API 调用
- 熟悉 OpenAI SDK
- 学习 messages 结构
- 掌握 system、user、assistant、tool 角色

## 第 3 天：实现工具调用
- 编写简单工具函数
- 定义 JSON Schema
- 实现工具映射和执行逻辑

## 第 4 天：实现多轮 Agent
- 学习循环调用机制
- 设置最大执行步数
- 处理工具异常

## 第 5 天：增加文件和数据处理能力
- 支持 Markdown 写入
- 支持 CSV 读取
- 支持文本总结

## 第 6 天：接入外部平台
- 接入飞书、企业微信或 Notion
- 实现消息推送
- 实现任务同步

## 第 7 天：完成一个实际项目
- 选择一个真实办公流程
- 设计工具集合
- 实现端到端自动化

这类任务非常适合 Agent 处理，因为它既需要生成内容，又需要执行保存动作。

十八、总结

本文从零实现了一个简单的 AI Agent 工作流自动化系统。核心思想可以概括为：

大模型负责理解和决策，工具负责执行具体动作。

一个可用的 Agent 通常包括以下部分：

1. 任务输入：用户提出目标；
2. 系统提示词：约束 Agent 的角色和行为；
3. 工具描述：告诉模型有哪些工具可用；
4. 工具函数：真正执行动作的代码；
5. 工具映射：把模型返回的工具名称映射到函数；
6. 执行循环：支持工具调用和多轮任务处理；
7. 最终回复：把执行结果整理给用户。

本文示例虽然轻量，但已经覆盖了 AI Agent 的核心机制。你可以在此基础上继续扩展：

• 接入搜索引擎；
• 接入数据库；
• 接入企业微信、飞书、钉钉；
• 接入邮件系统；
• 接入知识库；
• 增加长期记忆；
• 增加任务规划模块；
• 增加人工审批流程。

真正有价值的 AI Agent，并不是单纯追求复杂，而是要能稳定解决具体问题。
从一个简单的文件读写 Agent 开始，逐步扩展到你的业务流程中，才是最有效的落地方式。