✅使用LangGraph实现一个Agentic Workflow

我们使用LangGraph实现一个多角色协作的“研究助手”系统，当用户问一个问题（如“AI 对医疗行业有哪些影响？”），系统自动：

Planner Agent：制定研究计划（分几步查什么）
Researcher Agent：按计划搜索信息（模拟调用搜索引擎）
Writer Agent：整合信息写报告
Reviewer Agent：检查报告质量，若不满意则要求重写（形成反思循环）

使用langgraph来实现，初始化环境：

uv init agentic_workflow_test

cd agentic_workflow_test

uv venv

source .venv/bin/activate

编写main.py中的代码：

import osfrom typing import Annotated, Literal, TypedDict, Listfrom langgraph.graph import StateGraph, START, ENDfrom langchain_openai import ChatOpenAI
# === 配置 ===OPENAI_API_KEY = "你的 api key"OPENAI_API_BASE = "https://dashscope.aliyuncs.com/compatible-mode/v1"
llm = ChatOpenAI(    model="deepseek-v3",    api_key=OPENAI_API_KEY,    base_url=OPENAI_API_BASE,    temperature=0.1,)
# ================================# 1. 定义全局状态# ================================class ResearchState(TypedDict):    question: str    plan: List[str]    research_notes: str    draft: str    feedback: str    approved: bool    revision_count: int
# ================================# 2. 增强版 Agent（带详细日志）# ================================
def planner_agent(state: ResearchState):    pr
int("\n" + "="*50)    pr
int("🧠 [Planner] 接收到任务")    pr
int(f"  用户问题: {state['question']}")        prompt = f"""    用户问题：{state['question']}    请制定一个清晰的研究计划，分3-5个具体步骤，每步说明要查什么。    只输出步骤列表，不要解释。    """    response = llm.invoke(prompt)    plan_steps = [step.strip() for step in response.content.split("\n") if step.strip()]        pr
int("✅ [Planner] 输出研究计划:")    for i, step in enumerate(plan_steps, 1):        pr
int(f"   {i}. {step}")    return {
"plan": plan_steps
}

def researcher_agent(state: ResearchState):    pr
int("\n" + "="*50)    pr
int("🔍 [Researcher] 开始执行研究")    pr
int("  研究计划:")    for i, step in enumerate(state["plan"], 1):        pr
int(f"   {i}. {step}")        plan_text = "\n".join(f"{i+1}. {step}" for i, step in enumerate(state["plan"]))    prompt = f"""    根据以下研究计划，模拟搜索并总结关键信息：    {plan_text}    请输出简洁的研究笔记（300字以内）。    """    response = llm.invoke(prompt)        pr
int("✅ [Researcher] 输出研究笔记:")    pr
int(f"   {response.content.strip()}")    return {
"research_notes": response.content
}
def writer_agent(state: ResearchState):    pr
int("\n" + "="*50)    pr
int("✍️ [Writer] 开始撰写报告")    pr
int(f"  基于问题: {state['question']}")    pr
int(f"  研究笔记预览: {state['research_notes'][:100]}{'...' if len(state['research_notes']) > 100 else ''}")        # 构造基础 prompt    prompt = f"""    基于以下研究笔记，为问题“{state['question']}”写一份专业、简洁的回答（200字左右）：    {state['research_notes']}    """        # 如果存在反馈，则加入修改指令    if state.get("feedback"):        pr
int(f"  📝 评审反馈: {state['feedback']}")        prompt += f"\n\n请根据以下评审意见进行修改：\n{state['feedback']}\n确保回答准确、简洁、有洞察力。"
    response = llm.invoke(prompt)        pr
int("✅ [Writer] 输出草稿:")    pr
int(f"   {response.content.strip()}")    return {
"draft": response.content
}
def reviewer_agent(state: ResearchState):    pr
int("\n" + "="*50)    pr
int("👀 [Reviewer] 评审当前草稿")    pr
int(f"  草稿内容: {state['draft']}")        prompt = f"""    评审以下回答是否满足：准确、简洁、有洞察力。    如果满意，只回复“APPROVED”。    如果不满意，请给出具体修改建议（50字以内）：        回答：{state['draft']}    """    response = llm.invoke(prompt)    review_text = response.content.strip()        pr
int("✅ [Reviewer] 评审结果:")    if "APPROVED" in review_text.upper():        pr
int("   👍 批准通过！")        return {
"approved": True, "feedback": ""
}
    else:        pr
int(f"   ❌ 需要修改: {review_text}")        return {
            "approved": False,            "feedback": review_text,            "revision_count": state["revision_count"] + 1        }

# ================================# 3. 增强版路由逻辑# ================================def route_after_review(state: ResearchState) -> Literal["writer", "end"]:    pr
int("\n" + "-"*40)    pr
int("🚦 [Router] 决策中...")    pr
int(f"  当前修订次数: {state['revision_count']}")    pr
int(f"  是否批准: {state['approved']}")        if state["approved"]:        pr
int("  → 决定: 流程结束（已批准）")        return "end"    el
if state["revision_count"] >= 2:        pr
int("  → 决定: 流程结束（已达最大修订次数）")        return "end"    else:        pr
int(f"  → 决定: 返回 Writer 进行第 {state['revision_count'] + 1} 次修改")        return "writer"

