企业官网建设流程全解析

说实话，我一开始对 LangChain LCEL 里那些链式和工具调用的写法是持保留态度的。上个月接了个需求，要在问答系统里加一个“查时间再算数”的小能力，需求本身不大，但把原来固定流程的链改造成动态决定调哪个工具、甚至连续调多个工具的应用，折腾了我整整几天。这篇就当是那几天的记录：从确定性链一步步演进到动态 Agent，过程中哪些代码值得抄、哪些坑值得避开，我都摊开讲。

如果你正准备把几个工具接进 LangChain 应用，或者已经在用 LCEL 但总觉得哪里别扭，这篇应该能帮你节省不少试错时间。

1. 传统 Chain 写法让人头疼的地方

先说为什么我会从旧式 Chain 写法里跳出来。LCEL 出来之前，LangChain 的主流入门路径是LLMChain、SimpleSequentialChain这类类继承体系。当时写一个“天气问答助手”，代码大概长这样：

from langchain.chains import LLMChain from langchain.prompts import PromptTemplate from langchain_ollama import ChatOllama llm = ChatOllama(model="qwen2.5:7b-instruct", temperature=0) weather_prompt = PromptTemplate.from_template( "你是天气助手。请回答用户的问题：{question}" ) chain = LLMChain(llm=llm, prompt=weather_prompt) result = chain.invoke("云州今天适合跑步吗？")

你发现问题了吗？这个链里模型只会“回答”，它根本没真正调用天气工具。模型哪怕在训练语料里背过一点天气规律，也是过时的、幻觉的、和真实天气无关的。为了让链条真正拿到数据，我得做一堆脏活：把工具执行结果拼进提示词，再重新组织一轮 prompt，逻辑越来越乱。

旧式 Chain 让我头疼的其实是这么几个点：

• 数据流不统一。每个 Chain 内部各自处理 prompt、LLM 输出、解析结果，Chain 和 Chain 之间只是靠字符串/字典硬衔接。一旦其中一环输出结构变了，后面所有环节都得跟着改。
• 编排能力弱。想并行查天气、查时间、查空气质量，旧式 Chain 得手动写多线程或者用SequentialChain串行跑，硬生生把本来可以并发的任务变成排队。
• 模型没有工具选择权。所有逻辑都是写死在代码里的。模型只负责“生成一句话”，它无法根据用户问题决定“我先查一下天气再回答”。
• 组合和复用差。一个 Chain 往往只能服务一个场景，换个业务又要重新继承、重写。

后来 LangChain 推出 LCEL，设计思路从“类继承”转向“组件组合”，我才算真正找到顺手的工具。本文后面所有示例都以 LCEL 为主，本地推理用ChatOllama，模型选一个支持工具调用的开源模型即可。

2. LCEL 的立身之本：Runnable 接口与管道操作

2.1 Runnable 不是抽象思想，是具体协议

LCEL 最核心的抽象是 Runnable。它不是一个只能在文档里看的理论概念，而是一套实打实的接口约定。LangChain 里几乎所有可执行组件都实现了 Runnable：prompt 模板、大模型、输出解析器、检索器、工具，甚至连 Agent 本身都是 Runnable。

Runnable 至少提供四个方法：

• invoke：同步执行，返回结果
• ainvoke：异步执行
• stream：流式返回
• batch：批量执行

正因为大家都实现了同一套接口，才可以用管道符|把任意 Runnable 串联起来。哪怕你写的是自定义函数，只要包成RunnableLambda，它也能无缝接入这条管道。

2.2 管道操作符背后的语法糖

LCEL 最直观的入门写法是这么一行：

chain = prompt | llm | StrOutputParser()

它看着像 Unix 管道，实际也确实借鉴了管道思想。prompt.invoke()的输出会作为llm.invoke()的输入，llm.invoke()输出的 AIMessage 再传给StrOutputParser转成纯字符串。

不过它比 Unix 管道更灵活：上一步的输出不仅可以是字符串，还可以是字典、消息对象、任意自定义结构。|真正干的事是构建一个RunnableSequence，内部按顺序把上一步的输出传给下一步的输入。整个过程是懒加载的，只有最终调用时才真正执行。

我喜欢把 LCEL 管道理解成一条流水线：每个工位上站着一个 Runnable，物料（数据）从第一个工位进入，每个工位干完活自动丢给下一个，最后出来的就是成品。

2.3 并行与分支

