Chapter 6 学习笔记：框架开发实践

学习如何使用两个主流的多智能体框架——Microsoft AutoGen 和 LangChain 的 LangGraph——快速构建智能体应用。内容涵盖两大部分： AutoGen 多智能体团队协作和 LangGraph 状态图工作流。

---

Github地址：agent学习笔记与实践 https://github.com/m12305/hello-agent

第一部分：AutoGen 多智能体框架

1.1 什么是 AutoGen？

AutoGen 是微软开源的、一个用于构建多智能体对话系统的框架。它提供了一系列组件，帮助开发者快速构建多智能体协作团队，让不同角色的智能体按照预设的流程进行对话和任务协作。

核心设计理念：

• 智能体角色化：每个智能体有独立的 system prompt（角色定义）和职责边界
• 对话驱动：智能体通过多轮对话来传递信息和推进任务
• 可编排：通过 RoundRobinGroupChat 等团队容器控制智能体的交互顺序和终止条件

1.2 环境依赖

autogen-agentchat
autogen-ext[openai,azure]
openai>=1.0.0
streamlit>=1.28.0
requests>=2.31.0

核心依赖包括三个层面：

依赖	用途
autogen-agentchat	AutoGen 的对话智能体核心库（AssistantAgent、团队聊天等）
autogen-ext[openai]	AutoGen 的模型扩展（OpenAIChatCompletionClient）
streamlit	用于生成最终交付的 Web 应用

1.3 模型客户端初始化

AutoGen 提供了模型抽象层，支持接入任何兼容 OpenAI 接口的服务：

from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_core.models import ModelInfo, ModelFamily

def create_openai_model_client():
    """创建 OpenAI 模型客户端"""
    return OpenAIChatCompletionClient(
        model=os.getenv("LLM_MODEL_ID", "gpt-4o"),
        api_key=os.getenv("LLM_API_KEY"),
        base_url=os.getenv("LLM_BASE_URL", "https://api.openai.com/v1"),
        model_info=ModelInfo(
            family=ModelFamily.ANY,
            vision=False,
            function_calling=True,
            json_output=True,
        )
    )

ModelInfo 关键字段：

字段	说明
family	模型家族（ModelFamily.ANY 表示通用）
vision	是否支持图像理解
function_calling	是否支持函数/工具调用（设为 True 是构建智能体的关键）
json_output	是否支持 JSON 格式输出

1.4 智能体角色定义

本案例构建了一个四人软件开发团队，每个角色通过不同的 system prompt 获得专业能力：

1.4.1 产品经理（Product Manager）

def create_product_manager(model_client):
    system_message = """你是一位经验丰富的产品经理，专门负责软件产品的需求分析和项目规划。

你的核心职责包括：
1. **需求分析**：深入理解用户需求，识别核心功能和边界条件
2. **技术规划**：基于需求制定清晰的技术实现路径
3. **风险评估**：识别潜在的技术风险和用户体验问题
4. **协调沟通**：与工程师和其他团队成员进行有效沟通

当接到开发任务时，请按以下结构进行分析：
1. 需求理解与分析
2. 功能模块划分
3. 技术选型建议
4. 实现优先级排序
5. 验收标准定义

请简洁明了地回应，并在分析完成后说"请工程师开始实现"。"""

    return AssistantAgent(
        name="ProductManager",
        model_client=model_client,
        system_message=system_message,
    )

1.4.2 软件工程师（Engineer）

def create_engineer(model_client):
    system_message = """你是一位资深的软件工程师，擅长 Python 开发和 Web 应用构建。
    ...
    请提供完整的可运行代码，并在完成后说"请代码审查员检查"。"""

    return AssistantAgent(
        name="Engineer",
        model_client=model_client,
        system_message=system_message,
    )

1.4.3 代码审查员（Code Reviewer）

def create_code_reviewer(model_client):
    system_message = """你是一位经验丰富的代码审查专家，专注于代码质量和最佳实践。
    ...
    完成后说"代码审查完成，请用户代理测试"。"""

    return AssistantAgent(
        name="CodeReviewer",
        model_client=model_client,
        system_message=system_message,
    )

1.4.4 用户代理（User Proxy）

def create_user_proxy():
    return UserProxyAgent(
        name="UserProxy",
        description="""用户代理，负责以下职责：
1. 代表用户提出开发需求
2. 执行最终的代码实现
3. 验证功能是否符合预期
4. 提供用户反馈和建议

完成测试后请回复 TERMINATE。""",
    )

