从 Prompt 到工作流：用 LCEL 构建一个 AI 应用（下）

发表于 分类于 Waline： 本文字数： 12k 阅读时长 ≈ 42 分钟

1. 上篇回顾：从问题拆解到搜索词生成

上篇完成了自动化网页研究总结器的前半段。它先从一个核心问题出发：当任务从“回答一句话”变成“完成一个研究流程”时，单次 Prompt 调用已经不够用。系统需要先理解问题，再拆解搜索方向，接入外部工具，并在多个中间结果之间保持稳定的数据结构。

围绕这个目标，上篇先解释了 LCEL 的价值：它不是为了让代码写得更短，而是为了让 LLM 应用里的数据流、组件边界和执行顺序更加清楚。Prompt -> LLM -> Parser 是最基础的链式结构；当流程变复杂后，还可以通过 RunnableLambda 接入普通函数，通过 RunnableParallel 保留上下文和并行分支，通过 .map() 批量处理列表数据。

随后，上篇把网页研究总结器拆成几条小链：角色选择链负责判断任务类型，搜索词生成链负责把用户问题拆成多个可检索的信息切面。到上篇结束时，系统已经能够从一个中文问题生成多组搜索查询，例如围绕杭州旅行同时覆盖自然景观、历史文化、城市体验、美食茶文化和实用攻略。

但这些搜索词还只是研究入口。要生成真正有依据的报告，系统还需要继续完成资料获取和信息整合：搜索 URL、抓取网页、总结来源、批量处理多个网页，并把所有摘要合并成最终 Markdown 报告。下篇就从这里开始，重点处理“搜索词之后”的链路。

2. URL 搜索链：把搜索词转换成网页来源

搜索词生成链输出的是一组查询，但这些查询本身还不是可用资料。要进入资料收集阶段，需要把每个搜索词提交给搜索工具，获取对应的网页 URL。这个过程由 URL 搜索链 完成。

与前两条链不同，URL 搜索链不需要调用大模型。它的核心任务是调用网页搜索函数，并把搜索结果整理成后续网页总结链需要的数据结构。

2.1 URL 搜索链的输入与输出

搜索词生成链的输出是一个数组，每个元素包含搜索词和原始问题，URL 搜索链每次处理其中一个元素。以第一个搜索词为例，输入是：

1
2
3
4
{
    'search_query': '杭州 必去景点 西湖 灵隐寺 西溪湿地 推荐', 
    'user_question': '杭州有哪些值得一去的地方？'
}

调用搜索函数后，得到若干 URL。为了让后续链继续使用这些信息，需要把每个 URL 包装成统一结构：

字段 作用
result_url 后续网页抓取使用
search_query 标记该网页来自哪个搜索词
user_question 保留原始研究目标

这一步对后续摘要很重要。网页总结链不仅需要 URL，还需要知道这个 URL 是围绕哪个搜索词得到的，以及最终要回答什么问题。

2.2 用 RunnableLambda 封装搜索逻辑

URL 搜索链主要依赖前面准备好的 web_search() 函数：

1
2
3
4
5
6
7
8
9
10
from typing import List
from langchain_community.utilities import DuckDuckGoSearchAPIWrapper

def web_search(web_query: str, num_results: int) -> List[str]:
    results = DuckDuckGoSearchAPIWrapper().results(
        web_query,
        num_results
    )

    return [result["link"] for result in results]

这个函数输入搜索词，输出 URL 列表。为了把它接入 LCEL，需要使用 RunnableLambda：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from langchain_core.runnables import RunnableLambda
from web_searching import web_search

NUM_SEARCH_RESULTS = 3

# URL搜索链，把一个搜索词转换成多个带上下文信息的网页 URL 对象
search_result_urls_chain = (
    RunnableLambda(lambda x: [
        {
            "result_url": url,
            "search_query": x["search_query"],
            "user_question": x["user_question"],
        }
        for url in web_search(
            web_query=x["search_query"],
            num_results=NUM_SEARCH_RESULTS
        )
    ])
)

这条链完成了两件事。

第一，调用搜索工具：

1
2
3
4
web_search(
    web_query=x["search_query"],
    num_results=NUM_SEARCH_RESULTS
)

第二，把返回的 URL 列表转换成统一字段结构：

1
2
3
4
5
{
    "result_url": url,
    "search_query": x["search_query"],
    "user_question": x["user_question"],
}

这也是 RunnableLambda 的典型用途：把普通 Python 函数包装成 LCEL 链中的一个可执行节点，并在同一步完成数据结构转换。

执行示例：

1
2
3
4
5
6
7
8
input_data = {
    'search_query': '杭州 必去景点 西湖 灵隐寺 西溪湿地 推荐', 
    'user_question': '杭州有哪些值得一去的地方？'
}

urls = search_result_urls_chain.invoke(input_data)

print(urls)

可能得到：

1
2
3
4
5
[  
	{'result_url': 'https://www.klook.com/zh-CN/activity/155336-private-guide-1-day-tour-of-hangzhou-city-by-car/', 'search_query': '杭州 必去景点 西湖 灵隐寺 西溪湿地 推荐', 'user_question': '杭州有哪些值得一去的地方？'},  
	{'result_url': 'https://www.getyourguide.com/zh-cn/lingyin-temple-l150150/', 'search_query': '杭州 必去景点 西湖 灵隐寺 西溪湿地 推荐', 'user_question': '杭州有哪些值得一去的地方？'},  
	{'result_url': 'https://www.getyourguide.com/zh-cn/hangzhou-l1241/guided-tours-tc1144/', 'search_query': '杭州 必去景点 西湖 灵隐寺 西溪湿地 推荐', 'user_question': '杭州有哪些值得一去的地方？'}  
 ]

2.3 与 .map() 的配合方式

URL 搜索链一次只处理一个搜索词。但搜索词生成链输出的是一个列表，因此需要把 URL 搜索链应用到列表中的每个搜索词。这时可以使用 .map()：

1
all_search_result_urls_chain = search_result_urls_chain.map()

