多Agent协作模式：让AI Agents协同工作

用户请求├── Agent A（市场分析）├── Agent B（技术评估）├── Agent C（风险分析）└── Agent D（竞品研究）↓汇总报告

项目总监Agent├── 产品经理Agent│ └── 各种执行Agent├── 技术负责人Agent│ └── 各种执行Agent└── 运营负责人Agent└── 各种执行Agent

from typing import TypedDict, Listfrom langgraph.graph import StateGraph, ENDfrom langchain_openai import ChatOpenAI# 定义状态class ArticleState(TypedDict):    topic: str    research: str    outline: str    draft: str    feedback: str    final: str# 初始化LLMllm = ChatOpenAI(model="gpt-4")# Agent 1: 选题助手def research_agent(state: ArticleState) -> ArticleState:    """研究Agent：收集相关资料"""    topic = state["topic"]    prompt = f"""请对以下主题进行深入研究，收集相关资料：主题：{topic}请提供：1. 主题背景2. 关键概念3. 行业现状4. 发展趋势要求：专业、详实、有数据支撑。"""    response = llm.invoke(prompt)    return {"research": response.content}# Agent 2: 大纲生成def outline_agent(state: ArticleState) -> ArticleState:    """大纲Agent：根据研究结果生成文章大纲"""    prompt = f"""请根据以下研究资料，生成文章大纲：{state['research']}要求：1. 结构清晰，层次分明2. 重点突出，详略得当3. 符合技术文章写作规范"""    response = llm.invoke(prompt)    return {"outline": response.content}# Agent 3: 撰写Agentdef writing_agent(state: ArticleState) -> ArticleState:    """撰写Agent：根据大纲撰写文章"""    prompt = f"""请根据以下大纲撰写完整文章：{state['outline']}要求：1. 语言流畅，专业但易懂2. 内容充实，案例丰富3. 代码示例清晰准确4. 长度：3000-5000字"""    response = llm.invoke(prompt)    return {"draft": response.content}# Agent 4: 审核Agentdef review_agent(state: ArticleState) -> ArticleState:    """审核Agent：审核文章并提供反馈"""    prompt = f"""请审核以下文章，提供修改建议：{state['draft']}请从以下维度进行审核：1. 内容准确性2. 逻辑清晰度3. 语言表达4. 代码质量5. 整体结构请给出具体的修改建议："""    response = llm.invoke(prompt)    return {"feedback": response.content}# Agent 5: 终审Agentdef final_agent(state: ArticleState) -> ArticleState:    """终审Agent：根据反馈修改文章"""    prompt = f"""请根据以下反馈修改文章：原文：{state['draft']}修改建议：{state['feedback']}要求：1. 认真吸收所有合理建议2. 保持原文优点3. 修改后直接输出最终版本"""    response = llm.invoke(prompt)    return {"final": response.content}# 构建工作流workflow = StateGraph(ArticleState)# 添加节点workflow.add_node("research", research_agent)workflow.add_node("outline", outline_agent)workflow.add_node("writing", writing_agent)workflow.add_node("review", review_agent)workflow.add_node("final", final_agent)# 设置流程workflow.set_entry_point("research")workflow.add_edge("research", "outline")workflow.add_edge("outline", "writing")workflow.add_edge("writing", "review")workflow.add_edge("review", "final")workflow.add_edge("final", END)# 编译app = workflow.compile()# 执行result = app.invoke({    "topic": "AI Agent技术的发展与应用",    "research": "",    "outline": "",    "draft": "",    "feedback": "",    "final": ""})print(result["final"])

from typing import TypedDictfrom langgraph.graph import StateGraph, ENDfrom langgraph.constants import Sendfrom langchain_openai import ChatOpenAIimport jsonllm = ChatOpenAI(model="gpt-4")class AnalysisState(TypedDict):    query: str    market_analysis: str    tech_analysis: str    risk_analysis: str    final_report: str# 并行分析函数def market_analysis_agent(state: AnalysisState) -> AnalysisState:    prompt = f"""从市场角度分析以下问题：    {state['query']}    请分析：    1. 市场规模与增长趋势    2. 目标用户群体    3. 竞争格局    4. 商业机会    """    response = llm.invoke(prompt)    return {"market_analysis": response.content}def tech_analysis_agent(state: AnalysisState) -> AnalysisState:    prompt = f"""从技术角度分析以下问题：    {state['query']}    请分析：    1. 技术可行性    2. 技术难点与挑战    3. 技术选型建议    4. 技术风险    """    response = llm.invoke(prompt)    return {"tech_analysis": response.content}def risk_analysis_agent(state: AnalysisState) -> AnalysisState:    prompt = f"""从风险角度分析以下问题：    {state['query']}    请分析：    1. 项目风险    2. 合规风险    3. 运营风险    4. 风险缓解措施    """    response = llm.invoke(prompt)    return {"risk_analysis": response.content}# 分发函数：并行调用三个分析Agentdef parallel_analysis(state: AnalysisState):    return [        Send("market", state),        Send("tech", state),        Send("risk", state)    ]# 汇总Agentdef synthesis_agent(state: AnalysisState) -> AnalysisState:    prompt = f"""请综合以下三个维度的分析，生成最终报告：【市场分析】{state['market_analysis']}【技术分析】{state['tech_analysis']}【风险分析】{state['risk_analysis']}请生成一份综合性的分析报告，包含：1. 执行摘要2. 各维度要点总结3. 综合建议4. 下一步行动计划"""    response = llm.invoke(prompt)    return {"final_report": response.content}# 构建工作流workflow = StateGraph(AnalysisState)workflow.add_node("market", market_analysis_agent)workflow.add_node("tech", tech_analysis_agent)workflow.add_node("risk", risk_analysis_agent)workflow.add_node("synthesis", synthesis_agent)workflow.set_entry_point("parallel_analysis")workflow.add_conditional_edges("parallel_analysis", parallel_analysis)workflow.add_edge("market", "synthesis")workflow.add_edge("tech", "synthesis")workflow.add_edge("risk", "synthesis")workflow.add_edge("synthesis", END)app = workflow.compile()# 执行result = app.invoke({    "query": "是否应该投资开发企业级AI Agent平台？",    "market_analysis": "",    "tech_analysis": "",    "risk_analysis": "",    "final_report": ""})print(result["final_report"])