UserProxyAgent 与 AssistantAgent 的区别：UserProxyAgent 代表用户端的代理人——它不调用 LLM，而是作为用户的"替身"来执行操作、提供反馈。它通常用于模拟用户输入、执行代码或作为流程的最终确认者。

1.5 智能体角色定义的设计模式

设计要素	实现方式	说明
角色名称	name="ProductManager"	显示的标识符，在团队对话中标识发言人
角色人设	system_message="..."	定义专业领域、职责边界、输出格式——是 Prompt Engineering 的核心
流转信号	结尾关键词如 "请工程师开始实现"	虽然 AutoGen 按轮询顺序流转，但这些信号帮助 LLM 理解角色间交接逻辑
模型共享	所有 AssistantAgent 使用同一个 model_client	每个 agent 可以配置不同的模型，但通常共享以提高效率

1.6 团队编排：RoundRobinGroupChat

RoundRobinGroupChat 是 AutoGen 提供的多智能体轮询群聊容器，它将多个智能体按固定顺序依次激活：

from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination

# 终止条件：检测到 "TERMINATE" 关键词时停止
termination = TextMentionTermination("TERMINATE")

# 创建轮询团队
team_chat = RoundRobinGroupChat(
    participants=[
        product_manager,    # 第1顺位
        engineer,           # 第2顺位
        code_reviewer,      # 第3顺位
        user_proxy          # 第4顺位
    ],
    termination_condition=termination,
    max_turns=20,      # 安全上限：防止无限对话
)

组件解析：

组件	作用
participants	智能体参与者的有序列表——决定了对话的顺序
termination_condition	终止条件规则——满足即停止
max_turns	最大对话轮数——安全护栏，防止死循环

RoundRobinGroupChat 的执行流程：

                     ┌──────────────┐
                     │   Task 输入   │
                     └──────┬───────┘
                            ▼
              ┌─────────────────────────┐
              │   RoundRobin GroupChat  │
              │                         │
              │  ProductManager ──────► │  "请工程师开始实现"
              │  Engineer ───────────► │  "请代码审查员检查"
              │  CodeReviewer ───────► │  "请用户代理测试"
              │  UserProxy ──────────► │  "TERMINATE" → 终止
              └─────────────────────────┘

1.7 TextMentionTermination 终止条件

termination = TextMentionTermination("TERMINATE")

这是 AutoGen 提供的文本关键词终止器——当任意智能体的回复中包含 "TERMINATE" 时，整个团队对话立即结束。这是最简洁可靠的终止方式，避免了复杂的条件判断。

1.8 控制台实时输出

from autogen_agentchat.ui import Console

# Console 会将团队对话过程实时打印到控制台
result = await Console(team_chat.run_stream(task=task))

Console 是 AutoGen 的可视化工具——它将多智能体之间的对话流逐条输出到终端，便于开发者观察团队协作的全过程。

1.9 完整任务示例：比特币价格追踪应用

task = """我们需要开发一个比特币价格显示应用，具体要求如下：

核心功能：
- 实时显示比特币当前价格（USD）
- 显示24小时价格变化趋势（涨跌幅和涨跌额）
- 提供价格刷新功能

技术要求：
- 使用 Streamlit 框架创建 Web 应用
- 界面简洁美观，用户友好
- 添加适当的错误处理和加载状态

请团队协作完成这个任务，从需求分析到最终实现。"""

典型协作对话流：

👤 ProductManager：
   需求分析：这是一个实时价格展示应用，核心是数据获取与展示...
   技术选型：建议使用Streamlit + CoinGecko API...
   请工程师开始实现。

👤 Engineer：
   已根据需求规划完成Streamlit应用的完整实现。
   代码包含错误处理、数据缓存、刷新功能...
   请代码审查员检查。

👤 CodeReviewer：
   代码审查完成。
   优点：结构清晰，错误处理完善...
   建议：可考虑添加WebSocket实时推送...
   代码审查完成，请用户代理测试。

👤 UserProxy：
   已验证应用功能，数据展示正确，刷新功能可用。
   TERMINATE

1.10 AutoGen 设计模式总结

                    AutoGen 多智能体协作架构