假设输入是三个搜索查询：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
[
    {
        "search_query": "杭州 必去景点 西湖 灵隐寺 西溪湿地",
        "user_question": "杭州有哪些值得一去的地方？"
    },
    {
        "search_query": "杭州 历史文化 京杭大运河 南宋御街 博物馆",
        "user_question": "杭州有哪些值得一去的地方？"
    },
    {
        "search_query": "杭州 美食 龙井茶 城市漫步 旅行攻略",
        "user_question": "杭州有哪些值得一去的地方？"
    }
]

经过 .map() 后，每个搜索词都会分别进入 URL 搜索链，输出结果是一个嵌套列表：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
[
    [
        {
            "result_url": "url_1",
            "search_query": "杭州 必去景点 西湖 灵隐寺 西溪湿地",
            "user_question": "杭州有哪些值得一去的地方？"
        },
        {
            "result_url": "url_2",
            "search_query": "杭州 必去景点 西湖 灵隐寺 西溪湿地",
            "user_question": "杭州有哪些值得一去的地方？"
        }
    ],
    [
        {
            "result_url": "url_3",
            "search_query": "杭州 历史文化 京杭大运河 南宋御街 博物馆",
            "user_question": "杭州有哪些值得一去的地方？"
        },
        {
            "result_url": "url_4",
            "search_query": "杭州 历史文化 京杭大运河 南宋御街 博物馆",
            "user_question": "杭州有哪些值得一去的地方？"
        }
    ],
    [
        {
            "result_url": "url_5",
            "search_query": "杭州 美食 龙井茶 城市漫步 旅行攻略",
            "user_question": "杭州有哪些值得一去的地方？"
        },
        {
            "result_url": "url_6",
            "search_query": "杭州 美食 龙井茶 城市漫步 旅行攻略",
            "user_question": "杭州有哪些值得一去的地方？"
        }
    ]
]

这个嵌套结构的含义是：

1
2
3
第一个列表元素：第一个搜索词得到的一组 URL
第二个列表元素：第二个搜索词得到的一组 URL
第三个列表元素：第三个搜索词得到的一组 URL

如果后续链希望逐个网页处理，可以继续展开这个嵌套列表。但在本案例中，更常见的做法是：一个搜索词对应一组 URL，然后把这一组 URL 交给网页总结链批量处理。

也就是说，URL 搜索链通常会和下一章的网页总结链组合成：一个搜索词 -> 多个 URL -> 多个网页摘要 -> 合并为该搜索词下的摘要。

这种设计的好处是结构更清晰。搜索词之间保持相对独立，每个搜索词负责一个研究角度。例如，“杭州 必去景点”负责景点概览，“杭州 历史文化”负责文化背景，“杭州 美食 龙井茶”负责体验和美食。最后再把多个角度的摘要合并成完整研究材料。

2.4 搜索结果的数量控制与去重

URL 搜索链中有个重要配置：

1
NUM_SEARCH_RESULTS = 3

它表示每个搜索词返回多少个网页。这个值不宜过大。假设搜索词数量是 3，每个搜索词返回 3 个网页，那么系统最多需要处理 9 个网页。每个网页后续都可能触发一次模型摘要调用，因此搜索结果数量会直接影响执行时间和模型调用成本。

如果配置变成：

1
2
NUM_SEARCH_QUERIES = 5
NUM_SEARCH_RESULTS = 5

那么系统最多需要处理 25 个网页。对于正式应用，这可能是合理的；但对于案例演示，数量过大会让流程变慢，也会增加调试难度。除了数量控制，还应考虑 URL 去重。不同搜索词可能返回同一个网页，如果不去重，后续摘要链会重复处理同一网页，造成内容重复和资源浪费。

可以在后续合并阶段加入简单去重逻辑：

1
2
3
4
5
6
7
8
9
10
11
def deduplicate_urls(items):
    seen = set()
    deduped = []

    for item in items:
        url = item["result_url"]
        if url not in seen:
            seen.add(url)
            deduped.append(item)

    return deduped

如果输入是嵌套列表，可以先展开再去重：

1
2
3
4
5
6
7
8
9
10
11
12
def flatten_and_deduplicate(nested_items):
    seen = set()
    deduped = []

    for group in nested_items:
        for item in group:
            url = item["result_url"]
            if url not in seen:
                seen.add(url)
                deduped.append(item)

    return deduped

是否在此阶段去重，取决于后续链路设计。如果希望保留“每个搜索词对应一组摘要”的结构，可以暂时不展开；如果希望把所有 URL 统一交给网页总结链，就需要先展开和去重。

本文案例保持链路简洁，URL 搜索链主要负责获取和包装 URL，复杂去重可以作为优化项放在后续工程改进中。

小结

本章构建了 URL 搜索链。它接收单个中文搜索查询，调用网页搜索函数获取 URL，并把每个 URL 包装成包含 result_url、search_query 和 user_question 的结构化对象。

这条链的核心结构是 搜索查询 -> RunnableLambda -> web_search() -> URL 列表 -> 结构化搜索结果。

URL 搜索链的关键价值，是把抽象的搜索词转换成可抓取的网页来源，并保留后续摘要所需的上下文。 它本身不调用模型，但它把外部信息入口接入了 LCEL 工作流。

下一章将继续构建网页总结链，根据 URL 抓取正文，截断过长文本，并调用模型生成与原始问题相关的网页摘要。

3. 构建网页总结链：抓取正文并提取有用信息

URL 搜索链解决了“资料从哪里来”的问题，但 URL 本身还不能直接用于报告生成。要把网页变成可用研究材料，还需要完成两步：抓取网页正文，以及 围绕原始问题生成摘要。

本章构建网页总结链。它接收一个搜索结果对象，访问其中的 result_url，抓取网页文本，再调用模型生成摘要。最终输出会保留来源 URL 和原始问题，方便后续合并成完整研究报告。

3.1 网页总结链的输入与输出

上一章的 URL 搜索链会输出如下结构：

1
2
3
4
5
{
    "result_url": "https://isabellalife.tw/articles-1028455",
    "search_query": "杭州 必去景点 西湖 灵隐寺 西溪湿地",
    "user_question": "杭州有哪些值得一去的地方？"
  }

网页总结链要基于这个输入完成三件事。

  1. 根据 result_url 抓取网页正文。
  2. 把网页正文、搜索词和原始问题一起交给模型，让模型只提取与问题相关的信息。
  3. 把摘要和来源 URL 合并成统一输出。

