非科班出身的人和“虚拟员工”之间的距离没有想象中那么远。所谓虚拟员工这里指的不是一个聊天窗口而是一组由大模型驱动的 AI Agent它们接收任务、按照岗位提示词运转、调用必要工具最后把结果写回指定位置。当这样的 Agent 只有三五个时直接写脚本加提示词就能管理当规模靠近 46 个问题的性质就变了员工注册表怎么组织、任务队列怎么分发、失败怎么重试、日志怎么区分。这套方案会从概念讲起用一个最小可运行示例展示如何从零搭出一套能同时调度几十个虚拟员工的轻量系统。这套方案适合三类读者一是非科班出身、想用自动化替代重复工作的人二是刚接触 AI Agent、希望理解它和聊天机器人区别的开发者三是已经用低代码工作流平台想迁移到自建体系的同学。示例基于 Python 3.10 和 OpenAI 兼容的模型服务接口不绑定具体厂商也不要求先读完计算机基础课程。代码会尽量精简但保留工程化所需的配置、日志、重试和并发控制。1. 先理解虚拟员工到底是什么1.1 虚拟员工不是聊天机器人聊天机器人的典型体验是一个对话框用户提问模型回答对话结束。它没有任务边界也没有对结果负责的意识。用户问今天天气它回答天气用户让它整理一份会议纪要它也能整理但整理完就结束了不会自动保存到某个目录不会触发后续流程也不会在失败时重试。虚拟员工则完全不同。它更像一个“岗位”有明确的职责说明有固定的工作流程有输入和输出格式要求。比如一个“内容编辑”员工它的职责是把零散素材整理成结构化文章一个“数据整理助理”的职责是把原始表格清洗成分析摘要。每一个虚拟员工都对应一组系统提示词、一组允许调用的工具以及一个可监控的任务状态。判断一个系统是不是虚拟员工体系可以看三个特征任务是否可以脱离实时对话运行比如通过队列提交。员工是否拥有固定身份比如角色、工具权限、输出格式约束。运行结果是否可追踪比如任务 ID、状态、耗时、错误信息。如果这些都满足就不再是“聊几句”的机器人而是一个可以批量指挥的数字劳动力。1.2 虚拟员工的三项核心能力要让虚拟员工真正干活而不是只生成文本它需要三项能力。第一是规划。员工接到任务后要能够把任务拆成步骤。这个拆解在大模型里通常通过提示词完成告诉它“先分析再写草稿最后整理成 Markdown”。简单的场景靠提示词约束就够了复杂场景需要支持 ReAct 或 Function Calling 这类机制。第二是工具调用。虚拟员工要能读取文件、请求接口、查询数据库、发送消息而不是只输出文字。工具调用是虚拟员工区别于聊天机器人的关键点。模型生成调用请求代码负责执行真实操作再把执行结果回传给模型继续生成。第三是记忆。这个记忆分为两层一层是当前任务内部的上下文比如“当前整理的是哪份材料”另一层是跨任务的历史比如“这个员工最近处理了多少任务上次出错在哪里”。在轻量实现里任务内部上下文可以直接放在 payload 中跨任务历史则通过日志和数据库补全。1.3 为什么非科班也能做过去要实现类似效果需要掌握分布式系统、机器学习、后端开发等多门技术。现在门槛低了很多原因有三个。第一模型服务化。绝大多数情况下不需要自己训练模型只需要调用一个兼容 OpenAI Chat Completions 协议的接口即可。请求体是 JSON响应也是 JSON用 Python 的 requests 或 openai 客户端就能完成。第二配置驱动开发。把每个员工的身份、提示词、温度、输出长度抽成配置而不是写死在代码里。新增一个员工就等于新增一条配置。非科班开发者不需要理解复杂的类继承和设计模式只需要维护好字典、列表和 YAML 文件。第三低代码工作流平台已经验证了流程。很多平台把大模型封装成节点再用连线编排流程。自建虚拟员工体系本质上是把这种可视化编排换成代码编排思路完全一致只是自由度更高、可监控性更强。1.4 一套虚拟员工体系必须有的四个模块任何一个可用的虚拟员工体系至少包含四个模块模块作用缺少时的后果员工注册表集中管理每个人的角色、提示词、参数、工具白名单员工配置散落无法统一维护任务入口接收任务写入队列或数据库无法脱离人工实时触发执行器调用模型、执行工具、处理结果任务跑不起来或无人执行日志与监控记录状态、耗时、错误、结果出错时无法定位是哪个员工、哪一步失败这篇文章后面的代码就是围绕这四个模块展开的最小实现。先跑通一条链路再逐步增加员工数量最后补充监控和生产化配置。2. 从 1 个到 46 个规模增长带来哪些变化2.1 第 1 个虚拟员工单进程脚本第一个虚拟员工往往是一个脚本读取输入拼接提示词调用模型接口打印结果。它适合验证模型效果也适合验证某个岗位的提示词是否准确。这个阶段不需要任务队列不需要并发控制也不需要日志系统。脚本的问题在于每执行一次都要人工介入。如果每天只需要处理几份材料这种模式完全够用。真正出现变化是从“要处理的任务变多”开始的。2.2 第 10 个虚拟员工并发与任务队列当员工数量来到 10 个左右第一件不舒服的事情是执行速度。一个任务调用模型可能需要 20 到 60 秒如果所有任务排队执行效率会很低。此时需要引入并发。并发引入后第二个问题跟着出现任务在哪个员工上执行结果往哪里写失败要不要重试于是需要一个任务队列把“提交任务”和“执行任务”解耦。提交方只需要把任务放进队列执行方从队列里取任务执行完更新状态。这个阶段系统的复杂度开始接近一个小型的任务调度系统。2.3 第 46 个虚拟员工注册表、日志、监控、成本控制当员工数量继续膨胀到几十个新增员工本身变成了一项工程操作。46 个员工意味着 46 套提示词、46 个参数组合、46 个可能的失败场景。如果配置是写死在脚本里的那么每新增一个岗位都要改代码、重新部署。并且由于是并行执行一旦某个模型接口超时或者 API Key 限流多任务会同时失败。这个时候最需要做三件事把员工配置从代码中抽出来变成注册表。给每个任务分配唯一 ID记录完整生命周期。对模型调用加超时、重试和并发上限。这些改造做完以后新增员工不再需要修改核心调度代码只需要往注册表里加一条记录。2.4 46 这个数字背后其实是管理复杂度46 本身没有特殊意义有意义的是一旦员工数量达到几十个系统会从“写脚本”进入“做平台”的阶段。举个例子一个员工每天跑 10 次46 个员工就是每天 460 次模型调用。如果单次调用耗时 30 秒串行处理需要接近 4 小时。并发稍微提上来以后成本又变成主要约束。再叠加提示词版本变化、输出格式不匹配、外部工具超时等情况定位问题就会变得很困难。所以从 1 个到 46 个改变的从来不是数字而是管理方式。本文的第 4 节就是针对这个阶段设计的一套轻量实现。3. 环境准备先把最小技术栈跑起来3.1 环境检查清单本文示例使用 Python 3.10 及以上版本核心依赖很少建议先确认本机环境检查项要求说明Python 版本3.10使用asyncio和类型标注低版本可能报语法错误pip可用安装requirements.txt需要模型服务 API任意兼容 Chat Completions 的服务需要准备 API Base、API Key、模型名称Git可选管理代码和提示词版本时推荐Redis生产环境推荐本文示例不依赖 Redis生产化时用于持久化任务队列3.2 初始化项目目录创建一个目录并初始化虚拟环境mkdir virtual-staff cd virtual-staff python3 -m venv .venv source .venv/bin/activateWindows 环境下激活虚拟环境的命令是.venv\Scripts\activate接着安装依赖pip install pydantic pydantic-settings openai python-dotenv如果后面要加 HTTP 接口再安装 FastAPI 和 uvicornpip install fastapi uvicorn3.3 密钥配置单独放不要写进代码模型服务的 API Key 属于敏感信息推荐通过环境变量或.env文件管理。在项目根目录创建.envLLM_API_BASEhttps://your-model-service.example.com/v1 LLM_API_KEYyour-api-key-here LLM_MODELyour-model-name LLM_TIMEOUT60创建.gitignore把.env和日志目录排除掉.env .venv/ __pycache__/ *.pyc logs/注意不要把真实 API Key 提交到 Git 仓库。一旦泄露不仅会产生费用风险还可能造成数据安全问题。推荐给 API Key 设置最小权限只允许访问当前项目需要的模型服务。3.4 项目目录规划本文示例按下面结构组织代码virtual-staff/ ├── app/ │ ├── __init__.py │ ├── config.py │ ├── models.py │ ├── clients.py │ ├── registry.py │ ├── scheduler.py │ └── demo.py ├── .env ├── .gitignore └── requirements.txtconfig.py负责读取环境变量models.py定义员工和任务的数据结构clients.py封装模型调用registry.py管理员工注册表scheduler.py实现任务调度demo.py是运行入口。4. 实现最小可运行的虚拟员工调度系统4.1 定义员工和任务模型用 Pydantic 或 dataclass 定义数据模型。这里用 dataclass字段清晰适合学习# app/models.py from dataclasses import dataclass, field from typing import Any, Optional dataclass class Employee: name: str role: str system_prompt: str temperature: float 0.2 max_tokens: int 1000 enabled: bool True tools: list[str] field(default_factorylist) dataclass class Task: task_id: str employee_id: str payload: dict[str, Any] status: str pending result: str error: str Employee是虚拟员工的岗位定义Task是每次执行的任务记录。任务的状态管理非常重要后面日志和监控都依赖这个字段。4.2 读取环境变量config.py使用 pydantic-settings 读取环境变量# app/config.py from pydantic_settings import BaseSettings class Settings(BaseSettings): llm_api_base: str https://your-model-service.example.com/v1 llm_api_key: str llm_model: str your-model-name llm_timeout: float 60.0 class Config: env_file .env env_file_encoding utf-8 settings Settings()pydantic-settings 会自动读取.env文件变量名对应 Python 字段名。如果环境变量不存在会使用默认值。4.3 封装模型客户端统一封装一个chat函数所有虚拟员工都通过它调用模型# app/clients.py import json from openai import OpenAI from app.config import settings from app.models import Employee _client None def get_client() - OpenAI: global _client if _client is None: _client OpenAI( base_urlsettings.llm_api_base, api_keysettings.llm_api_key, timeoutsettings.llm_timeout, ) return _client def chat(employee: Employee, task) - str: messages [ {role: system, content: employee.system_prompt}, {role: user, content: json.dumps(task.payload, ensure_asciiFalse)}, ] response get_client().chat.completions.create( modelsettings.llm_model, messagesmessages, temperatureemployee.temperature, max_tokensemployee.max_tokens, ) return response.choices[0].message.content or base_url指向兼容 Chat Completions 协议的服务地址这样不绑定任何指定厂商。temperature、max_tokens这两个参数来自员工配置可以让不同岗位有不同的生成风格和长度限制。4.4 创建员工注册表员工注册表的核心是返回一个字典key 是员工 IDvalue 是员工配置。先定义两个典型岗位# app/registry.py from app.models import Employee def build_registry() - dict[str, Employee]: employees { content_editor: Employee( name内容编辑, rolecontent_editor, system_prompt( 你是一名中文技术内容编辑擅长把零散素材整理成结构清晰的文章。 输出 Markdown 格式先写背景再写要点最后给出建议。 ), temperature0.4, max_tokens2000, ), data_assistant: Employee( name数据整理助理, roledata_assistant, system_prompt( 你是一名数据处理助理。把用户提供的原始数据整理成表格摘要。 输出必须包含 Markdown 表格和一段文字总结。 ), temperature0.0, max_tokens1500, ), } return employees如果要扩展到 46 个员工不需要改调度逻辑只需继续往这个字典里加配置。实际项目中建议把配置放到 YAML 文件或数据库表而不是硬编码在 Python 里。这里保留 Python 字典是为了让示例最小可运行。4.5 用异步队列调度任务调度器是核心模块。它维护一个任务队列、一组执行器并把结果写回任务记录# app/scheduler.py import asyncio import logging import time import uuid from app.clients import chat from app.models import Task from app.registry import build_registry logger logging.getLogger(scheduler) class VirtualStaffScheduler: def __init__(self, worker_count: int 4): self.registry build_registry() self.queue: asyncio.Queue asyncio.Queue() self.worker_count worker_count self.semaphore asyncio.Semaphore(worker_count) self.tasks: dict[str, Task] {} self._workers: list[asyncio.Task] [] async def start(self): logger.info(启动 %s 个虚拟员工执行器, self.worker_count) for i in range(self.worker_count): self._workers.append(asyncio.create_task(self._work(i))) async def submit(self, employee_id: str, payload: dict) - str: task_id uuid.uuid4().hex[:8] task Task(task_idtask_id, employee_idemployee_id, payloadpayload) self.tasks[task_id] task await self.queue.put(task) return task_id async def _work(self, worker_id: int): while True: task await self.queue.get() async with self.semaphore: await self._execute(worker_id, task) self.queue.task_done() async def _execute(self, worker_id: int, task: Task): employee self.registry.get(task.employee_id) if not employee: task.status failed task.error employee not found return task.status running started time.time() try: result await asyncio.to_thread(chat, employee, task) task.result result task.status success except Exception as exc: task.status failed task.error str(exc) finally: logger.info( worker%s task%s employee%s status%s cost%.2fs, worker_id, task.task_id, task.employee_id, task.status, time.time() - started, )关键点有三个。第一asyncio.Queue让任务提交和执行解耦生产环境可以把队列换成 Redis但逻辑一致。第二asyncio.to_thread把同步的模型调用放到线程池执行避免阻塞事件循环。学习阶段这样写足够生产环境建议使用异步客户端或者独立 Worker 进程。第三semaphore限制并发数量防止同一时间发起过多模型请求。4.6 运行与验证创建一个demo.py跑通完整流程# app/demo.py import asyncio import logging from app.scheduler import VirtualStaffScheduler logging.basicConfig(levellogging.INFO, format%(asctime)s | %(levelname)s | %(message)s) async def main(): scheduler VirtualStaffScheduler(worker_count4) await scheduler.start() task_id await scheduler.submit(content_editor, { topic: 什么是 AI 虚拟员工, raw_material: 用大模型驱动、按岗位职责接收任务、自动执行并返回结果的 AI Agent。, }) print(已提交任务:, task_id) await scheduler.queue.join() task scheduler.tasks[task_id] print(任务状态:, task.status) print(执行结果:) print(task.result) if __name__ __main__: asyncio.run(main())运行时先确保.env文件已经配置好模型服务信息然后执行python -m app.demo预期输出大致如下2025-01-01 10:00:01 | INFO | 启动 4 个虚拟员工执行器 已提交任务: 3a8f9b2c 2025-01-01 10:00:05 | INFO | worker0 task3a8f9b2c employeecontent_editor statussuccess cost3.62s 任务状态: success 执行结果: # 什么是 AI 虚拟员工 ...到这里最小闭环已经跑通提交任务、调度执行、调用模型、写回结果、打印日志。5. 关键代码和参数怎么调5.1 提示词模板与岗位绑定虚拟员工的身份来自system_prompt。它不是随便写几句话而是要把岗位的职责、输入格式、输出格式、边界条件都写清楚。实际项目中建议按这个顺序组织 system prompt岗位身份一句话说明你是什么角色。工作职责说明收到输入后要做什么处理。输出约束说明输出格式比如必须 Markdown、必须包含表格、禁止输出无关内容。失败边界说明什么情况要明确告知无法处理不要编造结果。同一套模型接口只要 system prompt 不同员工行为就会明显不同。这也是虚拟员工体系“低代码扩展”的核心增加一个员工主要是增加一条高质量的提示词。5.2 模型参数速查表下面是几个最常用参数的作用以及调整建议参数默认值参考调小的影响调大的影响适用场景temperature0.2输出更稳定、更保守输出更多样、更有创造性数据整理用 0.0 到 0.2创意文案用 0.7 以上max_tokens1000输出更短可能截断输出更长费用更高长文生成需要调大但要注意成本timeout60 秒快速失败但可能误判慢请求更容忍慢响应但任务堆积普通任务 30 到 60 秒复杂任务可以到 120 秒n1一次返回一条结果返回多条候选成本翻倍需要人工选择场景才值得调大temperature是最需要根据岗位调整的参数。数据整理、信息抽取、代码生成建议调低让输出更可控文案创作、标题生成、头脑风暴建议调高让结果更有活力。5.3 工具调用的边界模型本身只能生成文本真正“干活”靠的是工具。在轻量实现里所谓工具就是提前定义好的 Python 函数在调用模型之前或之后执行。举个例子假设某个员工需要抓取网页正文def execute_tools(employee: Employee, payload: dict) - dict: data dict(payload) for tool in employee.tools: if tool url_summary and data.get(url): data[page_text] fetch_text(data[url]) return data然后在clients.py的chat函数中先用工具处理再把处理后的数据传给模型。这样模型看到的不是原始链接而是已经提取好的正文生成质量会好很多。工具列表建议在员工注册表里用白名单管理防止某个岗位调用不相关的能力。比如内容编辑只允许调用markdown、url_summary数据助理只允许调用table、data_clean。5.4 超时、重试和并发控制模型接口不是永远稳定。超时、限流、网络抖动都会出现。学习环境可以简单失败但真实场景必须考虑重试。推荐使用指数退避策略第一次失败后等 1 秒第二次等 2 秒第三次等 4 秒最多重试 3 次。核心思路是给服务恢复留时间。import time def chat_with_retry(employee, task, attempts: int 3): for i in range(attempts): try: return chat(employee, task) except Exception as exc: if i attempts - 1: raise time.sleep(2 ** i)并发控制同样重要。不要以为发请求越快越好。多数模型服务都有速率限制如果并发过高会在网关层直接报错导致大量任务失败。建议从 4 到 8 个并发开始观察失败率和平均耗时再逐步调大。6. 非科班最容易踩的五个坑6.1 API Key 被提交进仓库现象代码推送到远程仓库后发现 API Key 泄露担心被他人盗用。原因.env文件没有加入.gitignore或者把密钥硬编码在配置文件中。解决方式立即在模型服务控制台吊销该 Key并重新生成。然后把.env加入.gitignore。推荐使用环境变量注入密钥本地开发用.env生产环境用密钥管理服务。排查方法git status git log --oneline如果发现泄露历史需要清理 Git 历史并通知相关平台管理员。6.2 员工配置互相污染现象多个员工同时运行后发现某个员工的结果里出现了另一个员工的系统提示词或者参数被其他任务改掉。原因使用了共享的可变对象作为员工配置缓存比如把 Employee 实例放到全局列表后又在运行时修改它的字段。解决方式把员工配置设计成不可变数据运行时只读取不修改。如果需要临时调整参数新建一个副本不要直接改原对象。关键点是调度器只从注册表读取配置不允许任务执行过程中写回配置。6.3 模型返回内容不符合 JSON 格式现象让模型生成结构化 JSON结果偶尔多出解释性文字导致json.loads报错。原因大模型生成时没有严格约束输出格式或者输出被截断。解决方式在 system prompt 中明确“只输出 JSON不要任何解释”。代码侧再做一次兜底提取文本中的 JSON 片段。更稳的方式是使用模型的 Function Calling 能力让输出走结构化参数。推荐格式兜底示例import json import re def parse_json_content(content: str) - dict: match re.search(r\{.*\}, content, re.S) if not match: raise ValueError(未找到 JSON 片段) return json.loads(match.group())6.4 并发任务日志混在一起现象多个任务同时执行日志里无法判断某一行属于哪个员工、哪个任务。原因日志只记录了普通文本没有携带任务 ID、员工 ID。解决方式给每条日志加上task_id、employee_id、status、cost字段。这样可以直接用 grep 按任务 ID 过滤grep task3a8f9b2c logs/app.log强烈建议日志使用结构化格式比如 JSON Lines方便后续接入日志平台。6.5 同一套提示词模板硬套所有岗位现象新增员工时直接复制其他员工的 system prompt只改几个词。结果新员工行为很怪异输出风格不匹配。原因不同的岗位需要不同的约束。把“数据处理”的提示词用在“创意文案”上行为自然不对。解决方式为每个岗位写独立的 system prompt并重点明确输入输出格式。不要追求一套模板通吃。模板可以抽公共开头但职责描述和输出约束必须单独维护。现实项目中提示词本身就是需要版本管理的资产。7. 验证、监控与排查链路7.1 任务状态机每个任务都应该有明确的状态流转pending - running - success - failedpending表示已经进入队列还没有执行器处理。running表示正在调用模型或工具。success表示已经拿到结果并写回。failed表示执行中出现异常。状态机最大的价值是可排查。看到任务停在pending说明队列消费有问题看到长时间处于running说明模型调用超时或卡住看到大量failed说明配置错误或服务不可用。7.2 日志里必须留哪些字段日志不要只记录任务内容还要记录定位问题所需的信息。最小字段集合如下字段含义示例task_id任务唯一 ID3a8f9b2cemployee_id员工 IDcontent_editorstatus任务状态successcost耗时秒3.62error错误信息Connection timeouttimestamp时间2025-01-01 10:00:05日志格式建议写成 JSON方便用 jq 或日志平台处理。7.3 从现象倒推问题以下是常见的排查顺序现象第一步检查常见原因处理方向任务一直 pending查看 Worker 是否启动调度器没有消费队列检查scheduler.start()是否执行任务 failed错误为空查看日志的异常堆栈异常被吞掉捕获异常时记录完整堆栈模型返回空字符串查看 prompt 和返回内容输出被截断调大max_tokens所有任务都失败查看模型服务状态API Key 失效或服务限流检查.env配置和模型服务控制台结果不符合预期对比单个 script 调用提示词设置错误单独调试该员工 prompt7.4 5 分钟实现一个失败率统计在没有专业监控系统之前可以写一个简单的统计函数def stats(tasks: dict[str, Task]) - dict: total len(tasks) success sum(1 for t in tasks.values() if t.status success) failed sum(1 for t in tasks.values() if t.status failed) return { total: total, success: success, failed: failed, failure_rate: round(failed / total, 4) if total else 0.0, }放到定时任务里每小时输出一次。只要失败率突然上升就说明可能是模型服务或提示词出了问题。8. 从 Demo 到生产几十个虚拟员工的工程化清单8.1 配置外置化学习环境把配置写在 Python 字典和.env文件里但生产环境需要进一步拆分员工岗位配置放到数据库或配置中心。模型 API Key 放到密钥管理系统。并发数、超时时间、重试次数放到环境变量。新增员工时只需要在后台录入一条记录