多Agent协作实战：用qm搭建AI Agent工作流的完整教程

HN热榜上出现了一个有意思的项目——`qm`（Multiplayer agent harness for work），由yc-software开源。它解决了一个越来越紧迫的问题：当你有多个AI Agent需要协调工作时，怎么避免它们互相踩脚、重复劳动、甚至产生冲突输出？

本文带你从安装到搭建第一个多Agent协作流水线，完整实操。

qm的设计哲学

qm把自己定位为”Agent的工头”——它不做推理，不调用LLM，只做一件事：管理Agent之间的任务分配、状态同步和结果汇总。

和LangGraph、CrewAI这类”帮你编排Agent”的框架不同，qm更轻量——它假设你的Agent已经是独立的进程或API，qm只是给它们一个共享的”任务看板”。

Step 1：安装

pip install qm-harness

验证安装：

qm --version
# 输出: qm 0.1.0

Step 2：定义一个简单的工作流

创建 `my_workflow.py`：

from qm import Harness, Task, Agent

# 初始化Harness
harness = Harness(name="code-review-pipeline")

# 定义三个Agent角色
@harness.agent("code-writer")
class CodeWriter(Agent):
    async def execute(self, task: Task):
        spec = task.input.get("spec")
        # 这里调用你的LLM生成代码
        result = await self.call_llm(
            f"Write code for: {spec}"
        )
        return {"code": result}

@harness.agent("code-reviewer")
class CodeReviewer(Agent):
    async def execute(self, task: Task):
        code = task.context.get("code")
        # 调用LLM审查代码
        review = await self.call_llm(
            f"Review this code: {code}"
        )
        return {"review": review, "approved": "BUG" not in review}

@harness.agent("test-runner")
class TestRunner(Agent):
    async def execute(self, task: Task):
        code = task.context.get("code")
        # 执行测试
        return {"passed": True, "output": "All tests green"}

# 定义工作流
harness.chain(
    Task("write", agent="code-writer", input={"spec": "$input"}),
    Task("review", agent="code-reviewer", context={"code": "$write.code"}),
    Task("test", agent="test-runner", context={"code": "$write.code"}),
    condition="review.approved == True"
)

harness.run(input="Create a REST API for user authentication")

Step 3：理解核心概念

Task（任务）： 工作流的原子单元。包含输入参数（`input`）、上下文引用（`context`）、执行条件（`condition`）。

Agent（代理）： 执行任务的实体。可以是LLM调用、API请求、本地脚本——qm不关心Agent内部怎么实现。

Context传递： `$write.code` 语法让后续Task直接引用前面Agent的输出。这避免了Agent之间手动传递数据的麻烦。

条件执行： `condition=”review.approved == True”` 让测试阶段只在代码审查通过后才执行。

Step 4：进阶——并行Agent

qm支持让多个Agent并行处理同一个任务，然后汇总结果：

harness.parallel(
    Task("analyze_security", agent="security-scanner", 
         input={"code": "$write.code"}),
    Task("analyze_performance", agent="perf-analyzer", 
         input={"code": "$write.code"}),
    Task("analyze_style", agent="style-checker", 
         input={"code": "$write.code"}),
)

三个分析Agent同时运行，qm自动等待全部完成后再汇总。

生产环境注意事项

超时处理。 每个Task可以设置 `timeout` 参数。Agent卡住时qm不会无限等待。

重试机制。 用 `retry=3` 让失败的Task自动重试——适合处理LLM API的偶发超时。

状态持久化。 qm默认把工作流状态存在内存中。生产环境建议配置SQLite或Redis后端：

qm run my_workflow.py --backend sqlite:///workflow.db

监控与告警。 qm内置了简单的metrics接口（默认端口9090），可以集成到Prometheus：

harness = Harness(
    name="code-review-pipeline",
    metrics_port=9090,
    alert_on_failure=True
)

与主流框架的对比

qm的定位很清晰——它是Agent团队的”任务看板”，不是”编排引擎”。具体对比：

– LangGraph：强大的图状态管理，但学习曲线陡峭，适合复杂工作流

– CrewAI：开箱即用的多Agent框架，但灵活性有限

– qm：极简主义——只做任务分配和状态传递，适合已有Agent基础架构的团队

如果你已经在用LangChain或CrewAI，qm可以作为”胶水层”嵌入——在现有Agent之上加一层任务编排，而不是替换整个框架。

适用场景与限制

适合： 代码审查流水线、多Agent研究协作、文档生成管道、CI/CD中的AI辅助步骤。特别是那些”Agent角色已经定义好、但需要标准化协作流程”的团队。

不适合： 需要复杂条件分支的工作流（不如用n8n或Temporal）、实时性要求极高的场景（qm有调度延迟）、Agent之间需要大量数据交换的场景（qm的上下文传递机制比较朴素）。

一个真实的场景举例： 假设你要写一篇技术博客——Agent A负责调研和收集素材，Agent B负责撰写初稿，Agent C负责事实核查和技术审核，Agent D负责润色和排版。qm可以让这四个Agent像编辑部的同事一样，按顺序交接工作，自动传递上下文。整个过程你只需要在最后确认一遍即可。

**判断：** qm填补了”让多个Agent像团队一样工作”的工具空白。它的价值不在于功能多强，而在于足够简单——你不需要学一个新框架，只需要告诉qm”谁做什么、按什么顺序做”。如果你的团队已经在用AI Agent做具体任务，qm就是那个让它们学会”团队协作”的最后一块拼图。

— END —
LC 智趣厅 · 科技与生活的交点
ihygg.cn