理想输出如下：

1
2
3
4
5
{
    "summary": "来源 URL: https://isabellalife.tw/articles-1028455\n摘要: 该xxx",
    "search_query": "杭州 必去景点 西湖 灵隐寺 西溪湿地",
    "user_question": "杭州有哪些值得一去的地方？"
}

这里保留 来源 URL 是因为后续最终报告虽然不一定逐条展示所有 URL，但摘要阶段必须保留来源信息，否则难以判断内容来自哪里，也不方便排查错误原因。

3.2 设计网页摘要提示词

网页抓取函数返回的文本通常比较杂乱，可能包含导航、页脚、广告、版权声明以及无关链接。摘要提示词不能只要求“总结以下网页”，而应该明确要求模型围绕原始问题筛选信息。

可以在 prompts.py 中定义如下模板：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from langchain_core.prompts import ChatPromptTemplate

SUMMARY_PROMPT_TEMPLATE = ChatPromptTemplate.from_template(
    """
    请根据下面的网页内容，围绕用户问题提取有价值的信息。

    用户问题：
    {user_question}

    搜索词：
    {search_query}

    网页内容：
    {search_result_text}

    输出要求：
    1. 只总结与用户问题相关的信息。
    2. 忽略导航栏、广告、版权声明、无关链接等噪声内容。
    3. 不要编造网页中没有的信息。
    4. 摘要应简洁、具体，保留地点、景点、活动、历史背景或实用建议等关键信息。
    5. 使用中文输出。
    """
)

这个提示词有三个关键点。

  • 它明确输入了 user_question。这能让模型知道最终目标是什么，而不是机械总结网页全文。
  • 它保留了 search_query。搜索词可以提供额外上下文，说明这个网页是围绕哪个角度被找到的。
  • 它要求忽略网页噪声，并且不编造网页中没有的信息。网页摘要链的目标不是创作，而是从已有文本中提取有用信息。

网页摘要阶段越克制，最终报告越可靠。 如果摘要阶段让LLM自由发挥，那后续生成的报告质量就很难保证。

3.3 用 LCEL 实现单网页总结链

单网页总结链可以分成三个环节：准备摘要输入、生成摘要、格式化输出。

首先，需要根据 URL 抓取网页正文，并控制文本长度：

1
2
3
4
5
6
7
8
9
10
11
12
from langchain_core.runnables import RunnableLambda

from web_scraping import web_scrape

RESULT_TEXT_MAX_CHARACTERS = 10000

prepare_summary_input = RunnableLambda(lambda x: {
    "search_result_text": web_scrape(x["result_url"])[:RESULT_TEXT_MAX_CHARACTERS],
    "result_url": x["result_url"],
    "search_query": x["search_query"],
    "user_question": x["user_question"],
})

这一步把原来的输入：

1
2
3
4
5
{
    "result_url": "...",
    "search_query": "...",
    "user_question": "..."
}

转换成摘要 Prompt 需要的输入：

1
2
3
4
5
6
{
    "search_result_text": "...",
    "result_url": "...",
    "search_query": "...",
    "user_question": "..."
}

其中，RESULT_TEXT_MAX_CHARACTERS 用来限制传入模型的网页文本长度。网页正文可能非常长，如果不截断，模型调用可能失败，或者摘要成本过高。这里采用字符截断是一种简单方案，后续可以替换为分块摘要或正文抽取。

接着，使用 RunnableParallel 同时完成摘要生成和上下文保留：

1
2
3
4
5
6
7
8
9
10
11
from langchain_core.runnables import RunnableParallel
from langchain_core.output_parsers import StrOutputParser

from llm_models import get_llm
from prompts import SUMMARY_PROMPT_TEMPLATE

summary_parallel = RunnableParallel({
    "text_summary": SUMMARY_PROMPT_TEMPLATE | get_llm() | StrOutputParser(),
    "result_url": lambda x: x["result_url"],
    "user_question": lambda x: x["user_question"],
})

这里的关键是，text_summary 分支会调用模型生成网页摘要，而 result_url 和 user_question 分支会原样保留。

如果只写：

1
SUMMARY_PROMPT_TEMPLATE | get_llm() | StrOutputParser()

就只能得到摘要字符串，URL 和原始问题会丢失。后续合并报告时，系统就无法知道摘要来自哪个网页，也无法继续保留原始任务目标。

最后，对输出进行格式化：

1
2
3
4
format_summary = RunnableLambda(lambda x: {
    "summary": f"来源 URL: {x['result_url']}\n摘要: {x['text_summary']}",
    "user_question": x["user_question"],
})

把三个环节组合起来，就得到完整的单网页总结链：

1
2
3
4
5
search_result_text_and_summary_chain = (
    prepare_summary_input
    | summary_parallel
    | format_summary
)

也可以写成一条完整 LCEL 链：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from langchain_core.runnables import RunnableLambda, RunnableParallel
from langchain_core.output_parsers import StrOutputParser

from llm_models import get_llm
from web_scraping import web_scrape
from prompts import SUMMARY_PROMPT_TEMPLATE

RESULT_TEXT_MAX_CHARACTERS = 10000

search_result_text_and_summary_chain = (
    RunnableLambda(lambda x: {
        "search_result_text": web_scrape(x["result_url"])[:RESULT_TEXT_MAX_CHARACTERS],
        "result_url": x["result_url"],
        "search_query": x["search_query"],
        "user_question": x["user_question"],
    })
    | RunnableParallel({
        "text_summary": SUMMARY_PROMPT_TEMPLATE | get_llm() | StrOutputParser(),
        "result_url": lambda x: x["result_url"],
        "user_question": lambda x: x["user_question"],
    })
    | RunnableLambda(lambda x: {
        "summary": f"来源 URL: {x['result_url']}\n摘要: {x['text_summary']}",
        "user_question": x["user_question"],
    })
)

这条链体现了 LCEL 的一个典型模式：先用 RunnableLambda 准备数据，再用 RunnableParallel 同时生成内容和保留字段，最后再次用 RunnableLambda 整理输出。

3.4 运行单网页总结链

可以用一个搜索结果对象测试这条链：