有了 Runnable 接口，并行也变成了一件很自然的事。LCEL 里的RunnableParallel可以把多个 Runnable 并行执行，再把结果合并成一个字典。比如同时查天气和时间：

from langchain_core.runnables import RunnableParallel parallel_chain = RunnableParallel( weather=get_city_weather_chain, time=get_current_time_chain, ) result = parallel_chain.invoke({"city": "云州"})

这比串行快一倍，尤其在工具本身是 I/O 密集型请求时特别明显。它还支持字典字面量简化写法：

chain = { "weather": get_city_weather_chain, "time": get_current_time_chain, } | RunnableLambda(lambda x: f"天气：{x['weather']}，时间：{x['time']}")

只要RunnableParallel产出的是字典，后面接任何 Runnable 都能消费。这种设计让分支、汇合、并行都变得非常直观。

3. 工具调用在 LCEL 里的接入方式

3.1 用 @tool 定义工具与结构化 schema

工具调用的第一步，是让模型知道“有什么工具可用、每个工具的参数长什么样”。LCEL 里定义一个工具非常简单：

from datetime import datetime from langchain_core.tools import tool @tool def get_current_time() -> str: """返回当前系统时间，格式为 YYYY-MM-DD HH:MM:SS。""" return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

注意，真正给模型看的不只是函数返回值，而是函数签名、类型注解和 docstring 一起生成的 JSON Schema。比如上面的工具会自动生成类似这样的结构：

{ "name": "get_current_time", "description": "返回当前系统时间，格式为 YYYY-MM-DD HH:MM:SS。", "parameters": { "type": "object", "properties": {}, "required": [] } }

模型依靠这段 schema 决定“要不要调这个工具、参数怎么填”。所以写工具 docstring 时我建议刻意写上“什么时候用、什么时候别用、参数含义”。描述写得越清晰，模型调用越准确，幻觉调用越少。

3.2 bind_tools 给模型一个工具清单

定义好工具后，需要让模型实例“看到”这个清单：

llm_with_tools = llm.bind_tools([get_current_time, get_city_weather, calculator])

bind_tools本质是在发给模型的请求里注入工具 schema，提示模型“你可以选择调用这些工具”。但大模型不会真的执行工具，它只会在回答里返回一个tool_calls结构，每个元素包含工具名、参数和调用 ID：

ai_message = llm_with_tools.invoke("云州今天天气怎么样？") print(ai_message.tool_calls)

输出大概是这样：

[ { "name": "get_city_weather", "args": {"city": "云州"}, "id": "call_abc123", "type": "tool_call" } ]

到了这一步，真正执行工具的责任还在我们开发者手里。你需要手动调用函数，生成一条ToolMessage，再把它和之前的消息一起喂回模型，模型才能基于真实工具结果给出回答。这个“手动回传”的过程，就是后面 Agent 循环的核心。

3.3 bind_tools 与 with_structured_output 的区别

很多初学者会把bind_tools和with_structured_output搞混，其实它们的定位完全不同。

我用一句话总结：bind_tools是让模型拿到工具清单自由发挥；with_structured_output是给模型戴上模具强制按格式输出。如果你的需求只是“从文本里提取一个日期”，用with_structured_output更合适；如果你需要模型决定“接下来该查天气还是该算数”，才需要bind_tools。

4. 实战一：确定性多工具链

先做一个简单的确定性多工具链：用户输入一句包含计算需求的话，我们强制模型先调用计算器工具，再根据计算结果生成最终回答。流程在写代码时就固定了，不存在分支和循环。

4.1 工具设计

先准备一个计算器。由于计算器可能被喂入任意表达式，安全上要小心。生产环境不要直接对不可信输入执行eval，我这里先做一个字符白名单校验：

import re @tool def calculator(expression: str) -> str: """计算一个数学表达式的值。表达式内只允许出现数字、空格、+ - * / ( ) . 等字符。""" if re.fullmatch(r"[0-9+\-*/().\s]+", expression): result = eval(expression) return str(result) return "表达式不合法，请检查后重试"

再补一个查看天气的工具，数据先用内置字典模拟，实际项目里替换成气象接口就行：

@tool def get_city_weather(city: str) -> str: """查询某个城市的当前天气情况，返回天气与体感温度。""" fake_weather = { "云州": "多云转小雨，温度 19 度，体感偏凉", "滨海": "晴天，温度 27 度，体感舒适", "临江": "雷阵雨，温度 30 度，体感闷热", } return fake_weather.get(city, f"暂无 {city} 的天气数据")