┌────────────────────────────────────────────────────────────────┐
│  ┌──────────────────────┐                                      │
│  │ OpenAIChatCompletion │  ← 统一的模型客户端                    │
│  │       Client          │                                      │
│  └─────────┬────────────┘                                      │
│            │                                                    │
│            ▼                                                    │
│  ┌─────────────────────────────────────────────┐               │
│  │          RoundRobinGroupChat                │               │
│  │  (轮询群聊 — 固定顺序 + 终止条件)              │               │
│  │                                             │               │
│  │  ┌──────────────┐  ┌──────────────┐        │               │
│  │  │   PM Agent   │→│  Engineering │        │               │
│  │  │  (需求分析)    │  │  Agent (编码)  │       │               │
│  │  └──────────────┘  └──────────────┘        │               │
│  │         ↓                  ↓                │               │
│  │  ┌──────────────┐  ┌──────────────┐        │               │
│  │  │ CodeReviewer│←│  UserProxy   │        │               │
│  │  │  (审查)       │  │  (测试/终止)   │       │               │
│  │  └──────────────┘  └──────────────┘        │               │
│  └─────────────────────────────────────────────┘               │
│                              │                                  │
│                              ▼                                  │
│                    ┌─────────────────┐                         │
│                    │  最终交付物       │                         │
│                    │ (Streamlit App) │                         │
│                    └─────────────────┘                         │
└────────────────────────────────────────────────────────────────┘

---

第二部分：LangGraph 有状态图工作流

2.1 什么是 LangGraph？

LangGraph 是 LangChain 生态中的一个有状态图编排框架。它将智能体的执行流程建模为一个有向图（DAG），每个节点是一个处理函数，边定义了数据流转的方向。

核心设计理念：

• Graph-based：用图的节点（Node）和边（Edge）来描述工作流
• State-driven：所有节点共享一个状态对象，数据在图中有序流转
• Checkpointing：内置记忆模块（InMemorySaver），支持状态持久化与回溯

2.2 环境依赖

langgraph>=1.0.0
langchain_openai>=0.3.0
python-dotenv
tavily-python

2.3 状态定义：TypedDict + Annotated

LangGraph 通过 TypedDict 定义共享状态的结构，使用 Annotated 注解状态字段的合并策略：

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class SearchState(TypedDict):
    messages: Annotated[list, add_messages]   # 消息列表，自动追加合并
    user_query: str                            # 用户的原始查询
    search_query: str                          # 优化后的搜索关键词
    search_results: str                        # Tavily 搜索结果
    final_answer: str                          # 最终答案
    step: str                                  # 当前步骤标识

状态字段设计原则：

字段	类型	作用	数据流向
messages	Annotated[list, add_messages]	对话消息历史，add_messages 会自动追加而非覆盖	贯穿全流程
user_query	str	记录原始查询，供后续节点引用	第1节点写入，第3节点读取
search_query	str	保存优化后的搜索词	第1节点写入，第2节点读取
search_results	str	保存搜索结果	第2节点写入，第3节点读取
final_answer	str	最终回答	第3节点写入
step	str	当前阶段标记（用于错误处理分支）	每节点更新

add_messages 注解：这是 LangGraph 提供的特殊 reducer 函数。普通字段在节点返回新值时会被覆盖，但 Annotated[list, add_messages] 则会将新消息追加到现有列表中——这是实现对话历史的正确方式。

2.4 节点函数实现

每个节点是一个纯函数，接收 SearchState，返回 SearchState 的部分更新。

2.4.1 理解查询节点（understand）

def understand_query_node(state: SearchState) -> SearchState:
    """提取最近的用户消息，生成优化后的搜索关键词"""

    # 从消息列表中获取最新的用户消息
    user_message = ""
    for msg in reversed(state["messages"]):
        if isinstance(msg, HumanMessage):
            user_message = msg.content
            break

    understand_prompt = f"""分析用户的查询："{user_message}"

请完成两个任务：
1. 简洁总结用户想要了解什么
2. 生成最适合搜索的关键词（中英文均可，要精准）

格式：
理解：[用户需求总结]
搜索词：[最佳搜索关键词]"""

    response = llm.invoke([SystemMessage(content=understand_prompt)])

    # 从响应中提取搜索关键词
    search_query = user_message  # 默认使用原始查询
    if "搜索词：" in response.content:
        search_query = response.content.split("搜索词：")[1].strip()

    return {
        "user_query": user_message,
        "search_query": search_query,
        "step": "understand",
        "messages": [AIMessage(content=f"我理解您的需求：{response.content}")]
    }