1
2
3
4
5
6
7
8
9
input_data = {
    "result_url": "https://example.com/astorga-travel-guide",
    "search_query": "杭州 必去景点 西湖 灵隐寺 西溪湿地",
    "user_question": "杭州有哪些值得一去的地方？"
}

result = search_result_text_and_summary_chain.invoke(input_data)

print(result)

输出结构类似：

1
2
3
4
{
	'summary': '来源 URL: https://www.sohu.com/a/953786156_120932760\n摘要: xxxx', 
	'user_question': '杭州有哪些值得一去的地方？'
}

实际运行时，如果网页无法访问，web_scrape() 可能返回错误文本，例如：

1
无法获取网页内容，状态码：403

这种错误文本也会进入摘要 Prompt。为了避免模型对错误信息进行无意义总结，正式项目中可以在抓取阶段加入判断。如果抓取失败，就跳过该 URL，或将其标记为失败结果。

1
2
3
4
5
6
7
8
9
def prepare_summary_data(x):
    text = web_scrape(x["result_url"])

    return {
        "search_result_text": text[:RESULT_TEXT_MAX_CHARACTERS],
        "result_url": x["result_url"],
        "search_query": x["search_query"],
        "user_question": x["user_question"],
    }

后续再根据 text 是否以“无法获取网页内容”开头进行过滤。不过在当前案例中，为了保持链路清晰，先采用最小实现。

3.5 网页摘要链的关键问题

网页总结链是整个系统中最容易影响最终质量的环节。这里需要重点关注三个问题。

第一，网页正文存在噪声。BeautifulSoup.get_text() 会提取页面中的大量文本，但并不保证这些文本都是正文。导航栏、菜单、页脚、推荐链接和版权信息都可能被包含在内。因此，摘要 Prompt 必须要求模型忽略无关内容，只提取与用户问题相关的信息。

第二，网页内容可能过长。简单截断可以防止模型输入过长，但也可能截掉有价值内容。比如网页前半部分是导航和广告，真正正文在后面，那么直接截取前面字符可能效果一般。后续优化可以使用正文提取工具，或者先按段落分块，再筛选与问题相关的片段。

第三，摘要不能脱离来源。如果只输出摘要，不保留 URL，后续报告看起来完整，但信息不可追溯。网页研究类应用尤其需要保留来源字段。即便最终文章不展示全部 URL，中间数据也应保留来源，方便调试和核查。

因此，这条链的设计重点不是让模型“多写一点”，而是让模型在受约束的上下文中提取有用事实，并让系统保留必要元数据。

小结

本章构建了网页总结链。它接收包含 result_url、search_query 和 user_question 的搜索结果对象，抓取网页正文，截断过长文本，然后调用模型生成与原始问题相关的中文摘要。

这条链的核心结构是，搜索结果对象 -> RunnableLambda 抓取网页正文并整理字段 -> RunnableParallel 生成摘要并保留 URL、用户问题 -> RunnableLambda 格式化摘要 -> 带来源的网页摘要。

网页总结链是把网页 URL 转换成可用于报告生成的研究材料。 它既利用了模型的语义提取能力，又保留了来源 URL 和原始问题，使后续摘要合并与报告生成具备可靠上下文。

下一章将继续构建批量处理逻辑：把多个 URL 交给同一条网页总结链，并将多个网页摘要合并为一个搜索角度下的研究摘要。

4. 使用.map()批量总结多个网页

上一章已经完成单网页总结链，输入一个 URL 搜索结果，抓取网页正文，生成带来源摘要。但在完整流程中，系统不会只处理一个网页。一个搜索词通常会返回多个 URL，每个 URL 都需要经过同样的抓取和总结步骤。

本章的目标是把“单网页总结能力”扩展为“多网页总结能力”。这一步主要依赖 LCEL 的 .map()。它可以把同一条链应用到列表中的每个元素，使多个网页按照统一逻辑被处理。

4.1 从单个 URL 到多个 URL

URL 搜索链针对一个搜索词会返回一组网页结果。例如，搜索词是：

1
"杭州 必去景点 西湖 灵隐寺 西溪湿地"

返回结果可能整理成如下结构：

1
2
3
4
5
6
7
8
9
10
11
12
[  
    {  
	    "result_url": "https://www.sohu.com/a/953786156_120932760",  
		"search_query": "杭州 必去景点 西湖 灵隐寺 西溪湿地",  
        "user_question": "杭州有哪些值得一去的地方？"  
     },  
     {  
	    "result_url": "https://news.qq.com/rain/a/20260305A04ETV00",  
        "search_query": "杭州 必去景点 西湖 灵隐寺 西溪湿地",  
        "user_question": "杭州有哪些值得一去的地方？"  
     },...  
]

这组数据中的每个元素，都可以作为单网页总结链的输入。因此，处理逻辑非常明确：

1
2
3
URL 1 → 抓取正文 → 生成摘要
URL 2 → 抓取正文 → 生成摘要
URL 3 → 抓取正文 → 生成摘要

如果用普通 Python 写法，大致是：

1
2
3
4
5
summaries = []

for item in search_results:
    summary = search_result_text_and_summary_chain.invoke(item)
    summaries.append(summary)

这种写法可以工作，但在 LCEL 工作流中，更自然的方式是使用 .map()：

1
search_result_text_and_summary_chain.map()

.map() 表示：列表中的每个元素都会进入同一条链，最终返回一个新的结果列表。

4.2 构建搜索与总结链

现在可以把 URL 搜索链和网页总结链连接起来。URL 搜索链负责把搜索词转换为多个 URL，网页总结链负责处理每个 URL。组合后得到一条新的链：搜索与总结链。

1
2
3
4
5
6
7
8
9
10
from langchain_core.runnables import RunnableLambda

search_and_summarization_chain = (
    search_result_urls_chain
    | search_result_text_and_summary_chain.map()
    | RunnableLambda(lambda x: {
        "summary": "\n".join([item["summary"] for item in x]),
        "user_question": x[0]["user_question"] if len(x) > 0 else "",
    })
)

这条链可以拆成三步理解，搜索词 -> 获取多个 URL -> 对每个 URL 执行网页总结链 -> 合并多个网页摘要。

第一步，search_result_urls_chain 接收一个搜索查询对象，例如：

1
2
3
4
{
	"search_query": "杭州 历史文化 京杭大运河 南宋御街 博物馆",  
	"user_question": "杭州有哪些值得一去的地方？"
}