4.2 确定性链路完整代码

先定义一个消息模板，告诉模型它有两种选择：直接回答问题，或调用工具。这一步不做强制约束，但确定性链我们的假设是“默认会调用一次工具”。

from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.messages import AIMessage, ToolMessage from langchain_core.runnables import RunnableLambda from langchain_ollama import ChatOllama llm = ChatOllama(model="qwen2.5:7b-instruct", temperature=0) llm_with_tools = llm.bind_tools([calculator, get_city_weather]) message_prompt = ChatPromptTemplate.from_messages( [ ("system", "你是一个能调用工具的小助手。"), ("human", "{question}"), ] ) def execute_tools(ai_message: AIMessage): tool_messages = [] for tool_call in ai_message.tool_calls: tool_name = tool_call["name"] tool_args = tool_call["args"] if tool_name == "calculator": result = calculator.invoke(tool_args) elif tool_name == "get_city_weather": result = get_city_weather.invoke(tool_args) else: result = "未知工具" tool_messages.append( ToolMessage(content=str(result), tool_call_id=tool_call["id"]) ) return tool_messages answer_prompt = ChatPromptTemplate.from_messages( [ ("system", "根据工具返回结果回答用户。"), ("human", "{question}"), ("ai", "{ai_message}"), ("tool", "{tool_message}"), ] ) def build_answer_inputs(query: str, tool_messages: list): return { "question": query, "ai_message": tool_messages[0].content if tool_messages else "无", "tool_message": tool_messages[0].content if tool_messages else "无", }

链路本身是这样：

chain = ( {"question": lambda x: x} | message_prompt | llm_with_tools | RunnableLambda(execute_tools) | RunnableLambda( lambda tm: build_answer_inputs("请根据上下文回答", tm) ) | answer_prompt | llm | StrOutputParser() )

这段链路的执行顺序是固定的：

1. 用户问题进入 prompt
2. 模型决定是否产生tool_calls
3. execute_tools执行所有工具并把结果封装成ToolMessage
4. 把工具结果塞进新 prompt
5. 模型基于工具输出生成最终答案

调用一次试试：

result = chain.invoke("帮我算一下 123 * 45 等于多少？") print(result)

如果模型正确走了工具调用，输出应该是计算器返回后再生成的答案，而不是模型瞎猜的“5535”。

4.3 确定性链的局限

跑通之后我很开心，但一到真实业务就发现问题了。

这链路是“一次性”的。如果模型第一轮只生成了tool_calls，我必须把所有工具结果一次喂回去；如果某个工具结果本身需要再被另一个工具消费，这链路就死了。比如用户问“今天是几号？把日期的月份数字乘以 42 再加 7，结果是多少？”，这需要两步：先查日期拿到月份，再拿着月份去算数。确定性链根本做不到，因为第二步的参数在写代码时未知。

更烦的一点是：如果模型第一轮就是没调用工具，execute_tools返回空列表，build_answer_inputs里取tool_messages[0]直接报错。你得写一堆判空逻辑，把链路搞得越来越臃肿。这也是我转向 Agent 循环的直接原因。

5. 确定性链到动态 Agent：转折点在哪里

5.1 确定性链的隐藏假设

确定性链默认“步骤的顺序在写代码时是已知的”，比如：先查天气，再生成回答。真实业务里这个假设几乎不成立。用户可能先问天气，再问“那我要不要穿外套”，还可能接着问“明天下雨概率呢”。如果每个场景都单独写一条链，代码量会爆炸，而且每条链的边界也没法划清。

真正的转折点在于：应当让模型根据当前信息决定下一步动作，而不是让代码作者提前决定一切。这就是 Agent 的核心理念：模型参与路由，动态选择调用哪些工具，直到它有足够信息直接回答。

5.2 Agent 的本质是“模型参与路由 + 循环”

Agent 听起来高大上，拆开看并不神秘。它的内部就是一个循环：

1. 把所有消息（系统消息、用户消息、历史消息）喂给模型
2. 模型决定：是直接回答，还是返回若干tool_calls
3. 如果返回tool_calls，执行工具，把结果作为ToolMessage追加进消息列表
4. 回到第 1 步，继续把完整的消息列表喂给模型
5. 直到某一次模型不再返回tool_calls，直接输出最终答案