2.4.2 搜索节点（search）

def tavily_search_node(state: SearchState) -> SearchState:
    search_query = state["search_query"]

    try:
        response = tavily_client.search(
            query=search_query,
            search_depth="basic",
            include_answer=True,       # 请求 Tavily 的综合摘要
            max_results=5
        )

        search_results = ""
        # 优先使用 Tavily 的综合答案
        if response.get("answer"):
            search_results = f"综合答案：\n{response['answer']}\n\n"

        # 追加具体搜索结果
        if response.get("results"):
            search_results += "相关信息：\n"
            for i, result in enumerate(response["results"][:3], 1):
                search_results += f"{i}. {result.get('title', '')}\n"
                search_results += f"{result.get('content', '')}\n"
                search_results += f"来源：{result.get('url', '')}\n\n"

        return {
            "search_results": search_results,
            "step": "searched",
            "messages": [AIMessage(content="✅ 搜索完成！正在为您整理答案...")]
        }

    except Exception as e:
        return {
            "search_results": f"搜索失败：{e}",
            "step": "search_failed",
            "messages": [AIMessage(content="❌ 搜索遇到问题...")]
        }

2.4.3 生成答案节点（answer）

def generate_answer_node(state: SearchState) -> SearchState:
    # 如果搜索失败，走降级路径
    if state["step"] == "search_failed":
        fallback_prompt = f"""搜索API暂时不可用，请基于您的知识回答..."""
        response = llm.invoke([SystemMessage(content=fallback_prompt)])
        return {"final_answer": response.content, "step": "completed", ...}

    # 正常路径：基于搜索结果生成答案
    answer_prompt = f"""基于以下搜索结果为用户提供完整、准确的答案：
用户问题：{state['user_query']}
搜索结果：{state['search_results']}
要求：综合搜索结果、引用来源、结构清晰..."""

    response = llm.invoke([SystemMessage(content=answer_prompt)])
    return {
        "final_answer": response.content,
        "step": "completed",
        "messages": [AIMessage(content=response.content)]
    }

2.5 图构建：StateGraph

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

def create_search_assistant():
    # 1. 创建状态图
    workflow = StateGraph(SearchState)

    # 2. 添加节点
    workflow.add_node("understand", understand_query_node)
    workflow.add_node("search", tavily_search_node)
    workflow.add_node("answer", generate_answer_node)

    # 3. 设置线性边：START → understand → search → answer → END
    workflow.add_edge(START, "understand")
    workflow.add_edge("understand", "search")
    workflow.add_edge("search", "answer")
    workflow.add_edge("answer", END)

    # 4. 编译（含记忆模块）
    memory = InMemorySaver()
    app = workflow.compile(checkpointer=memory)

    return app

图结构可视化：

     ┌──────────┐
     │  START   │
     └────┬─────┘
          │
          ▼
   ┌──────────────┐
   │  understand  │  查询分析 + 搜索词优化
   │  Node        │
   └──────┬───────┘
          │
          ▼
   ┌──────────────┐
   │   search     │  Tavily API 真实搜索
   │   Node       │
   └──────┬───────┘
          │
          ▼
   ┌──────────────┐
   │   answer     │  合成最终回答
   │   Node       │
   └──────┬───────┘
          │
          ▼
     ┌──────────┐
     │   END    │
     └──────────┘

2.6 Checkpointing 记忆机制

from langgraph.checkpoint.memory import InMemorySaver

memory = InMemorySaver()
app = workflow.compile(checkpointer=memory)

InMemorySaver 是 LangGraph 的内置检查点管理器，它在每个节点执行后自动保存当前状态。在对话应用中，每次调用时传入不同的 thread_id 即可实现多会话隔离：

config = {"configurable": {"thread_id": f"search-session-{session_count}"}}

# 同一 thread_id 下的多次调用会共享状态（对话历史）
async for output in app.astream(initial_state, config=config):
    ...

2.7 流式执行：astream

async for output in app.astream(initial_state, config=config):
    for node_name, node_output in output.items():
        if "messages" in node_output and node_output["messages"]:
            latest_message = node_output["messages"][-1]
            if isinstance(latest_message, AIMessage):
                if node_name == "understand":
                    print(f"🧠 理解阶段: {latest_message.content}")
                elif node_name == "search":
                    print(f"🔍 搜索阶段: {latest_message.content}")
                elif node_name == "answer":
                    print(f"\n💡 最终回答:\n{latest_message.content}")