它会输出多个搜索结果：

1
2
3
4
5
[
    {"result_url": "url_1", "search_query": "...", "user_question": "..."},
    {"result_url": "url_2", "search_query": "...", "user_question": "..."},
    {"result_url": "url_3", "search_query": "...", "user_question": "..."}
]

第二步，search_result_text_and_summary_chain.map() 对这个列表中的每个元素执行单网页总结链，得到多个摘要对象：

1
2
3
4
5
6
7
8
9
10
[
    {
        "summary': '来源 URL: https://www.sohu.com/a/953786156_120932760\n摘要: xxx'",
        "user_question": "..."
    },
    {
        "summary': '来源 URL: https://news.qq.com/rain/a/20260305A04ETV00\n摘要: xxx'",
        "user_question": "..."
    },...
]

第三步，最后的 RunnableLambda 负责把多个摘要合并成一个字符串：

1
2
3
4
RunnableLambda(lambda x: {
    "summary": "\n".join([item["summary"] for item in x]),
    "user_question": x[0]["user_question"] if len(x) > 0 else "",
})

合并后的输出大致如下：

1
2
3
4
{
    "summary": "来源 URL: url_1\n摘要: ...\n来源 URL: url_2\n摘要: ...\n来源 URL: url_3\n摘要: ...",
    "user_question": "..."
}

这样，一个搜索词对应的一组网页，就被转换成了一段可用于后续报告生成的研究摘要。

4.3 .map()的实际价值

.map() 最直接的作用是减少手写循环，但它的价值不只在于代码更短。更重要的是，它让“单项处理链”和“批量处理链”之间形成清晰关系。

在本案例中，单网页总结链只关心一个 URL：

1
一个 URL → 一个摘要

通过 .map()，它自然扩展为：

1
多个 URL → 多个摘要

这符合 LCEL 的组合思想：先把最小处理单元定义清楚，再把它扩展到列表数据。

如果后续需要修改网页总结逻辑，例如更换摘要 Prompt、增加失败过滤、保留网页标题，只需要修改单网页总结链。批量处理部分不需要改变，因为 .map() 仍然负责把这条链应用到每个元素。

这比在多个地方手写循环更容易维护。

4.4 运行搜索与总结链

可以使用一个中文搜索查询测试这条链：

1
2
3
4
5
6
7
8
input_data = {  
    "search_query": "杭州 必去景点 西湖 灵隐寺 西溪湿地",  
    "user_question": "杭州有哪些值得一去的地方？"  
}

result = search_and_summarization_chain.invoke(input_data)

print(result)

输出结构类似：

1
2
{'summary': '来源 URL: https://www.klook.com/zh-CN/activity/155336-private-guide-1-day-tour-of-hangzhou-city-by-car/\n摘要: xxx\n来源 URL: https://www.getyourguide.com/zh-cn/lingyin-temple-l150150/\n摘要: xxx。\n来源 URL: https://www.facebook.com/cutitrip/posts/杭州自由行必去九大景点️来杭州游玩这九大热门景点不容错过西湖江南四大名湖之一西湖十景美不胜收雷峰塔推荐傍晚时分前往欣赏雷峰夕照的绝美景色灵隐寺千年古刹佛教圣地氛/1436398781860225/\n摘要: xxx。', 'user_question': '杭州有哪些值得一去的地方？'}

从结果可以看出，搜索与总结链已经把“搜索词”转换成了“多来源摘要”。这正是最终报告生成前需要的中间材料。

需要注意的是，真实运行时可能出现部分网页抓取失败。例如某些网站返回 403，或者网页需要 JavaScript 渲染。为了保持流程不中断，基础抓取函数通常会返回错误信息。正式项目中可以在合并摘要前过滤失败内容，避免无效结果进入最终报告。

例如：

1
2
3
4
5
6
7
8
9
10
def merge_valid_summaries(items):
    valid_items = [
        item for item in items
        if "无法获取网页内容" not in item["summary"]
    ]

    return {
        "summary": "\n".join([item["summary"] for item in valid_items]),
        "user_question": valid_items[0]["user_question"] if len(valid_items) > 0 else "",
    }

然后替换最后的合并逻辑：

1
2
3
4
5
search_and_summarization_chain = (
    search_result_urls_chain
    | search_result_text_and_summary_chain.map()
    | RunnableLambda(merge_valid_summaries)
)

这个优化不是必需步骤，但它能提高最终报告质量。

小结

本章将单网页总结链扩展为批量网页总结流程。通过 .map()，同一条网页总结链可以应用到多个 URL，系统无需手写复杂循环，也能保持 LCEL 数据流的统一表达。

本章构建的搜索与总结链结构，单个搜索查询 -> URL 搜索链 -> 多个搜索结果 URL -> 网页总结链 .map() -> 多个网页摘要 -> 摘要合并。

.map() 的关键价值，是把“单个对象处理能力”扩展成“列表批量处理能力”。 在自动化网页研究总结器中，它让多个网页可以进入同一条处理链，并最终合并为一个搜索角度下的研究摘要。

下一章将把前面的角色选择链、搜索词生成链、搜索与总结链继续组合起来，形成完整的研究报告生成流程。

5. 组装最终研究报告链

前面几章已经分别完成了角色选择、搜索词生成、URL 搜索、网页摘要和批量总结。到这里，自动化网页研究总结器的关键部件已经具备，但它们还只是分散的小链。要形成完整应用，还需要把这些小链按照数据流方向组合起来，并在最后生成一篇结构化 Markdown 报告。

本章的目标是构建 最终研究报告链。这条链会接收一个问题，依次完成任务理解、搜索规划、网页总结和报告生成。外部调用时，只需要执行一次 invoke()，内部就会完成完整流程。

5.1 最终链路的整体结构

先回顾已经构建好的几条链：

1
2
3
4
5
6
7
8
assistant_instructions_chain
角色选择链：根据用户问题生成助手设定

web_searches_chain
搜索词生成链：根据用户问题和助手设定生成多个搜索词

search_and_summarization_chain
搜索与总结链：根据单个搜索词获取多个 URL，并总结多个网页

