1. 项目概述:基于通义千问与LangGraph的多Agent协作PRD生成系统
这个项目实现了一个由多个AI Agent协作完成产品需求文档(PRD)生成的自动化流程。通过模拟真实产品开发团队中的角色分工(产品经理、技术负责人、设计师、测试工程师),每个Agent专注于自己的专业领域,最终输出经过多方评审的完整PRD文档。
核心价值在于:
- 解决了传统PRD撰写中信息不完整、视角单一的问题
- 通过多角色协作自动完成需求分析、技术评估、交互设计和测试方案
- 基于状态机模型实现流程控制,模拟真实产品开发流程
技术栈组合非常巧妙:
- 通义千问(Qwen)作为底层大语言模型提供专业领域知识
- LangGraph构建多Agent协作的工作流引擎
- 状态机模型确保评审流程有序推进
我在实际测试中发现,相比单一大模型直接生成PRD,这种多Agent协作方式产出的文档:
- 技术可行性评估更准确
- 交互设计考虑更周全
- 测试用例覆盖更全面
- 最终文档结构更规范
2. 核心架构设计解析
2.1 角色分工与协作流程
系统设计了5个核心Agent,模拟真实产品团队:
-
PM(产品经理):
- 职责:根据初始需求撰写PRD初稿
- 输出:prd_draft
- 工作内容:需求背景、用户故事、功能列表、优先级
-
Tech Lead(技术负责人):
- 职责:评估技术可行性
- 输出:tech_review
- 工作内容:技术方案选型、实现难度评估、风险点
-
Designer(设计师):
- 职责:提供交互设计方案
- 输出:design_review
- 工作内容:界面流程、交互细节、用户体验优化
-
QA(测试工程师):
- 职责:制定测试方案
- 输出:qa_review
- 工作内容:测试场景、用例设计、验收标准
-
Moderator(协调者):
- 职责:控制流程流转
- 输出:next_agent
- 工作内容:判断当前阶段是否完成,决定下一步执行哪个Agent
实际测试中发现,Tech Lead和Designer的评审意见经常会出现冲突(如技术实现限制与理想交互设计的矛盾),这时就需要Moderator介入协调,让PM重新调整需求。
2.2 状态机模型设计
系统使用TypedDict定义了PRDState状态对象:
python复制class PRDState(TypedDict):
demand: str # 原始需求
messages: Annotated[List[BaseMessage], operator.add] # 对话历史
prd_draft: str # PM初稿
tech_review: str # 技术评审
design_review: str # 设计评审
qa_review: str # 测试评审
final_prd: str # 终版PRD
next_agent: str # 流程控制
状态流转逻辑:
- 初始化:输入原始需求(demand)
- PM Agent生成初稿(prd_draft)
- Tech/Design/QA依次评审
- Moderator判断是否需迭代
- 输出final_prd
2.3 技术选型考量
2.3.1 大模型选择
提供两种Qwen接入方式:
python复制# 方式1:阿里云DashScope(生产推荐)
llm = ChatOpenAI(
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
api_key="sk-你的APIKey",
model="qwen-max", # 也可用qwen-plus/qwen-turbo
temperature=0.3,
max_tokens=4000
)
# 方式2:本地Ollama(开发调试)
# llm = ChatOllama(model="qwen2:7b-instruct-q4_0", temperature=0.3)
选型建议:
- 生产环境:优先使用DashScope的qwen-max,响应快且效果稳定
- 本地开发:可用Ollama+Qwen2 7B模型,但需注意:
- 需先执行
ollama pull qwen2:7b-instruct-q4_0 - 生成速度较慢(约15-20秒/步骤)
- 长文本处理能力较弱
- 需先执行
2.3.2 LangGraph工作流引擎
使用StateGraph构建流程:
python复制workflow = StateGraph(PRDState)
# 添加节点(每个Agent一个节点)
workflow.add_node("pm", pm_agent)
workflow.add_node("tech", tech_agent)
...
# 设置边(流程流转规则)
workflow.add_edge("pm", "tech")
...
# 设置入口
workflow.set_entry_point("pm")
# 编译执行
app = workflow.compile()
关键设计点:
- 使用MemorySaver实现检查点机制,支持流程中断恢复
- 通过next_agent字段实现动态路由
- 每个Agent的output会更新到全局state
3. 核心实现细节
3.1 Agent实现模板
以PM Agent为例的完整实现:
python复制def pm_agent(state: PRDState):
# 构造对话历史
prompt = ChatPromptTemplate.from_messages([
("system", "你是一名资深产品经理,需要根据需求撰写PRD"),
("human", "需求背景:{demand}"),
("human", "请输出完整PRD,包含:\n"
"1. 项目背景\n2. 用户画像\n"
"3. 功能列表\n4. 优先级\n"
"5. 成功指标")
])
# 调用LLM
chain = prompt | llm
response = chain.invoke({"demand": state["demand"]})
# 更新状态
return {
"prd_draft": response.content,
"next_agent": "tech" # 默认流转给Tech
}
3.2 评审Agent的差异化设计
不同评审Agent需要定制化提示词:
Tech Lead提示词重点:
python复制("system", "你作为技术负责人,需要评估:\n"
"1. 技术可行性(1-5分)\n"
"2. 实现复杂度(高/中/低)\n"
"3. 潜在技术风险\n"
"4. 建议技术方案")
Designer提示词特点:
python复制("system", "你作为UX设计师,需要提供:\n"
"1. 核心用户旅程图\n"
"2. 关键界面线框图描述\n"
"3. 交互细节说明\n"
"4. 与现有设计系统的兼容性")
QA提示词关键点:
python复制("system", "你作为测试工程师,需要制定:\n"
"1. 测试场景分类\n"
"2. 关键测试用例(含预期结果)\n"
"3. 自动化测试建议\n"
"4. 性能测试方案")
3.3 Moderator的协调逻辑
Moderator的核心决策逻辑:
python复制def moderator(state: PRDState):
# 收集各阶段输出
reviews = {
"tech": state["tech_review"],
"design": state["design_review"],
"qa": state["qa_review"]
}
# 判断是否需要迭代
conflict = False
if "不可行" in state["tech_review"]:
conflict = True
elif "重大调整" in state["design_review"]:
conflict = True
if conflict:
return {"next_agent": "pm"} # 需PM重新调整
else:
return {"next_agent": "finalize"} # 进入终版生成
实际使用中发现,当技术评审出现"不可行"或设计评审出现"重大调整"时,直接让PM重新修改往往效果不好。更好的做法是让Moderator组织多方讨论,可以添加一个discussion_agent专门处理冲突场景。
4. 完整工作流示例
4.1 初始化与执行
python复制# 初始化状态
initial_state = {
"demand": "开发一个智能待办事项App,支持语音输入、智能提醒",
"messages": [HumanMessage(content="开始生成PRD")],
"prd_draft": "",
# ...其他字段初始为空
}
# 执行工作流
for step in app.stream(initial_state):
print(f"当前阶段: {step['next_agent']}")
if 'final_prd' in step:
print(step['final_prd']) # 最终输出
4.2 典型输出结构
最终生成的PRD包含:
markdown复制# [产品名称] PRD
## 1. 项目背景
- 用户痛点:...
- 市场机会:...
## 2. 功能列表
1. 核心功能:
- 语音输入待办事项
- 智能提醒(基于时间/地点)
2. 扩展功能:
- 多设备同步
- 共享待办事项
## 3. 技术方案
- 语音识别:采用...
- 提醒引擎:基于...
## 4. 交互设计
- 核心流程:...
- 关键界面:...
## 5. 测试方案
- 核心测试场景:...
- 性能测试指标:...
5. 实战经验与优化建议
5.1 效果提升技巧
-
角色人设强化:
- 为每个Agent添加详细的角色背景:
python复制("system", "你是拥有8年电商经验的资深产品经理,曾主导过3个千万级用户产品...") -
评审标准量化:
- 要求Tech Lead给出具体评分:
python复制("human", "请从1-5分评估可行性,并说明扣分点") -
迭代控制:
- 限制最大迭代次数(避免无限循环):
python复制if state.get("iteration", 0) > 3: return {"next_agent": "finalize"}
5.2 常见问题排查
-
流程卡住不动:
- 检查next_agent赋值是否正确
- 确认edges是否完整添加
-
输出质量不稳定:
- 调整temperature(建议0.3-0.5)
- 添加输出格式约束:
python复制("human", "请严格按以下格式输出:\n## 技术评估\n- 可行性:...") -
长文本截断问题:
- 对于qwen-turbo等模型,控制max_tokens
- 复杂需求可拆分为多个子需求分别处理
5.3 扩展应用场景
-
会议纪要生成:
- 添加Chairman、Recorder等角色
- 输入原始讨论记录,输出结构化纪要
-
技术方案设计:
- 角色调整为Architect、DevOps、DBA等
- 输出技术架构图+实施方案
-
商业计划书撰写:
- 设置Market Analyst、Financial Expert等角色
- 生成完整的BP文档
这个框架最让我惊喜的是,通过合理设计Agent角色和流程,可以复用到各种需要多视角协作的内容生成场景。最近我在一个内部项目中,用类似的架构实现了自动生成技术评审报告的功能,相比人工撰写效率提升了70%。