想象一下这个业务场景你会怎么做：

需要自动生成一份《关于某米产品的市场分析报告》，包含：
1）市场规模概况
2）竞品分析
3）用户口碑
最后综合成一份报告。

一、版本 1：Chain = 单线程流水线（一步一步顺序做）

思路：

  1. 先查「市场规模」
  2. 再查「竞品」
  3. 再查「用户口碑」
  4. 最后把三个结果交给 LLM 写成报告

一段从上到下顺序执行的脚本。

1.1 关键代码（使用 LangChain 做一个顺序 Chain）

下面示例用 OpenAI，你可以换成自己模型（比如通义、DeepSeek，对应的 LangChain 封装）：

# 安装（按需）
# pip install "langchain>=0.3.0" "langchain-openai" 

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# 设置环境变量并加载 OPENAI_API_KEY
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)


# ====== 1. 假装有三个“同步函数”，分别做三种查询（单线程顺序执行） ======
# 这里为了演示，先用 print + 伪结果，真实场景你可以接搜索 API、向量库等

def search_market_size(product: str) -> str:
    print("【单线程】正在查询市场规模...")
    # TODO: 调用真实搜索/RAG
    return f"{product} 的市场规模大概是多少（这里是伪造的搜索结果）"

def search_competitors(product: str) -> str:
    print("【单线程】正在查询竞品...")
    # TODO: 调用真实搜索/RAG
    return f"{product} 的主要竞品有哪些（这里是伪造的搜索结果）"

def search_user_reviews(product: str) -> str:
    print("【单线程】正在查询用户口碑...")
    # TODO: 调用真实搜索/RAG
    return f"{product} 在用户中的口碑如何（这里是伪造的搜索结果）"


# ====== 2. 用 LangChain 写一个“最后一跳”的 LLM 调用 ======

report_prompt = ChatPromptTemplate.from_template("""
你是一名专业市场分析师，请根据以下三部分信息，写一份结构清晰的《{product} 市场分析报告》。

【市场规模】
{market}

【竞品情况】
{competitors}

【用户口碑】
{reviews}

请至少包含：市场概况、竞品对比、用户反馈、小结与建议四个部分。
""")

report_chain = report_prompt | llm  # LCEL 链式写法


# ====== 3. 把“单线程流水线”串起来 ======

def generate_report_with_chain(product: str) -> str:
    # —— 关键：这 3 步是严格“顺序执行”，是单线程逻辑 ——
    market = search_market_size(product)
    competitors = search_competitors(product)
    reviews = search_user_reviews(product)

    # 最后一步交给 LLM，生成报告
    result = report_chain.invoke({
        "product": product,
        "market": market,
        "competitors": competitors,
        "reviews": reviews,
    })
    return result.content


if __name__ == "__main__":
    report = generate_report_with_chain("某米手表")
    print(report)

要点：

  • 这就是典型的 Chain 思维 = 单线程顺序执行：
    • 查市场 → 查竞品 → 查口碑 → 生成报告
  • 每一步的顺序都是你写死在代码里的，不会变，也不会并行。

二、版本 2：单 Agent = 会自己规划步骤的“智能单线程”

现在换一种思路：

我不再写死「先查什么后查什么」，而是：

  • 给大模型一个“任务目标”：写市场报告
  • 给它几把“工具”：查市场规模、查竞品、查口碑
  • 让它自己决定：
    • 先调用哪个工具
    • 需不需要多查几次
    • 查完才写报告

执行本质仍然是一条线（循环），所以可以类比“智能单线程”。

2.1 定义工具（Tools）

使用 LangChain 的 @tool 装饰器，把原来的 3 个函数变成 Agent 可调用的工具：

from langchain_core.tools import tool

@tool
def market_size_tool(product: str) -> str:
    """查询某个产品的市场规模（示例工具）。"""
    print("【Agent】调用 market_size_tool")
    # TODO: 真实实现
    return f"[工具返回] {product} 的市场规模信息"

@tool
def competitors_tool(product: str) -> str:
    """查询某个产品的主要竞品。"""
    print("【Agent】调用 competitors_tool")
    return f"[工具返回] {product} 的竞品信息"

@tool
def user_reviews_tool(product: str) -> str:
    """查询某个产品的用户口碑。"""
    print("【Agent】调用 user_reviews_tool")
    return f"[工具返回] {product} 的用户口碑信息"

2.2 创建一个“单 Agent”

用 LangChain 的工具调用 Agent（基于 OpenAI 的 tool calling）：

# 安装（按需）
# pip install "langchain" "langchain-openai"

from langchain.agents import create_openai_tools_agent, AgentExecutor

tools = [market_size_tool, competitors_tool, user_reviews_tool]