现在要做的是把它们连接起来，用户问题 -> 角色选择链 -> 搜索词生成链 -> 搜索与总结链 .map() -> 合并所有搜索角度下的摘要 -> 报告生成 Prompt -> LLM -> Markdown 报告。

这里有一个重要变化：search_and_summarization_chain 一次只处理一个搜索词，而 web_searches_chain 会输出多个搜索词。因此需要再次使用 .map()，让每个搜索词都进入同一条搜索与总结链。

整体思路是，多个搜索词 → 多组网页摘要 → 合并为研究材料 → 生成最终报告。

5.2 设计最终报告提示词

最终报告生成阶段不再直接处理网页原文，而是处理前面合并得到的摘要集合。这样做有两个好处：一是减少输入噪声，二是让模型专注于组织结构和表达，而不是从杂乱网页中重新筛选信息。

可以在 prompts.py 中定义报告生成模板：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from langchain_core.prompts import ChatPromptTemplate

RESEARCH_REPORT_PROMPT_TEMPLATE = ChatPromptTemplate.from_template(
    """
    请根据下面的研究摘要，围绕用户问题生成一篇结构化 Markdown 报告。

    用户问题：
    {user_question}

    研究摘要：
    {research_summary}

    写作要求：
    1. 使用中文输出。
    2. 报告应结构清晰，包含标题、概述、分主题内容和总结。
    3. 内容必须基于研究摘要，不要编造摘要中没有的信息。
    4. 如果不同来源的信息有重复，请合并表达，避免重复堆砌。
    5. 如果摘要中包含来源 URL，可以在相关内容后保留来源说明。
    6. 语言自然、书面化，适合作为正式研究报告阅读。
    """
)

这个提示词的重点是限制模型的发挥范围。最终报告阶段确实需要模型重新组织语言，但不能脱离摘要材料自由创作。因此，提示词中明确要求：

1
内容必须基于研究摘要，不要编造摘要中没有的信息。

同时，报告要解决摘要集合的两个常见问题：信息重复和结构松散。多个网页可能都提到同一个景点，如果最终报告逐条复述摘要，会显得冗余。报告生成链需要把重复信息合并，并按照主题重新组织。

这里的标题结构并不需要硬编码死。只要提示词要求“结构清晰”，模型可以根据摘要内容选择合适章节。但如果需要更稳定的输出，也可以在提示词中固定章节名称。

5.3 合并多个搜索角度的摘要

在生成报告之前，需要先把多个搜索词对应的摘要合并成一个 research_summary 字段。

前一章的 search_and_summarization_chain 针对一个搜索词输出：

1
2
3
4
{
    "summary": "来源 URL: url_1\n摘要: ...\n来源 URL: url_2\n摘要: ...",
    "user_question": "杭州有哪些值得一去的地方？"
}

而搜索词生成链会输出多个搜索词，因此经过 .map() 后会得到：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
[
    {
        "summary": "搜索词 1 对应的多个网页摘要...",
        "user_question": "杭州有哪些值得一去的地方？"
    },
    {
        "summary": "搜索词 2 对应的多个网页摘要...",
        "user_question": "杭州有哪些值得一去的地方？"
    },
    {
        "summary": "搜索词 3 对应的多个网页摘要...",
        "user_question": "杭州有哪些值得一去的地方？"
    }
]

最终报告 Prompt 需要的是：

1
2
3
4
{
    "research_summary": "...所有摘要合并后的文本...",
    "user_question": "杭州有哪些值得一去的地方？"
}

可以用 RunnableLambda 完成合并：

1
2
3
4
5
6
from langchain_core.runnables import RunnableLambda

merge_research_summaries = RunnableLambda(lambda x: {
    "research_summary": "\n\n".join([item["summary"] for item in x]),
    "user_question": x[0]["user_question"] if len(x) > 0 else "",
})

这里的 "\n\n".join(...) 用两个换行分隔不同搜索角度的摘要，方便模型识别不同信息块。

如果担心有空摘要或失败摘要，也可以写成更稳妥的版本：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def merge_research_summaries_fn(items):
    valid_items = [
        item for item in items
        if item.get("summary")
    ]

    return {
        "research_summary": "\n\n".join(
            [item["summary"] for item in valid_items]
        ),
        "user_question": valid_items[0]["user_question"] if len(valid_items) > 0 else "",
    }

merge_research_summaries = RunnableLambda(merge_research_summaries_fn)

这个函数会过滤没有摘要的结果，避免空内容进入最终 Prompt。

5.4 构建完整研究报告链

现在可以把所有链组合起来：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda

from llm_models import get_llm
from prompts import RESEARCH_REPORT_PROMPT_TEMPLATE
from chains.assistant_chain import assistant_instructions_chain
from chains.search_query_chain import web_searches_chain
from chains.summarize_url_chain import search_and_summarization_chain

merge_research_summaries = RunnableLambda(lambda x: {
    "research_summary": "\n\n".join([item["summary"] for item in x]),
    "user_question": x[0]["user_question"] if len(x) > 0 else "",
})

web_research_chain = (
    assistant_instructions_chain
    | web_searches_chain
    | search_and_summarization_chain.map()
    | merge_research_summaries
    | RESEARCH_REPORT_PROMPT_TEMPLATE
    | get_llm()
    | StrOutputParser()
)

这条链从外部看只接收一个输入，但内部会经过多个阶段。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
assistant_instructions_chain
接收问题，输出助手设定和原始问题

web_searches_chain
接收助手设定和原始问题，输出多个搜索词

search_and_summarization_chain.map()
对每个搜索词执行搜索与网页总结

merge_research_summaries
合并多个搜索角度下的摘要

RESEARCH_REPORT_PROMPT_TEMPLATE
构造最终报告 Prompt

get_llm()
生成 Markdown 报告

StrOutputParser()
输出字符串报告

这里最值得注意的是：

1
search_and_summarization_chain.map()

它表示搜索词生成链输出的每个搜索词，都会进入同一条搜索与总结链。这样，系统就能围绕多个角度收集网页信息，而不是只依赖一个搜索结果。

5.5 运行完整流程

最终使用时，可以在 main.py 中写入：

1
2
3
4
5
6
7
8
9
from chains.research_chain import web_research_chain

question = {
    "user_question": "杭州有哪些值得一去的地方？"
}