astream 是 LangGraph 的异步流式执行方法，每次节点完成执行后 yield 一次输出。通过监不听同节点的输出，可以实现分阶段展示（理解阶段 → 搜索阶段 → 最终回答）的实时反馈效果。

2.8 条件边与动态路由

LangGraph 最强大的功能之一就是条件边（Conditional Edges）——它通过一个判断函数来检查当前状态，然后动态地决定下一步应该跳转到哪个节点。这与线性边（add_edge）的"固定流向"形成鲜明对比，让工作流具备了分支决策能力。

2.8.1 线性边 vs 条件边

线性边（静态路由）：                 条件边（动态路由）：

  Node A                               Node A
    │                                     │
    ▼                              ┌──────┴──────┐
  Node B         固定路径           │  判断函数    │  运行时决策
    │                              └──────┬──────┘
    ▼                              ┌──────┼──────┐
  Node C                           ▼      ▼      ▼
                                Node B Node C Node D

2.8.2 条件边工作原理

条件边的核心是路由函数（Router Function）——一个接收当前状态、返回目标节点名称的 Python 函数：

def router(state: SearchState) -> str:
    """根据搜索是否成功决定下一步"""
    if state["step"] == "search_failed":
        return "fallback_answer"    # 搜索失败 → 走降级路径
    else:
        return "generate_answer"    # 搜索成功 → 走正常路径

LangGraph 在每个节点执行后调用路由函数，根据返回的字符串决定下一个要执行的节点。路由函数必须是纯函数——同样的状态输入始终返回同样的路由结果。

2.8.3 判断函数创建示例

以下是三个典型场景的路由函数示例：

场景一：基于步骤状态的路由

def route_by_step(state: SearchState) -> str:
    """根据 step 字段判断当前处于哪个阶段，决定下一步"""
    if state["step"] == "search_failed":
        return "fallback"
    elif state["step"] == "completed":
        return END
    else:
        return "continue"

场景二：基于搜索结果质量的路由

def route_by_result_quality(state: SearchState) -> str:
    """检查搜索结果是否足够完整"""
    if not state["search_results"] or "抱歉" in state["search_results"]:
        return "rewrite_query"     # 结果不满意 → 重写查询词重新搜索
    elif "综合答案" in state["search_results"]:
        return "generate_answer"   # 有综合答案 → 直接生成回答
    else:
        return "summarize"         # 仅有碎片结果 → 先汇总再回答

场景三：基于意图分类的路由（多分支）

def route_by_intent(state: SearchState) -> str:
    """根据用户意图分发到不同处理节点"""
    query = state["user_query"].lower()
    if any(word in query for word in ["天气", "weather"]):
        return "weather_handler"
    elif any(word in query for word in ["代码", "code", "编程"]):
        return "code_handler"
    elif any(word in query for word in ["翻译", "translate"]):
        return "translate_handler"
    else:
        return "general_search"

2.8.4 条件边的添加

使用 add_conditional_edges() 将路由函数绑定到图中：

from langgraph.graph import StateGraph, START, END

def create_dynamic_search_assistant():
    workflow = StateGraph(SearchState)

    # 添加所有可能的节点
    workflow.add_node("understand", understand_query_node)
    workflow.add_node("search", tavily_search_node)
    workflow.add_node("generate_answer", generate_answer_node)
    workflow.add_node("fallback_answer", fallback_answer_node)  # 降级路径节点

    # 线性边：固定的起始和必经路径
    workflow.add_edge(START, "understand")
    workflow.add_edge("understand", "search")

    # 条件边：搜索节点之后，根据结果动态路由
    workflow.add_conditional_edges(
        "search",                    # 源节点
        route_by_step,               # 路由判断函数
        {                            # 路由映射表
            "generate_answer": "generate_answer",
            "fallback": "fallback_answer",
        }
    )

    # 两条路径最终都到达 END
    workflow.add_edge("generate_answer", END)
    workflow.add_edge("fallback_answer", END)

    return workflow.compile()

add_conditional_edges 参数解析：

参数	说明	示例
source	源节点名称，条件边从该节点出发	"search"
path	路由函数，签名为 (state) -> str	route_by_step
path_map	路由函数返回值 → 目标节点名称的映射字典	{"fallback": "fallback_answer", ...}

