适用场景
适合已经能调用大模型接口、想把“一个模型干一件事”升级成“多个角色分工协作”的开发者与运维。本文用 DeepSeek 提供的 OpenAI 兼容接口和工具调用能力作为推理底座，在其上搭一层轻量编排：规划者拆任务、执行者调工具、审核者收口，全程写入结构化日志，任何一步都能回放和排查。不依赖重型框架，一台普通开发机即可跑通，方便后续替换成自建模型服务。
环境与前置条件
- 操作系统：Linux（Ubuntu / Debian / CentOS 系均可）、macOS；Windows 建议在 WSL 中操作，避免路径与编码差异。
- Python：3.10 及以上（以官方文档当前版本为准），用虚拟环境隔离依赖。
- 内存与磁盘：8 GB 内存起步，项目与日志预留 20 GB 磁盘。若把 base_url 指向本机部署的开源模型，显存需求以模型卡与量化方式为准，通常需要独立 GPU。
- 网络：能访问模型服务的 API 地址；若走内网代理，提前配置好
HTTPS_PROXY。 - 账号：在 DeepSeek 开放平台创建 API Key，并确认账户有可用额度。模型名、接口地址、计费规则均以官方文档当前版本为准。
- 基础技能：会写 Python 函数、看得懂 JSON、能在终端执行命令。
分步骤部署
第 1 步：创建项目骨架与虚拟环境
这一步建立目录结构并隔离依赖，避免污染系统 Python。
```bash
mkdir -p agent-pipeline && cd agent-pipeline
mkdir -p pipeline config workspace logs
python3 -m venv .venv
source .venv/bin/activate # Windows(WSL) 同样适用
python -V # 预期输出 Python 3.10.x 或更高
```
成功标志：命令行提示符前出现 (.venv)，python -V 打印的版本不低于 3.10。
第 2 步：安装依赖并写入配置
安装官方 OpenAI SDK（DeepSeek 接口兼容该协议）、环境变量加载与 YAML 之外的轻量依赖。
```bash
cat > requirements.txt <<'EOF'
openai
python-dotenv
EOF
pip install -r requirements.txt
```
接着写入密钥文件。只写进本机 .env，不要提交到代码仓库。
```bash
cat > .env <<'EOF'
DEEPSEEK_API_KEY=替换成你自己的密钥
DEEPSEEK_BASE_URL=替换成官方文档给出的接口地址
DEEPSEEK_MODEL=deepseek-chat
AGENT_LOG=logs/run.jsonl
EOF
printf '.env\n.venv/\nlogs/\nworkspace/\n__pycache__/\n' > .gitignore
```
成功标志：pip list 中能看到 openai；.env 文件存在且 .gitignore 已包含它。模型名 deepseek-chat 为示例，实际可选项以官方文档当前版本为准。
第 3 步：实现可观测层（日志是一切的地基）
先写观测模块，后面所有调用都经过它。每条事件带同一个 trace_id，串联整条流水线。
```python
pipeline/obs.py
import json, os, uuid, threading
from datetime import datetime, timezone
LOG_PATH = os.environ.get("AGENT_LOG", "logs/run.jsonl")
_lock = threading.Lock()
def new_trace() -> str:
return uuid.uuid4().hex[:12]
def log_event(trace_id: str, step: str, **fields):
os.makedirs(os.path.dirname(LOG_PATH) or ".", exist_ok=True)
rec = {"ts": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
"trace_id": trace_id, "step": step, **fields}
with _lock, open(LOG_PATH, "a", encoding="utf-8") as f:
f.write(json.dumps(rec, ensure_ascii=False) + "\n")
brief = " ".join(f"{k}={v}" for k, v in fields.items()
if k in ("role", "task_id", "status", "duration_ms", "tool"))
print(f"[{trace_id}] {step} {brief}")
```
成功标志：文件能通过 python -c "from pipeline.obs import new_trace; print(new_trace())" 打印出 12 位十六进制字符串（pipeline/__init__.py 需为空文件）。
第 4 步：封装模型调用（带重试与耗时统计）
把“调用模型”收敛成一个函数，统一处理重试、超时和 token 计量。
```python
pipeline/llm.py
import os, time
from openai import OpenAI
from pipeline import obs
client = OpenAI(api_key=os.environ["DEEPSEEK_API_KEY"],
base_url=os.environ.get("DEEPSEEK_BASE_URL") or None)
MODEL = os.environ.get("DEEPSEEK_MODEL", "deepseek-chat")
def chat(messages, *, trace_id, step, tools=None, temperature=0.2, retries=3):
last = None
for i in range(retries):
t0 = time.time()
try:
kwargs = {"model": MODEL, "messages": messages, "temperature": temperature}
if tools:
kwargs["tools"] = tools
resp = client.chat.completions.create(**kwargs)
u = getattr(resp, "usage", None)
obs.log_event(trace_id, step, status="ok",
duration_ms=int((time.time() - t0) * 1000),
prompt_tokens=getattr(u, "prompt_tokens", None),
completion_tokens=getattr(u, "completion_tokens", None))
return resp.choices[0].message
except Exception as e:
last = e
obs.log_event(trace_id, step, status="error",
attempt=i + 1, error=type(e).__name__)
time.sleep(2 ** i)
raise last
```
成功标志：此时还不能直接跑通，但 python -c "import pipeline.llm" 不报错，说明依赖与导入链正常。
第 5 步：定义角色（AI 指挥 AI 的“岗位说明书”）
角色就是三段AI 词典：系统提示词">系统提示词，用 JSON 存放，方便不改代码调整策略。
```bash
cat > config/roles.json <<'EOF'
{
"planner": {
"system": "你是任务规划者。把用户目标拆成 1-5 个可独立执行的子任务。只输出 JSON，格式为 {\"tasks\":[{\"id\":\"t1\",\"role\":\"worker\",\"goal\":\"...\",\"acceptance\":\"...\"}]}。不要输出解释文字，不要使用 Markdown 代码块。"
},
"worker": {
"system": "你是执行者。只完成当前这一条子任务，优先调用工具把结果落盘，不要在回复里贴大段文件内容。任务完成后输出一句不超过 80 字的完成说明。"
},
"reviewer": {
"system": "你是审核者。根据各子任务的完成说明，判断整体是否达成用户目标。输出 JSON：{\"pass\":true/false,\"summary\":\"一句话结论\",\"gaps\":[\"未完成项\"]}。"
}
}
EOF
```
成功标志：python -c "import json;print(list(json.load(open('config/roles.json')).keys()))" 输出 ['planner', 'worker', 'reviewer']。
第 6 步：定义工具，并加路径沙箱
工具让“执行者”真的能干活。这里给三个文件类工具，所有路径都限制在 workspace/ 内，防止读写越界。
```python
pipeline/tools.py
import pathlib
WORKDIR = pathlib.Path("workspace").resolve()
WORKDIR.mkdir(exist_ok=True)
TOOL_SCHEMAS = [
{"type": "function", "function": {
"name": "write_file",
"description": "在 workspace 相对路径写入文本文件，父目录自动创建",
"parameters": {"type": "object",
"properties": {"path": {"type": "string"}, "content": {"type": "string"}},
"required": ["path", "content"]}}},
{"type": "function", "function": {
"name": "read_file",
"description": "读取 workspace 内文本文件内容",
"parameters": {"type": "object", "properties": {"path": {"type": "string"}},
"required": ["path"]}}},
{"type": "function", "function": {
"name": "list_files",
"description": "列出 workspace 内文件，用于确认产出",
"parameters": {"type": "object", "properties": {"path": {"type": "string", "default": "."}}}}},
]
def _safe(p: str) -> pathlib.Path:
f = (WORKDIR / p).resolve()
if f != WORKDIR and WORKDIR not in f.parents:
raise ValueError(f"路径越界: {p}")
return f
def execute(name: str, args: dict) -> str:
if name == "write_file":
f = _safe(args["path"]); f.parent.mkdir(parents=True, exist_ok=True)
f.write_text(args["content"], encoding="utf-8")
return f"已写入 {args['path']}，{len(args['content'])} 字符"
if name == "read_file":
return _safe(args["path"]).read_text(encoding="utf-8")[:4000]
if name == "list_files":
d = _safe(args.get("path", "."))
return "\n".join(str(p.relative_to(WORKDIR)) for p in sorted(d.rglob("*")) if p.is_file()) or "(空)"
raise ValueError(f"未知工具: {name}")
```
成功标志：python -c "from pipeline.tools import execute;print(execute('write_file',{'path':'a.txt','content':'hi'}))" 打印“已写入 a.txt”，且 workspace/a.txt 出现。
第 7 步：编排主循环（分派 → 执行 → 回收）
规划者出任务清单，逐条交给执行者并允许其调工具，最后交审核者收口。
```python
pipeline/orchestrator.py
import json, re
from pipeline import llm, tools, obs
ROLES = json.load(open("config/roles.json", encoding="utf-8"))
MAX_TOOL_STEPS = 6
def _parse_json(text: str) -> dict:
text = re.sub(r"^``(?:json)?|``$", "", text.strip(), flags=re.M).strip()
m = re.search(r"\{.*\}", text, flags=re.S)
if not m:
raise ValueError("模型输出中未找到 JSON")
return json.loads(m.group(0))
def _as_dict(msg):
if hasattr(msg, "model_dump"):
return msg.model_dump(exclude_none=True)
return {k: v for k, v in msg.__dict__.items() if v is not None}
def plan(goal: str, trace: str) -> list:
msg = llm.chat([{"role": "system", "content": ROLES["planner"]["system"]},
{"role": "user", "content": goal}],
trace_id=trace, step="plan")
tasks = _parse_json(msg.content or "").get("tasks", [])
obs.log_event(trace, "plan_parsed", status="ok", task_count=len(tasks))
return tasks
def run_task(task: dict, trace: str) -> str:
messages = [{"role": "system", "content": ROLES["worker"]["system"]},
{"role": "user", "content": f"子任务：{task['goal']}\n验收标准：{task.get('acceptance','无')}"}]
for _ in range(MAX_TOOL_STEPS):
msg = llm.chat(messages, trace_id=trace, step=f"work:{task['id']}",
tools=tools.TOOL_SCHEMAS)
messages.append(_as_dict(msg))
calls = getattr(msg, "tool_calls", None)
if not calls:
return msg.content or ""
for tc in calls:
try:
out = tools.execute(tc.function.name, json.loads(tc.function.arguments or "{}"))
st = "ok"
except Exception as e:
out, st = f"ERROR: {e}", "error"
obs.log_event(trace, "tool_call", task_id=task["id"],
tool=tc.function.name, status=st)
messages.append({"role": "tool", "tool_call_id": tc.id, "content": str(out)[:4000]})
return "达到最大工具步数，子任务未收敛"
def review(goal: str, results: list, trace: str) -> dict:
body = "\n".join(f"- {r['id']}: {r['output']}" for r in results)
msg = llm.chat([{"role": "system", "content": ROLES["reviewer"]["system"]},
{"role": "user", "content": f"用户目标：{goal}\n执行结果：\n{body}"}],
trace_id=trace, step="review")
return _parse_json(msg.content or "")
def run(goal: str, trace: str) -> dict:
tasks = plan(goal, trace)
results = []
for t in tasks:
out = run_task(t, trace)
results.append({"id": t["id"], "output": out})
obs.log_event(trace, "task_done", task_id=t["id"], status="ok")
verdict = review(goal, results, trace)
obs.log_event(trace, "run_finished", status="pass" if verdict.get("pass") else "fail")
return {"trace_id": trace, "verdict": verdict, "results": results}
```
写一个入口脚本：
```python
run.py
import sys, json
from dotenv import load_dotenv
load_dotenv()
from pipeline import orchestrator, obs
goal = sys.argv[1] if len(sys.argv) > 1 else \
"在 workspace 下生成 3 个文件：project/README.md、project/plan.md、project/risks.md，内容分别是项目简介、三步计划、三条风险，并确保三个文件都存在"
trace = obs.new_trace()
print(json.dumps(orchestrator.run(goal, trace), ensure_ascii=False, indent=2))
```
成功标志：命令能执行完毕不抛异常；workspace/ 下出现 project/ 目录与三个文件。
第 8 步：跑起来
```bash
python run.py "在 workspace 下生成 3 个文件：project/README.md、project/plan.md、project/risks.md，内容分别是项目简介、三步计划、三条风险，并确保三个文件都存在"
```
终端会逐行打印形如 [a1b2c3d4e5f6] work:t1 status=ok duration_ms=... 的事件，最后输出包含 verdict.pass 的 JSON。同一次运行的 trace_id 全程一致，这就是“AI 指挥 AI”的流水线轨迹。
验证部署是否成功
按顺序执行三条检查，全部通过即部署完成。
```bash
1) 产出是否真的落盘
ls -R workspace
2) 日志是否完整闭环
python - <<'PY'
import json, collections
rows = [json.loads(l) for l in open("logs/run.jsonl", encoding="utf-8")]
print("事件总数:", len(rows))
print("按步骤统计:", dict(collections.Counter(r["step"].split(":")[0] for r in rows)))
print("错误事件:", sum(1 for r in rows if r.get("status") == "error"))
print("token 合计:", sum(r.get("prompt_tokens") or 0 for r in rows) +
sum(r.get("completion_tokens") or 0 for r in rows))
print("trace 数:", len({r["trace_id"] for r in rows}))
PY
3) 失败可复现性：故意给一个信息不足的任务，看审核者是否判 fail
python run.py "不准调用任何工具，直接完成一份年度总结"
```
预期结果：第 1 条能看到 workspace/project/ 下三个文件；第 2 条中“按步骤统计”出现 plan、plan_parsed、work、tool_call、task_done、review、run_finished，错误事件为 0，token 合计大于 0，trace 数为 2（两次运行）。第 3 条中审核环节应输出 pass: false 并给出 gaps——这说明流水线不是“一律说好”，而是真的做了回收校验。
常见报错与解决
1) openai.AuthenticationError: Error code: 401
原因：.env 未被加载，或密钥值带了引号、空格、换行。
解决：
```bash
grep -c DEEPSEEK_API_KEY .env # 确认存在
python -c "from dotenv import load_dotenv;import os;load_dotenv();print(len(os.environ.get('DEEPSEEK_API_KEY','')))"
```
长度明显异常说明值里混入了空格或引号，重写 .env 后重跑。注意 load_dotenv() 必须写在导入 pipeline.llm 之前。
2) openai.RateLimitError: Error code: 429
原因：单位时间请求过于密集，或账户额度不足。
解决：降低并发（本教程是串行，不会自身打满，但同机多进程会叠加）；代码里已内置 1/2/4 秒指数退避，可把 retries 提到 5；同时确认账户额度与限流策略，以官方文档当前版本为准。
3) json.decoder.JSONDecodeError: Expecting value 或 模型输出中未找到 JSON
原因：规划者返回了带解释文字的段落，或用了未闭合的代码围栏，_parse_json 正则没能剥干净。
解决：已在 _parse_json 中先剥围栏再 re.search(r"\{.*\}")；若仍失败，在 planner 系统提示词末尾追加一句“第一个字符必须是 { ，最后一个字符必须是 }”，并把 temperature 降到 0。切忌盲目加重试次数——提示词没改，重试只会重复失败。
4) ValueError: 路径越界: ../xxx 或 未知工具: xxx
原因：模型编造了工具名，或给出的相对路径跑出了 workspace/。
解决：沙箱拦截本身是正确行为，要让错误回到模型侧自我修正——execute 抛出的异常已被转成 ERROR: ... 回填到 tool 消息里，模型下一轮通常会改正。若频繁发生，在 worker 系统提示词里明确“仅可使用 write_file / read_file / list_files，路径必须相对 workspace”。
5) openai.APITimeoutError 或 httpx.ConnectError
原因：网络抖动、代理未生效，或单次响应过长超过默认超时。
解决：
```bash
echo $HTTPS_PROXY # 需要代理时确认已导出
curl -sS -o /dev/null -w "%{http_code}\n" $DEEPSEEK_BASE_URL
```
在 client.chat.completions.create(...) 中显式传入 timeout=60，并保留指数退避重试。
后续维护
备份：config/roles.json、pipeline/ 下的提示词与工具定义是整条流水线的“资产”，纳入 Git 管理；workspace/ 与 logs/ 属于运行产物，按天归档到对象存储或打包压缩，保留最近 30 天足够定位问题。.env 永不入库，用密钥管理服务或部署平台的环境变量注入。
升级：SDK 与模型名都可能变化。升级前先在独立分支跑通第 8 步的示例任务与第 3 条失败用例，对比日志中的 duration_ms 与 token 合计，确认没有明显劣化再合并。模型名、接口地址以官方文档当前版本为准，不要凭记忆写死。
日志与监控：logs/run.jsonl 建议接入日志轮转，避免长期运行撑满磁盘：
```bash
cat > /etc/logrotate.d/agent-pipeline <<'EOF'
/path/to/agent-pipeline/logs/*.jsonl {
daily
rotate 30
compress
missingok
copytruncate
}
EOF
```
需要盯的四个指标：run_finished 中 status=fail 的比例（目标：稳定下降）、单次运行总 duration_ms、单次运行 token 合计（直接对应成本）、tool_call 中 status=error 的占比（超过一成说明工具描述或沙箱规则需要调整）。把这几项做成每日汇总脚本输出到聊天工具，比事后翻日志有效得多。
安全收口：日志里可能带业务内容，写入前对敏感字段做脱敏；如果把 worker 扩展成能执行 shell，务必放在容器内并限制可执行命令白名单，不要在有生产凭证的机器上直接放开。