report = web_research_chain.invoke(question)

print(report)

如果前面的角色选择链被设计为直接接收字符串，也可以写成：

1
2
3
4
5
question = "杭州有哪些值得一去的地方？"

report = web_research_chain.invoke(question)

print(report)

两种方式都可以，但更推荐统一使用字典输入：

1
2
3
{
    "user_question": "..."
}

对于正式项目，统一字典输入会比直接传字符串更灵活，完整流程运行后，输出大致会是一篇 Markdown 报告：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 杭州深度旅行研究报告：值得探索的地方与规划建议

## 1. 报告概述
...

## 2. 核心景区与游览主题
...

## 3. 整合行程建议
...

## 4. 实用参考信息
...

## 5. 总结
...

实际内容取决于搜索结果和网页抓取结果。由于搜索引擎返回结果具有不确定性，不同时间运行可能得到不同网页，也可能生成略有差异的报告。这是网页研究类应用的正常现象。

5.6 完整链的设计要点

最终研究报告链把多个小链组合成完整流程，但它并不是简单地把代码串起来。这里有几个设计要点需要特别关注。

第一，每条链的输入输出必须对齐。角色选择链要输出 user_question 和 assistant_instructions，搜索词生成链要输出包含 search_query 的列表，搜索与总结链要输出 summary 和 user_question。只要其中一个字段丢失，后续链就会报错。

第二，每个阶段只承担自己的职责。角色选择链不负责搜索，搜索词生成链不负责抓取网页，网页总结链不负责写最终报告。职责边界清楚，后续替换和调试才方便。

第三，最终报告应基于摘要，而不是直接基于网页原文。网页原文噪声大、长度长，直接交给最终报告 Prompt 会增加混乱。先摘要再报告，是更稳定的方式。

第四，中间结果应保留来源信息。来源 URL 至少应保留到摘要阶段。这样即使最终报告没有列出完整引用，也能在调试时追踪内容来源。

第五，链式组合不等于无条件增加步骤。每个新增链都应该解决明确问题。如果只是把简单逻辑拆得过碎，反而会增加理解成本。本案例中的链拆分，基本对应了任务自然阶段：理解、搜索、抓取、摘要、报告。

小结

本章完成了最终研究报告链的组装。通过 LCEL，可以把角色选择链、搜索词生成链、搜索与总结链、摘要合并逻辑和报告生成 Prompt 连接成完整流程。

最终链的核心结构如下，问题 -> 角色选择链 -> 搜索词生成链 -> 搜索与总结链 .map() -> 合并研究摘要 -> 报告生成 Prompt -> LLM -> Markdown 报告

这条链的关键价值，是把多个分散能力封装成一个可直接调用的研究工作流。 外部只需要提供中文问题，内部会自动完成搜索规划、网页资料收集、摘要提取和报告生成。

下一章将进一步总结这个案例中 LCEL 的核心价值，并说明这种链式结构相比普通函数式写法在可组合性、可替换性和可扩展性上的优势。

6. 总结：用 LCEL 搭建可维护的 AI 工作流

前面围绕自动化网页研究总结器，完整实现了从用户问题到 Markdown 报告生成的流程。这个案例表面上是在构建一个网页研究工具，实际上展示的是一种更通用的大模型应用组织方式：把模型调用、普通函数、数据转换、批量处理和结果生成统一编排成稳定的数据流。

LCEL 在这个过程中承担的是工作流编排层的角色。它的价值不是让代码看起来更短，而是让复杂 AI 应用从单次 Prompt 调用，升级为可以拆解、调试、替换和扩展的链式工作流。

6.1 从单次调用到链式工作流

最简单的大模型应用通常是，输入 → Prompt → LLM → 输出。这种结构适合单步任务，但很难支撑网页研究总结器这样的复杂流程。因为系统不仅要生成文本，还要完成搜索词生成、URL 搜索、网页抓取、网页摘要、摘要合并和最终报告生成。

在本案例中，完整流程被拆成多条小链：

1
2
3
4
5
6
assistant_instructions_chain
web_searches_chain
search_result_urls_chain
search_result_text_and_summary_chain
search_and_summarization_chain
web_research_chain

每条链都有独立职责。角色选择链负责生成助手设定，搜索词生成链负责拆解查询，URL 搜索链负责获取网页来源，网页总结链负责把单个网页转成摘要，最终报告链负责整合结果并输出 Markdown 报告。

这种拆分方式让系统从“一个大函数”变成了“多个可组合模块”。调试时可以单独运行某一条链，正式运行时只需要调用最终链：

1
2
3
report = web_research_chain.invoke({
    "user_question": "..."
})

外部看到的是一次调用，内部完成的是完整研究流程。这正是 LCEL 链式应用的核心价值。

6.2 模型、工具与编排的分工

这个案例还体现了一个重要原则：大模型应用不是所有事情都交给模型完成，而是模型、普通函数和编排层各司其职。

模型适合处理语义任务，例如：

1
2
3
4
判断任务类型
生成搜索词
总结网页内容
生成结构化报告

普通函数适合处理确定性任务，例如：

1
2
3
4
5
6
调用搜索工具
请求网页 URL
解析 HTML
截断文本
合并摘要
过滤失败结果

LCEL 则负责把这些能力连接起来：

1
Prompt、Model、Parser、RunnableLambda、RunnableParallel、.map()

例如，网页总结链中既有普通函数：

1
web_scrape(x["result_url"])

也有模型调用：

1
SUMMARY_PROMPT_TEMPLATE | get_llm() | StrOutputParser()

这体现了一个关键原则：模型只处理需要语义能力的部分，确定性逻辑交给普通函数完成。这样比“所有事情都问模型”更稳健，也更接近真实工程中的大模型应用形态。

6.3 显式的数据流是可维护性的基础

LCEL 应用中常见的问题并不一定来自模型，而是来自数据结构不匹配。比如 Prompt 需要 user_question，但上一条链输出的是 question；某个节点需要 result_url，但前一步只返回了字符串；合并函数预期输入是列表，但实际传入的是字典。

因此，LCEL 工作流的关键不是链式语法本身，而是清楚设计每个节点的输入和输出。

在本案例中，数据从最初的问题逐步变成最终报告，user_question → assistant_instructions → search_query → result_url → search_result_text → summary → research_summary → markdown_report。