在 LangChain 生态里，这个模式被封装成了AgentExecutor、create_react_agent、create_tool_calling_agent等。但封装得越黑盒，出问题时越难查。我个人建议先用 LCEL 手动实现一遍这个循环，理解了状态流转，再决定要不要上 LangGraph 这类更重的编排框架。

5.3 为什么“循环”才是动态的关键

回到那个多步场景：今天是几号？把月份乘以 42 再加 7 是多少？

这个问题的特殊性在于：第二步的数字依赖第一步的结果。确定性链写不出来的原因不是“没想到步骤”，而是第二步的输入在运行时才能确定。循环机制的意义正在于此——模型查完日期后，工具结果进入消息历史，模型看到2025-05-18后提取月份 5，才知道下一步要算5 * 42 + 7。

信息在循环中逐步生产、逐步消费，这正是动态 Agent 区别于静态链路最关键的一点。

6. 实战二：LCEL 实现的动态 Agent

我直接给出手写版工具调用 Agent 的完整实现。它比AgentExecutor更透明，你能在每一轮迭代里看到模型到底做了什么决定。

6.1 工具注册表

先维护一张工具名到工具对象的映射表，后续循环里只需按名字查找即可。

from langchain_core.tools import tool from datetime import datetime @tool def get_current_time() -> str: """返回当前系统时间，格式为 YYYY-MM-DD HH:MM:SS。""" return datetime.now().strftime("%Y-%m-%d %H:%M:%S") @tool def get_city_weather(city: str) -> str: """查询某个城市的当前天气情况，返回天气与体感温度。""" fake_weather = { "云州": "多云转小雨，温度 19 度，体感偏凉", "滨海": "晴天，温度 27 度，体感舒适", "临江": "雷阵雨，温度 30 度，体感闷热", } return fake_weather.get(city, f"暂无 {city} 的天气数据") @tool def calculator(expression: str) -> str: """计算一个数学表达式的值。表达式内只允许出现数字、空格、+ - * / ( ) . 等字符。""" if re.fullmatch(r"[0-9+\-*/().\s]+", expression): result = eval(expression) return str(result) return "表达式不合法，请检查后重试" TOOL_MAP = { "get_current_time": get_current_time, "get_city_weather": get_city_weather, "calculator": calculator, }

6.2 Agent 循环核心代码

from langchain_core.messages import AIMessage, SystemMessage, HumanMessage, ToolMessage from langchain_core.runnables import RunnableLambda from langchain_ollama import ChatOllama from typing import Sequence llm = ChatOllama(model="qwen2.5:7b-instruct", temperature=0) llm_with_tools = llm.bind_tools(list(TOOL_MAP.values())) MAX_ITERATIONS = 5 def run_agent(messages: Sequence): current_messages = list(messages) for _ in range(MAX_ITERATIONS): ai_message = llm_with_tools.invoke(current_messages) if not ai_message.tool_calls: return ai_message current_messages.append(ai_message) for tool_call in ai_message.tool_calls: tool = TOOL_MAP[tool_call["name"]] tool_result = tool.invoke(tool_call["args"]) current_messages.append( ToolMessage( content=str(tool_result), tool_call_id=tool_call["id"], ) ) return AIMessage(content="已达到最大迭代次数，无法完成回答。") agent = RunnableLambda(run_agent) messages = [ SystemMessage(content="你是一个智能助手，可以逐步调用工具来完成任务。"), HumanMessage(content="今天是几号？把日期的月份数字乘以 42 再加 7，最后告诉我结果。"), ] response = agent.invoke(messages) print(response.content)

运行过程中模型大概会经历这样几轮：

整个过程里，开发者只负责“执行工具”和“维护消息历史”，模型自己做路由决策。这才是动态 Agent 的完整形态。

6.3 与确定性链路的直观对比

选型建议：能确定步骤的场景，比如固定格式报告生成，用确定性链；步骤不确定、需要模型动态探索的场景，比如数据分析、多步计算、复杂查询，再上 Agent 循环。不要所有任务都用 Agent，那是成本爆炸的开始。

6.4 为什么还用 LangGraph

我知道很多人会问：既然手写了循环，为什么生产环境还要用 LangGraph？

手写循环最大的问题是不好控制复杂状态。比如每轮工具执行完后要更新外部数据库、要支持条件分支、要并行调用多个工具，手写循环的current_messages列表会变得很难维护。LangGraph 把状态封装成图，节点是 Runnable，边是条件判断，循环变成显式的“回边”。当你需要做人工审批节点、多 Agent 协作、子图嵌套时，LangGraph 的优势就很明显了。