system_prompt = """
你是一名资深市场分析师。
你的任务是：为用户指定的产品，写一份完整的市场分析报告。
你可以使用以下工具查询市场规模、竞品、用户口碑。
请在需要时多次调用工具，最后给出一份结构化的报告。
"""

# 创建“会用工具”的 Agent
agent = create_openai_tools_agent(
    llm=llm,
    tools=tools,
    prompt=ChatPromptTemplate.from_messages([
        ("system", system_prompt),
        ("human", "{input}")
    ]),
)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,         # 打印中间工具调用过程
    max_iterations=10,    # 防止死循环
)

2.3 调用单 Agent

def generate_report_with_single_agent(product: str) -> str:
    query = f"请为产品「{product}」写一份市场分析报告，必要时自己查资料。"
    result = agent_executor.invoke({"input": query})
    return result["output"]

if __name__ == "__main__":
    report = generate_report_with_single_agent("某智能手表")
    print(report)

执行时你会看到类似日志：

【Agent】调用 market_size_tool
【Agent】调用 competitors_tool
【Agent】调用 user_reviews_tool
...

区别在哪儿？

  • Chain（版本 1）：
    • 调用顺序：你写死 = 单线程脚本
  • 单 Agent（版本 2）：
    • 调用顺序：LLM 自己决定，但整体仍是一条线在跑 = 智能单线程

从“线程模型”看，它还不是“多线程”，只是更聪明的单线程。

三、版本 3：多 Agent 编排 = 多线程调度（并行干活）

你不止有一个 Agent，而是有三个：

  • 市场规模 Agent
  • 竞品分析 Agent
  • 用户口碑 Agent

它们可以并行干活，最后再由一个“总编辑 Agent”汇总成报告。

再加一个“调度器”，负责：

  • 启动这三个 Agent
  • 等它们都完成
  • 把结果交给总编辑 Agent

这就很像：

  • 三个工作线程：thread_market / thread_competitor / thread_reviews
  • 一个主线程：负责合并结果

在 LangChain 生态里，做这种“多 Agent + 流程图”的编排，推荐用 LangGraph。

注意：下面是骨架代码，演示结构和“并行”的思路，实际项目你可以继续细化 prompt 和工具。

3.1 安装 LangGraph

pip install "langgraph>=0.2.0" "langchain-openai"

3.2 定义整个流程的“状态”（State）

我们用一个 TypedDict 表示图里流转的状态：

from typing import TypedDict, Optional

class ReportState(TypedDict, total=False):
    product: str
    market: Optional[str]
    competitors: Optional[str]
    reviews: Optional[str]
    report: Optional[str]

3.3 准备 3 个“分析 Agent”和 1 个“总编辑 Agent”

可以复用第 2 步中的工具，也可以直接用纯 LLM。下面简单起见，用纯 LLM + Prompt（你也可以把它们做成各自有工具的 Agent）。

from langchain_core.prompts import ChatPromptTemplate

market_agent_prompt = ChatPromptTemplate.from_template("""
你是一名市场研究员，请针对产品「{product}」简要分析市场规模和趋势。
输出一段中文分析。
""")
market_agent_chain = market_agent_prompt | llm

competitor_agent_prompt = ChatPromptTemplate.from_template("""
你是一名竞品分析师，请针对产品「{product}」简要分析主要竞品及差异化。
输出一段中文分析。
""")
competitor_agent_chain = competitor_agent_prompt | llm

review_agent_prompt = ChatPromptTemplate.from_template("""
你是一名用户研究员，请针对产品「{product}」总结用户口碑和常见反馈点（可基于你已有知识模拟）。
输出一段中文分析。
""")
review_agent_chain = review_agent_prompt | llm

synthesis_agent_prompt = ChatPromptTemplate.from_template("""
你是总编辑，请根据以下三个部分内容，写一份完整的《{product} 市场分析报告》。

【市场规模与趋势】
{market}

【竞品情况】
{competitors}

【用户口碑】
{reviews}

要求结构清晰，有小标题。
""")
synthesis_agent_chain = synthesis_agent_prompt | llm

3.4 用 LangGraph 画出“多线程流程图”

在这里插入图片描述

核心点：

  • market_node / competitor_node / review_node 可以并行执行
    （在 LangGraph 中，它们更新不同的 state 字段，调度器可以把它们并发跑）
  • 全部跑完后，进入 synthesis_node 汇总成最终报告
from langgraph.graph import StateGraph, END

# 1. 创建一个有状态的图
graph = StateGraph(ReportState)

# 2. 定义四个“节点函数”，每个节点处理一部分任务

def market_node(state: ReportState) -> ReportState:
    print("【多 Agent】market_node 开始")
    res = market_agent_chain.invoke({"product": state["product"]})
    return {"market": res.content}

def competitor_node(state: ReportState) -> ReportState:
    print("【多 Agent】competitor_node 开始")
    res = competitor_agent_chain.invoke({"product": state["product"]})
    return {"competitors": res.content}