RunnableLambda 的作用就是把这些数据转换显式写出来。例如搜索词生成链会明确传入：

1
2
3
assistant_instructions
num_search_queries
user_question

最终摘要合并阶段则把多个摘要对象转换成最终报告需要的字段：

1
2
research_summary
user_question

这种显式转换让每个阶段的数据结构都能被看见，也让错误更容易定位。稳定的数据流，才是 LCEL 应用可维护性的核心。

6.4 批量处理与上下文传递

网页研究总结器天然包含批量任务。一个问题会生成多个搜索词，一个搜索词会返回多个 URL，每个 URL 都要进入同一条网页总结链。LCEL 的 .map() 让这种批量处理变得自然：

1
2
一个 URL → 一个摘要
多个 URL → 多个摘要

同样，搜索与总结链可以从：

1
一个搜索词 → 多个网页摘要

扩展为：

1
多个搜索词 → 多组网页摘要

.map() 的意义不只是省掉循环，而是让“单项处理能力”和“批量处理能力”保持同一种链式表达。

同时，多步骤应用还必须持续传递上下文。本案例中，几个字段会贯穿多个阶段：

字段 作用
user_question 保留原始研究目标
search_query 标记网页来自哪个搜索角度
result_url 保留网页来源
summary 存储网页摘要或合并摘要

这说明在多步骤 AI 应用中，生成新内容和保留上下文同样重要。如果摘要阶段丢失 URL，最终报告就难以追溯来源；如果丢失原始问题，摘要就可能偏离研究目标。

6.5 可替换与可扩展的工程结构

LCEL 链式结构的另一个价值是组件可替换。搜索工具、网页抓取方式、模型、Prompt 和输出格式都可以独立调整。

例如，当前搜索工具使用 DuckDuckGo，如果后续要换成其他搜索 API，只需要修改搜索函数；网页抓取当前使用 requests + BeautifulSoup，如果后续要换成专业正文提取工具，也只需要替换抓取模块；模型通过 get_llm() 初始化，后续更换模型时无需逐条修改链。

这种可替换性来自清晰的职责拆分：

1
2
3
4
5
6
7
llm_models.py：统一初始化模型
web_searching.py：封装网页搜索
web_scraping.py：封装网页抓取
utilities.py：放置工具函数
prompts.py：集中管理提示词
chains/：存放各条 LCEL 链
main.py：运行入口

后续扩展也可以按节点接入，例如：

扩展能力 接入位置
URL 去重 URL 搜索链之后
网页正文清洗 网页抓取之后
摘要质量过滤 网页总结链之后
来源可信度排序 报告生成之前
报告格式选择 最终 Prompt 阶段
流式输出 最终报告生成阶段

因此，可维护性不是来自代码少，而是来自边界清楚。每个模块职责稳定，系统才能持续演进。

6.6 当前实现的局限与优化方向

虽然当前版本已经能完整运行，但它仍然是一个基础版本。网页研究类应用的不确定性较高，主要问题包括：

1
2
3
4
5
6
搜索结果可能不稳定
网页抓取可能失败
网页正文可能包含噪声
模型摘要可能遗漏或偏移
最终报告可能重复
来源管理可能不够清晰

这些问题不是 LCEL 单独能解决的。LCEL 解决的是流程组织问题，而不是搜索质量、抓取质量和模型可靠性的全部问题。

后续可以从五个方向优化：

方向 优化方式
搜索阶段 查询多样化、URL 去重、低质量结果过滤
抓取阶段 失败处理、正文清洗、必要时引入浏览器渲染或抓取 API
摘要阶段 分块摘要、结构化摘要、强化“只基于网页内容”的约束
报告阶段 合并重复信息、保留来源、增加参考来源部分
系统层面 缓存、并行处理、日志、异常重试、质量评估

更现实的演进方式是先保证链路简单可运行，再根据实际失败模式逐步增强。LCEL 的优势在于，这些优化都可以作为新的 Runnable 节点接入原有链路，而不需要推翻整体结构。

6.7 从案例走向通用模式

虽然本文案例是自动化网页研究总结器，但背后的模式可以迁移到很多 AI 应用中。

文档分析系统可以采用：

1
用户问题 → 检索文档片段 → 总结片段 → 合并证据 → 生成答案

技术调研系统可以采用：

1
技术问题 → 生成检索查询 → 抓取资料 → 摘要重点 → 对比方案 → 生成建议

舆情分析系统可以采用：

1
分析主题 → 搜索相关内容 → 提取观点与情绪 → 聚类归纳 → 生成分析报告

这些应用看似不同，但结构上都包含共同环节：

1
任务输入 → 信息获取 → 信息提取 → 信息合并 → 结果生成

因此，本案例的价值不只在于完成一个网页研究工具，而是提供了一种可复用的大模型工程方法：先拆任务，再定数据结构，再用 LCEL 组合链路。

总结

自动化网页研究总结器展示了一个完整的大模型应用流程：输入问题，生成搜索词，搜索网页，抓取正文，总结内容，合并信息，最终输出 Markdown 报告。这个流程之所以能够保持清晰，是因为 LCEL 将不同能力组织成了可组合的数据流。

全文可以归纳为三点。

第一，LCEL 让复杂 AI 应用从单次调用变成可维护的链式工作流。它通过 Prompt、Model、Parser、RunnableLambda、RunnableParallel 和 .map() 等组件，把多个步骤连接成完整流程。

第二，大模型应用需要明确区分模型能力和程序能力。语义理解、摘要和生成交给模型，搜索、抓取、解析和合并交给普通函数，LCEL 负责二者之间的编排。

第三，工程质量取决于每个节点的输入输出设计。搜索词是否合理、URL 是否去重、网页正文是否干净、摘要是否保留来源、报告是否合并重复信息，都会影响最终效果。

因此，LCEL 的核心价值不是替代搜索、抓取、清洗和质量评估，而是提供清晰的应用编排框架。只要每个节点职责清楚、输入输出稳定、上下文传递完整，复杂大模型应用就能具备可调试、可扩展和可复用的工程基础。

7.备注

完整代码：https://github.com/keychankc/AI_agent_code/tree/main/web-research-lcel