# ================================# 4. 构建工作流（保持不变）# ================================workflow = StateGraph(ResearchState)workflow.add_node("planner", planner_agent)workflow.add_node("researcher", researcher_agent)workflow.add_node("writer", writer_agent)workflow.add_node("reviewer", reviewer_agent)
workflow.add_edge(START, "planner")workflow.add_edge("planner", "researcher")workflow.add_edge("researcher", "writer")workflow.add_edge("writer", "reviewer")
workflow.add_conditional_edges(    "reviewer",    route_after_review,    {
"writer": "writer", "end": END
})
app = workflow.compile()
# ================================# 5. 运行示例# ================================if __name__ == "__main__":    user_question = "AI 对医疗行业有哪些影响？"        pr
int("🚀 启动 Agentic Research Workflow")    pr
int(f"📌 用户问题: {user_question}\n")
    result = app.invoke({
        "question": user_question,        "plan": [],        "research_notes": "",        "draft": "",        "feedback": "",        "approved": False,        "revision_count": 0    })        pr
int("\n" + "="*60)    pr
int("✅ 最终报告：")    pr
int(result["draft"])        pr
int(f"\n📊 执行统计:")    pr
int(f"  - 总修订次数: {result['revision_count']}")    pr
int(f"  - 是否最终批准: {'是' if result['approved'] else '否（达到最大修订次数）'}")

添加依赖：uv add langgraph langchain-openai

运行代码：uv run main.py

最终输出结果如下：

🚀 启动 Agentic Research Workflow📌 用户问题: AI 对医疗行业有哪些影响？

==================================================🧠 [Planner] 接收到任务  用户问题: AI 对医疗行业有哪些影响？✅ [Planner] 输出研究计划:   1. 1. 调查AI在医疗诊断中的应用，包括影像识别、病理分析和早期疾病预测。   2. 2. 研究AI在医疗管理中的作用，如电子病历处理、医院资源优化和患者流程自动化。   3. 3. 分析AI在药物研发中的贡献，包括靶点发现、临床试验设计和个性化药物方案。   4. 4. 评估AI对医患互动的影响，如虚拟助手、远程医疗和健康监测设备。   5. 5. 探讨AI在医疗行业中的伦理与法律问题，包括数据隐私、算法偏见和监管框架。
==================================================🔍 [Researcher] 开始执行研究  研究计划:   1. 1. 调查AI在医疗诊断中的应用，包括影像识别、病理分析和早期疾病预测。   2. 2. 研究AI在医疗管理中的作用，如电子病历处理、医院资源优化和患者流程自动化。   3. 3. 分析AI在药物研发中的贡献，包括靶点发现、临床试验设计和个性化药物方案。   4. 4. 评估AI对医患互动的影响，如虚拟助手、远程医疗和健康监测设备。   5. 5. 探讨AI在医疗行业中的伦理与法律问题，包括数据隐私、算法偏见和监管框架。✅ [Researcher] 输出研究笔记:   **AI在医疗领域的研究笔记**
1. **医疗诊断**：AI在影像识别（如X光、MRI）中准确率接近专家水平，病理分析可辅助癌症检测，早期预测模型（如糖尿病、阿尔茨海默病）提升干预效率。
2. **医疗管理**：AI优化电子病历结构化与检索，通过预测患者流量改善资源分配，自动化流程（如分诊）缩短等待时间。
3. **药物研发**：加速靶点筛选（如AlphaFold预测蛋白质结构），优化临床试验设计（患者匹配），推动个性化药物（基于基因组数据）。
4. **医患互动**：虚拟助手（如ChatGPT）解答基础咨询，远程医疗平台扩大覆盖，可穿戴设备实时监测慢性病。
5. **伦理与法律**：数据隐私（匿名化处理）、算法偏见（数据多样性不足）、监管滞后（需跨学科协作制定标准）是主要挑战。
**总结**：AI显著提升医疗效率与精准度，但需平衡创新与伦理风险，完善法规以实现可持续发展。 （298字）
==================================================✍️ [Writer] 开始撰写报告  基于问题: AI 对医疗行业有哪些影响？  研究笔记预览: **AI在医疗领域的研究笔记**
1. **医疗诊断**：AI在影像识别（如X光、MRI）中准确率接近专家水平，病理分析可辅助癌症检测，早期预测模型（如糖尿病、阿尔茨海默病）提升干预效率。...✅ [Writer] 输出草稿:   **AI对医疗行业的影响**
AI正在深刻变革医疗行业。在**诊断领域**，AI影像识别（如X光、MRI）准确率媲美专家，病理分析助力癌症早期筛查，预测模型提升慢性病干预效率。**医疗管理**方面，AI优化电子病历处理、预测患者流量以合理分配资源，自动化分诊缩短等待时间。**药物研发**中，AI加速靶点筛选（如AlphaFold）、优化临床试验，推动个性化药物开发。**医患互动**通过虚拟助手、远程医疗和可穿戴设备，提升服务可及性与实时监测能力。
然而，AI应用面临**数据隐私**、**算法偏见**和**监管滞后**等挑战，需通过跨学科协作完善伦理与法律框架。总体而言，AI显著提升医疗效率与精准度，但需平衡创新与风险，以实现可持续发展。 （199字）
==================================================👀 [Reviewer] 评审当前草稿  草稿内容: **AI对医疗行业的影响**
AI正在深刻变革医疗行业。在**诊断领域**，AI影像识别（如X光、MRI）准确率媲美专家，病理分析助力癌症早期筛查，预测模型提升慢性病干预效率。**医疗管理**方面，AI优化电子病历处理、预测患者流量以合理分配资源，自动化分诊缩短等待时间。**药物研发**中，AI加速靶点筛选（如AlphaFold）、优化临床试验，推动个性化药物开发。**医患互动**通过虚拟助手、远程医疗和可穿戴设备，提升服务可及性与实时监测能力。
然而，AI应用面临**数据隐私**、**算法偏见**和**监管滞后**等挑战，需通过跨学科协作完善伦理与法律框架。总体而言，AI显著提升医疗效率与精准度，但需平衡创新与风险，以实现可持续发展。 （199字）✅ [Reviewer] 评审结果:   👍 批准通过！
----------------------------------------🚦 [Router] 决策中...  当前修订次数: 0  是否批准: True  → 决定: 流程结束（已批准）
============================================================✅ 最终报告：**AI对医疗行业的影响**
AI正在深刻变革医疗行业。在**诊断领域**，AI影像识别（如X光、MRI）准确率媲美专家，病理分析助力癌症早期筛查，预测模型提升慢性病干预效率。**医疗管理**方面，AI优化电子病历处理、预测患者流量以合理分配资源，自动化分诊缩短等待时间。**药物研发**中，AI加速靶点筛选（如AlphaFold）、优化临床试验，推动个性化药物开发。**医患互动**通过虚拟助手、远程医疗和可穿戴设备，提升服务可及性与实时监测能力。
然而，AI应用面临**数据隐私**、**算法偏见**和**监管滞后**等挑战，需通过跨学科协作完善伦理与法律框架。总体而言，AI显著提升医疗效率与精准度，但需平衡创新与风险，以实现可持续发展。 （199字）
📊 执行统计:  - 总修订次数: 0  - 是否最终批准: 是