注意：路由函数返回的字符串必须与 path_map 中的键完全匹配，否则 LangGraph 会抛出异常。映射表的设计让路由逻辑和节点名称解耦——路由函数只需返回语义化的键名（如 "fallback"），映射表负责将其关联到实际的节点（如 "fallback_answer"）。

2.8.5 条件边改造后的完整图结构

          ┌──────────┐
          │  START   │
          └────┬─────┘
               │          线性边
               ▼
        ┌──────────────┐
        │  understand  │
        └──────┬───────┘
               │          线性边
               ▼
        ┌──────────────┐
        │   search     │
        └──────┬───────┘
               │
          ┌────┴────┐     条件边（动态路由）
          │ 路由函数  │
          └────┬────┘
               │
      ┌────────┼────────┐
      ▼                 ▼
┌───────────┐    ┌───────────────┐
│ generate  │    │  fallback     │
│ _answer   │    │  _answer      │
└─────┬─────┘    └───────┬───────┘
      │                  │          线性边
      ▼                  ▼
   ┌──────┐          ┌──────┐
   │ END  │          │ END  │
   └──────┘          └──────┘

对比最早只有一条线性路径的图，加入条件边后，工作流具备了自动容错能力——搜索失败时不再强行生成低质量答案，而是优雅地走降级路径。

2.8.6 条件边与多轮循环

条件边不仅支持分支路由，还可以通过回环实现多轮迭代——让路由函数返回上游节点的名称，形成循环：

def route_with_retry(state: SearchState) -> str:
    """搜索失败时返回搜索节点重试，最多重试3次"""
    if state["step"] == "search_failed" and state.get("retry_count", 0) < 3:
        state["retry_count"] = state.get("retry_count", 0) + 1
        return "search"                # 回到搜索节点，形成循环
    elif state["step"] == "search_failed":
        return "fallback_answer"       # 重试用尽，走降级
    else:
        return "generate_answer"

workflow.add_conditional_edges(
    "search",
    route_with_retry,
    {
        "search": "search",              # 回环！重新执行搜索
        "fallback_answer": "fallback_answer",
        "generate_answer": "generate_answer",
    }
)

   ┌─────────┐
   │  search │◄──────────┐
   └────┬────┘           │
        │                │  回环：重试
   ┌────┴────┐           │
   │ 路由函数  │── 失败且  ─┘
   │         │   未超重试上限
   └────┬────┘
        │ 成功或重试用尽
        ▼
   ┌──────────┐
   │  answer  │
   └──────────┘

注意：使用回环时务必设置退出条件（如最大重试次数），否则可能形成无限循环。

2.9 LangGraph vs AutoGen 对比

对比维度	AutoGen	LangGraph
核心范式	多智能体角色对话	有向图 + 状态流转
编排方式	RoundRobinGroupChat 轮询	StateGraph 节点+边
节点类型	AssistantAgent / UserProxyAgent	普通 Python 函数
状态管理	对话历史（隐式）	TypedDict + Checkpointer（显式）
终止条件	TextMentionTermination 关键词	图边流向 END 节点
工具调用	function_calling（模型原生支持）	在节点函数中手动调用
适用场景	多角色协作（PM→Dev→Review）	流水线式工作流（搜索→分析→回答）
学习曲线	中等（理解 Agent 类型和 Team 配置）	中高（理解图、状态、边的概念）
框架生态	微软 AutoGen 生态	LangChain 生态（LangSmith、LangServe等）

2.10 框架选择指南

你的任务是什么类型的？
    │
    ├── 多角色协作（多人讨论、角色扮演）
    │   └── AutoGen：天然的智能体角色 + 群聊容器
    │
    ├── 固定流程管线（检索→处理→生成）
    │   └── LangGraph：节点+边建模清晰，状态管理显式
    │
    ├── 需要复杂的条件分支和回溯
    │   └── LangGraph：支持条件边、子图、人机交互暂停
    │
    └── 快速原型 + 简单工具调用
        └── 两者均可，AutoGen 上手更快

---

本章文件结构

chapter6/
├── AutoGen/
│   ├── autogen_team.py       # AutoGen 多智能体软件开发团队
│   ├── output.py             # Streamlit 比特币价格追踪应用（团队交付物）
│   └── requirement.txt       # AutoGen 项目依赖
│
└── langgraph/
    ├── langgraph_demo.py     # LangGraph 智能搜索助手
    └── requirement.txt       # LangGraph 项目依赖