def review_node(state: ReportState) -> ReportState:
    print("【多 Agent】review_node 开始")
    res = review_agent_chain.invoke({"product": state["product"]})
    return {"reviews": res.content}

def synthesis_node(state: ReportState) -> ReportState:
    print("【多 Agent】synthesis_node 开始汇总")
    res = synthesis_agent_chain.invoke({
        "product": state["product"],
        "market": state.get("market", ""),
        "competitors": state.get("competitors", ""),
        "reviews": state.get("reviews", ""),
    })
    return {"report": res.content}


# 3. 往图里加节点
graph.add_node("market", market_node)
graph.add_node("competitor", competitor_node)
graph.add_node("review", review_node)
graph.add_node("synthesis", synthesis_node)

# 4. 设置入口节点：先从三个分析节点开始
#    注意：这里我们模拟“并行”——LangGraph 可以让这三个节点并发执行
graph.set_entry_point("market")
graph.add_edge("market", "competitor")
graph.add_edge("market", "review")

# 更形象的写法（用条件或子图做 fan-out/fan-in 会更优雅，这里简化理解）：
# 实际工程中，常用一个“dispatcher 节点”一次性触发三个分支，然后汇总到 synthesis

# 5. 当 competitor 和 review 都跑完后，进入 synthesis
graph.add_edge("competitor", "synthesis")
graph.add_edge("review", "synthesis")

# 6. synthesis 结束后，图结束
graph.add_edge("synthesis", END)

# 7. 编译图
app = graph.compile()

上面第 4 步示意写法稍微“直线化”了一点，真实项目里通常会：

  • 用一个 dispatcher 节点启动 3 个分支（天然并行）
  • 用 LangGraph 的「并行分支 + 汇合」模式

关键是：这三个节点逻辑上是可并行的，不像 Chain 那样必须 1→2→3 排队做。

3.5 运行多 Agent 工作流

def generate_report_with_multi_agents(product: str) -> str:
    # 初始状态只有 product
    init_state: ReportState = {"product": product}

    final_state = app.invoke(init_state)
    return final_state["report"]

if __name__ == "__main__":
    report = generate_report_with_multi_agents("某智能手表")
    print(report)

在真实部署中，如果你给 LangGraph 配上异步执行和多 worker，这三个节点可以真正并发执行，就像：

  • market 线程查市场
  • competitor 线程查竞品
  • review 线程查口碑
    主流程等三个都完成后，再跑 synthesis 节点。

这就很接近传统编程里的「多线程任务调度」。

四、从线程视角帮你再总结一遍

把刚才 3 个版本放在一起看：

1）Chain = 单线程顺序执行脚本

  • 代码里写死：先 A 再 B 再 C

  • 类似：

    a = step_A()
b = step_B(a)
c = step_C(b)

  • 适合：流程一眼能画完、步骤固定的任务

2）单 Agent = 会自己规划的智能单线程

  • 仍然是“一条线”，但每一步干啥由 LLM 决定：

    while not done:
    想下一步
    选工具
    执行

  • 适合：一个 AI 助手自己看情况多轮调用工具的场景

3）多 Agent 编排 = 多线程调度器

  • 有多个 Agent 像多个线程：
    • 市场 Agent
    • 竞品 Agent
    • 口碑 Agent
    • 总编辑 Agent
  • 工作流（LangGraph 等）像线程调度器：
    • 谁可以并行
    • 谁要等谁
    • 全部结束后再汇总
  • 适合：
    • 任务本身天然可拆分、很多子任务可并行
    • 或需要「多角色长期协作」的复杂系统

4）LangChain 的位置

  • 帮你：
    • 封装 LLM / RAG / Tools（“线程里要干的活”）
    • 快速写出 Chain（单线程流水线）
    • 快速搭出 Agent（智能单线程）
  • 再配合 LangGraph 等框架：
    • 让多个 Agent 组成「多线程 AI 团队」，做真正复杂的工作流。

现在你明白了吗？

# python # 人工智能 # 数据库 # 深度学习
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

Agent Skill 学习笔记

这篇文章介绍了AI Agent的Skill概念及其应用。Skill是为AI提供的一套可复用任务执行指南，包含流程、输入输出规范、异常处理等内容，不同于一次性Prompt。 主要内容包括： Skill的结构（YAML元数据+Markdown正文） 适合做成Skill的场景（流程明确、重复出现、经验可复用） Skill与CLI、MCP的区别与配合（CLI调用命令，Skill指导流程，MCP连接工具）

avatar AI Agent技术社区
cover

你的 Agent 架构选错了：越复杂的 Agent 系统，越可能走向失败

avatar AI Agent技术社区
cover

当 AI Agent 要同时接入 Telegram、飞书、钉钉：Channel Plugin 抽象层设计之道

avatar AI Agent技术社区