但从 LCEL 到 LangGraph 之间其实是平滑过渡的，因为 LangGraph 的节点本质上还是 Runnable。把run_agent里那一轮循环拆成一个节点函数，就能无缝接到图里。

7. 实测中的几个坑与优化小技巧

7.1 工具异常会毁掉整个 Agent

Agent 循环里最大的坑就是工具抛异常。假如get_city_weather调用第三方 API 超时，异常一路向上抛，整个 Agent 就崩了，用户拿不到任何回复，更别提让模型根据异常信息换一种策略了。

最稳妥的做法是：工具内部自己兜底，永远不向 Agent 循环抛异常，而是返回错误字符串。

@tool def get_city_weather(city: str) -> str: """查询某个城市的当前天气情况。""" try: result = fake_weather_api(city) return f"{city}天气：{result}" except Exception as e: return f"查询 {city} 天气失败，原因：{e}"

把异常信息返回给模型，模型反而能理解发生了什么，甚至可能换一个工具或用其他方式补偿。我在项目里发现，用这种“错误即消息”的方式，Agent 的存活率明显比直接抛异常高很多。

7.2 循环失控与最大迭代次数

模型有时候会陷入循环，反复调用同一个工具，甚至自己编造工具参数。现象很迷惑：它已经拿到了结果，却还要“再确认一次”。这类问题跟后端模型的行为习惯有关，不同模型、不同温度的失控概率差异很大。

解决办法是在循环外圈加MAX_ITERATIONS限制，并为超限情况设计出口。我习惯设置为 5 到 10，超过后让 Agent 拿当前已有的工具结果尝试回答，而不是直接报错。

另一个优化是给系统提示词加约束：“如果你已经拿到足够信息，必须直接回答；不要重复调用相同工具。”这行提示能显著减少无意义的循环。

7.3 Token 消耗与消息历史膨胀

Agent 循环每轮都会把完整消息历史重新发给模型，工具一多、轮次一多，Token 开销会快速上涨。我有一次只让它查三个城市的天气，结果它连续调用了七八次工具，中间的消息堆了两三千 token。

几个实际建议：

• 工具返回内容尽量精简，能只返回“温度 19 度、小雨”就不要返回一长串 JSON
• 历史消息只保留最近几轮，更早的可以截断或总结
• 对工具结果做摘要，尤其是长文本
• 确认模型真的需要多个工具时再放开多工具绑定，否则绑两三个必要的就够了

7.4 流式输出与 Agent 的兼容问题

确定性链用chain.stream()很顺畅，但 Agent 循环里流式就变得很别扭，因为一次最终回答要经历好几轮 LLM 调用，中途还有工具执行。LCEL 的.stream()按最外层 Runnable 的粒度输出，Agent 循环内部多次.invoke()不会被打包成一个流畅的流。

实践中的折中方案是：要么只用不流式的.invoke()，要么在实现里用stream接收每一轮增量，再把工具执行过程用回调或额外接口暴露给前端。别指望 Agent 版本的.stream()能和普通链一样自然。

7.5 调试三板斧

Agent 一旦行为不对，先别急着改代码，按这个顺序排查：

1. 打印每一轮的ai_message.tool_calls。这一步能直接看出模型到底有没有想做工具调用，以及参数填得对不对。
2. 打印每轮ToolMessage的内容。工具结果是否合法很关键，很多 Agent 翻车是因为工具返回了空字符串或异常文本。
3. 用agent.get_graph().print_ascii()查看 Runnable 图结构。它会以文本形式完整展现调用链，比空想可靠得多。

写在最后

这次从确定性链改造成动态 Agent，最大的感触是：Agent 不是一个神秘的黑盒，它只是把“路由决策权”交还给模型，同时用循环让决策结果能被下游消费。LCEL 的 Runnable 接口把整个过程做成了可组合的组件，无论是简单链还是循环 Agent，都能在一个统一的模型下描述。

如果你也卡在“这里要不要换成 Agent”的决策上，我的建议是先老老实实用确定性链，直到确定性链已经明显写不动了，再上 Agent 循环，并且务必保留MAX_ITERATIONS、工具异常兜底、逐轮日志这三件套。踩过几次坑之后，你会回来感谢这三件套的。