以上的示例中，这个工作流具备了目标驱动、自主决策、多智能体协作、自我反思与迭代等特点。

目标驱动

整个工作流围绕一个明确的用户问题 question 展开，所有 Agent 的行为都服务于回答该问题。

入口状态包含 "question": "AI 对医疗行业有哪些影响？"
所有 Agent 的行为都服务于回答这个问题：
Planner 制定针对该问题的研究计划
Researcher 搜集与该问题相关的信息
Writer 撰写对该问题的回答
Reviewer 评估回答是否准确、简洁、有洞察力（即是否达成“高质量回答”这一子目标）

系统不是随机生成内容，而是以“高质量回答用户问题”为终极目标，所有步骤都是为目标服务的——典型的目标驱动架构。

自主决策

每个 Agent 基于当前状态独立生成输出，Reviewer 能动态决定流程走向。

每个节点函数（如 planner_agent）不依赖外部指令，而是：
读取共享状态（如 state['question']）
调用 LLM 自主生成下一步内容（如 plan、notes、draft）
最关键的是 reviewer_agent 和 route_after_review：
Reviewer 自主判断草稿是否合格（通过 LLM 评审）
路由函数根据 approved 和 revision_count自主决定是结束还是返回修改

没有中央控制器硬编码“第几步该做什么”，而是每个 Agent 根据上下文自主推理并行动，符合“自主决策”原则。

多智能体协作

✅ 体现方式：四个角色分工明确，通过共享状态接力协作。

Column 1	Column 2	Column 3	Column 4
智能体	职责	依赖输入	输出贡献
Planner	制定研究路径	用户问题	`plan`
Researcher	信息搜集	`plan`	`research_notes`
Writer	内容生成	`research_notes`	`draft`
Reviewer	质量把关	`draft`	`approved` / `feedback`

它们不直接调用彼此，而是通过读写 ResearchState 进行松耦合协作
流程图构成一个协作流水线 + 反馈环，体现“团队合作”模式

不是单个大模型端到端生成，而是多个专业化“角色”像团队一样协同完成复杂任务——这是多智能体系统的核心思想。

自我反思与迭代

系统具备“写 → 评 → 改”的反馈循环，实现自我优化。

Reviewer 不仅判断好坏，还提供具体修改建议（如“请补充临床应用案例”）；
Writer 在下一轮会结合反馈重写（prompt 中显式加入 feedback）；
虽然最多允许 2 次修订，体现有限但有效的自我修正能力。（也可以设置更多得次数）

系统能识别自身输出的不足，并主动改进——这是“反思”的本质。