LangChain AI Agent 架构实战:从链到图的五大核心概念与落地路径

📅 2026/7/5 6:08:06 👁️ 阅读次数 📝 编程学习
LangChain AI Agent 架构实战:从链到图的五大核心概念与落地路径

🚀 30+款热门AI模型一站整合,DeepSeek/GLM/Qwen 随心用,限时 5 折。 👉 点击领海量免费额度

这类工具最值得先看的不是功能列表,而是能不能在普通环境里稳定跑起来。LangChain 和围绕它的 AI Agent 概念,现在讨论热度很高,但很多刚接触的朋友容易陷入两个误区:要么觉得它太复杂,一上来就被各种概念吓退;要么觉得它无所不能,直接拿官方示例去跑生产任务,结果在环境、依赖和流程设计上踩坑。

我建议先从最小样例开始。这篇文章不会罗列所有功能,而是围绕5 个最核心、最能帮你理解 LangChain AI Agent 架构的概念,拆清楚它们各自解决什么问题,以及在实际搭建一个能用的 Agent 时,你应该按什么顺序去验证。这五个概念是:链(Chains)、代理(Agents)、工具(Tools)、记忆(Memory)和状态(State)。如果你正在评估是否要用 LangChain 来开发 AI 应用,或者已经上手但感觉流程不清晰,这篇文章会帮你理出一条从单任务验证到工作流设计的实操路径。

1. 先理解“链”:它不只是把调用连起来

很多人第一次接触 LangChain,看到的第一个例子可能就是LLMChain。代码看起来很简单:一个提示词模板(PromptTemplate)加一个大语言模型(LLM),跑起来就能得到结果。但这容易让人产生误解,以为“链”就是简单的顺序调用。

1.1 链的核心是“确定性工作流”

链(Chain)在 LangChain 里的核心价值,是定义一个确定性的、可复用的执行序列。它把多个组件(比如:提示词、模型、解析器、其他链)按固定顺序组装起来。这里的“确定性”是关键:给定相同的输入,链应该产生相同的输出(在不涉及随机性的情况下)。这听起来简单,但它是构建更复杂、非确定性 Agent 的基础。

为什么先要从链开始?因为它是你验证环境、模型连接和基础逻辑是否通畅的最小单元。我一般会建议按这个顺序测试:

  1. 环境连通性测试:用一个最简单的LLMChain,问一个诸如“请说‘你好’”的问题。目的不是得到有意义的答案,而是确认你的 API Key 有效、网络通畅、依赖包版本兼容。很多后续的复杂错误,根源都出在这里。
  2. 多步骤逻辑测试:创建一个包含检索(Retrieval)和总结(Summarization)的链。例如,先从一篇文章里检索关键段落,再让模型总结。这里你测试的是组件间的数据传递是否正常。
  3. 输出解析测试:使用PydanticOutputParserStructuredOutputParser,让模型返回结构化的 JSON 数据。这是将自然语言输出转化为程序可处理数据的关键一步,很多实际应用都依赖于此。

一个常见的坑是,链里某个组件失败了,但错误信息被淹没。所以,在构建链时,我通常会先给每个组件设置独立的日志或打印语句,确认数据流到了哪一步。

1.2 链的两种主要类型:LCEL 与 Legacy Chain

LangChain 提供了两种构建链的方式,这也是新手容易困惑的地方。

  • LCEL(LangChain Expression Language):这是当前推荐的方式。它使用管道符|来连接组件,写法更声明式、更 Pythonic,并且原生支持流式输出、异步、批量处理等高级特性。
    # LCEL 方式示例 from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI prompt = ChatPromptTemplate.from_template("请用一句话介绍 {topic}") model = ChatOpenAI(model="gpt-3.5-turbo") chain = prompt | model result = chain.invoke({"topic": "机器学习"})
  • Legacy Chain:早期的方式,通过继承Chain类或使用SequentialChain等。代码看起来更面向对象,但在灵活性和功能支持上不如 LCEL。

我的建议是,除非维护旧代码,否则一律从 LCEL 开始学起和用起。它的设计更符合现代数据流的思想,并且官方的新特性会优先在 LCEL 上支持。

2. 掌握“代理”与“工具”:让 AI 学会使用外部能力

如果链是固定的“流水线”,那么代理(Agent)就是配备了“大脑”和“工具箱”的自主决策者。这是实现 AI Agent 智能体的核心。

2.1 代理 = 大脑(LLM)+ 工具集 + 决策逻辑

一个代理的核心部件包括:

  1. LLM:负责思考、规划和决策。
  2. Tools:代理可以调用的外部函数,比如搜索网络、查询数据库、执行计算、调用 API。
  3. Agent Executor:驱动代理运行的引擎,它负责调用 LLM、解析其决定(是使用工具还是直接回答)、执行工具、并将工具结果返回给 LLM 进行下一轮思考,直到得出最终答案。

代理的工作模式是非确定性的。同样的问题,根据中间工具执行结果的不同,代理的思考路径和最终答案可能不同。这模拟了人类解决问题时“尝试-观察-调整”的过程。

2.2 工具(Tools)的设计与验证

工具是代理能力的延伸。设计一个好的工具至关重要:

  • 单一职责:一个工具只做一件事。例如,search_web负责搜索,calculate负责计算,query_database负责查库。不要设计一个“万能工具”。
  • 清晰的描述(Description):工具的描述(description参数)是给 LLM 看的“说明书”。描述必须清晰、准确,说明工具的用途、输入参数和输出。LLM 靠这个来决定是否以及如何调用它。
  • 健壮的错误处理:工具函数内部必须有try...except,避免因为网络超时、数据格式异常等问题导致整个代理崩溃。工具应返回明确的错误信息,供 LLM 或执行器处理。

在让代理使用工具前,务必单独测试每个工具。用一些典型的输入参数手动调用工具函数,确保它能返回你期望的结果。这是后续代理能正常工作的基础。

2.3 代理执行流程与常见问题

当你运行一个代理时,Agent Executor内部大致是这样的循环:

  1. 将用户问题、历史对话(记忆)和可用工具列表传给 LLM。
  2. LLM 返回一个“动作”(Action),格式通常是工具名:工具输入
  3. 执行器解析动作,调用对应的工具。
  4. 获取工具返回的“观察结果”(Observation)。
  5. 将动作和观察结果追加到对话历史中。
  6. 将更新后的历史再次传给 LLM,进行下一轮思考。
  7. 重复以上步骤,直到 LLM 返回一个“最终答案”(Final Answer)。

在这个过程中,最容易出问题的地方是:

  • 工具选择错误:LLM 调用了错误的工具。这通常需要优化工具描述或提示词。
  • 无限循环:代理在两个动作间来回切换,无法结束。需要设置max_iterationsmax_execution_time来强制终止。
  • 解析失败:LLM 返回的文本无法被解析为有效的动作或答案。需要检查提示词中关于输出格式的指令是否明确。

调试代理时,一定要开启详细日志(verbose=True),这样你能看到每一轮 LLM 的思考、它决定的动作以及工具返回的结果,这是定位问题最直接的方法。

3. 用好“记忆”:让对话拥有上下文

没有记忆的 Agent,每次对话都是独立的,无法进行连贯的多轮交流。LangChain 中的记忆(Memory)组件就是用来管理对话历史的。

3.1 记忆的两种粒度:对话历史与实体记忆

记忆不仅仅是保存上一轮的用户消息和 AI 回复。从实用角度,可以分为两类:

  1. 对话历史(Conversation Buffer):最简单直接的方式,保存完整的对话记录。ConversationBufferMemory就是这种。优点是信息完整,缺点是对话变长后,会消耗大量 Token,可能触发模型上下文长度限制,且无关信息可能干扰模型。
  2. 摘要记忆或实体记忆:更高级的方式。ConversationSummaryMemory会定期让 LLM 对之前的对话进行摘要,只保存摘要,节省空间。ConversationEntityMemory则尝试识别和记忆对话中提到的具体实体(如人名、地点、任务指标),并在后续对话中关联这些实体。这些方式更智能,但也更复杂,可能引入摘要偏差。

对于大多数应用,我建议从ConversationBufferWindowMemory开始。它只保留最近 K 轮对话,在上下文长度和记忆效果间取得了一个很好的平衡。你可以先设定k=3k=5,根据实际效果调整。

3.2 记忆如何与链和代理集成

记忆对象需要被正确地注入到链或代理中。对于 LCEL 风格的链,通常使用RunnableWithMessageHistory这个包装器。它需要一个返回独立会话 ID 的函数,以便为不同用户或不同对话线程维护独立的记忆。

from langchain.memory import ChatMessageHistory from langchain_core.runnables.history import RunnableWithMessageHistory # 假设 chain 是你的基础链 chain_with_memory = RunnableWithMessageHistory( chain, get_session_history=lambda session_id: ChatMessageHistory(), # 这里应返回对应 session_id 的记忆存储 input_messages_key="input", # 你的链中,用户输入对应的键名 history_messages_key="history", # 你的链中,历史消息对应的键名 )

这里的关键是get_session_history函数。在生产环境中,你需要实现这个函数,使其能够根据session_id从数据库(如 Redis、PostgreSQL)或文件中加载和保存对应的ChatMessageHistory对象。不要小看这一步,它是实现多用户、长会话支持的核心。

对于代理,记忆通常通过AgentExecutormemory参数直接传入,代理执行器会自动负责在每一轮思考后将消息存入记忆。

4. 理解“状态”:LangGraph 与复杂工作流

当你需要构建的不仅仅是简单的问答代理,而是包含分支、循环、并行、人工审核等步骤的复杂工作流时,传统的链或代理就显得力不从心了。这就是LangGraph出场的时候,而它的核心抽象就是状态(State)

4.1 从链到图:工作流哲学的转变

如前文搜索材料中提到的,LangChain 的链是“链式工作流”,而 LangGraph 是“图结构工作流”。这个区别至关重要:

  • :A -> B -> C。固定顺序,一条路走到黑。
  • :节点(Node)和边(Edge)组成。执行可以从一个节点到另一个节点,也可以根据条件(边)跳转到不同的节点,甚至可以循环回到之前的节点。

这种图结构非常适合描述现实中的业务流程,比如:“先分析用户需求,然后根据需求类型决定是调用工具A还是工具B,调用后检查结果质量,如果合格就结束,如果不合格就转人工审核或重试。”

4.2 状态(State)是什么?

在图工作流中,每个节点都是一个函数。这些函数需要共享和修改一些数据。这些数据就是“状态”(State)。在 LangGraph 中,状态通常是一个字典(TypedDict),定义了工作流中所有需要流转的信息,例如:

  • messages: 对话消息列表。
  • user_query: 原始用户问题。
  • selected_tool: 当前选中的工具名。
  • tool_result: 工具执行的结果。
  • needs_human_review: 一个布尔值,标记是否需要人工介入。

每个节点函数接收当前状态作为输入,执行一些操作(如调用 LLM、运行工具),然后返回一个更新后的状态(或状态的一部分)。LangGraph 的引擎负责将这些更新合并到全局状态中,并决定下一个要执行的节点。

4.3 一个简单的 LangGraph 工作流示例

假设我们要构建一个带有关键词检查的问答流程:先检查用户问题是否包含敏感词,如果是,则直接回复拒绝;如果不是,则调用 LLM 正常回答。

from typing import TypedDict, Literal from langgraph.graph import StateGraph, END # 1. 定义状态 class GraphState(TypedDict): user_input: str contains_sensitive_word: bool response: str # 2. 定义节点函数 def check_sensitive_word(state: GraphState) -> GraphState: sensitive_words = ["暴力", "违禁词"] contains = any(word in state["user_input"] for word in sensitive_words) return {"contains_sensitive_word": contains} def generate_rejection(state: GraphState) -> GraphState: return {"response": "您的问题包含不当内容,无法回答。"} def generate_answer(state: GraphState) -> GraphState: # 这里模拟调用 LLM answer = f"根据您的问题'{state['user_input']}',模拟生成的答案是..." return {"response": answer} # 3. 构建图 workflow = StateGraph(GraphState) # 添加节点 workflow.add_node("check", check_sensitive_word) workflow.add_node("reject", generate_rejection) workflow.add_node("answer", generate_answer) # 设置入口点 workflow.set_entry_point("check") # 添加条件边 def decide_next_step(state: GraphState) -> Literal["reject", "answer"]: if state["contains_sensitive_word"]: return "reject" else: return "answer" workflow.add_conditional_edges( "check", decide_next_step, { "reject": "reject", "answer": "answer", } ) # 添加普通边(指向结束) workflow.add_edge("reject", END) workflow.add_edge("answer", END) # 编译图 app = workflow.compile() # 4. 执行 initial_state = {"user_input": "请介绍一下机器学习。", "contains_sensitive_word": False, "response": ""} result = app.invoke(initial_state) print(result["response"])

这个例子展示了状态如何在不同节点间传递和修改,以及如何根据状态值(contains_sensitive_word)决定下一步走向。对于更复杂的、需要循环(比如代理的多轮工具调用)或并行(同时调用多个 API)的场景,LangGraph 提供了更强大的原语(如tools_condition边、Pregel引擎)来支持。

什么时候该用 LangGraph?当你的业务逻辑无法用一条直线(链)描述,而是充满了“如果...那么...”、“循环直到...”、“同时做 A 和 B”这样的逻辑时,就该考虑 LangGraph 了。

5. 架构落地:从概念到可运行系统的关键步骤

理解了以上五个核心概念,最后我们来梳理一下,如何将它们组合起来,搭建一个真正能运行的 AI Agent 系统。这个过程不是一蹴而就的,我建议遵循以下步骤:

5.1 第一步:环境与基础链验证

不要一开始就想着构建复杂的代理或图。首先,确保你的开发环境是干净的。

  1. 创建新的虚拟环境(conda 或 venv)。
  2. 安装 LangChain 及相关包(如langchain-openai,langchain-community)。强烈建议使用poetryuv进行依赖管理,并锁定版本,避免后续因版本升级导致的不兼容。
  3. 写一个最简单的LLMChain(用 LCEL 方式),调用 OpenAI 或本地模型,确保能收到回复。这一步验证了网络、API 密钥和基础库的连通性。

5.2 第二步:设计并测试工具

根据你的 Agent 需要的能力,设计工具函数。

  1. 每个工具函数都要有清晰的输入输出和错误处理。
  2. 编写单元测试,模拟各种正常和异常输入,确保工具行为符合预期。
  3. 将这些工具封装成 LangChain 的Tool对象,并写好描述。

5.3 第三步:构建并调试单一代理

使用上一步创建的工具,构建一个简单的代理。

  1. 选择合适的代理类型(如ReAct代理,它鼓励模型“思考”后再行动,适合复杂任务)。
  2. 创建AgentExecutor,传入 LLM、工具列表,并设置max_iterations(例如 10)防止无限循环。
  3. 开启verbose=True,用几个典型问题测试代理。仔细观察日志:
    • LLM 的“思考”过程是否合理?
    • 它是否正确地选择了工具?
    • 工具返回的结果是否被正确理解?
    • 代理是否在合适的时机结束了?
  4. 根据测试结果,反复优化提示词和工具描述。

5.4 第四步:引入记忆管理

为你的代理加上记忆,支持多轮对话。

  1. 选择一种记忆类型(如ConversationBufferWindowMemory)。
  2. 将其集成到代理执行器中。
  3. 测试多轮对话:第二轮的回复是否能正确引用第一轮的信息?
  4. 关键点:设计并实现session_id的管理机制。对于 Web 应用,这通常对应一个用户会话 ID;对于 CLI 工具,可能对应一个线程 ID。你需要一个存储后端(内存、数据库、文件)来持久化每个会话的记忆。

5.5 第五步:用 LangGraph 编排复杂流程(如需要)

如果你的业务逻辑需要分支、循环或并行,将第 3 步调试好的代理作为一个“节点”,嵌入到 LangGraph 定义的工作流中。

  1. 明确定义你的工作流状态(State),想清楚每个节点需要读什么、写什么。
  2. 将代理节点、工具调用节点、条件判断节点等添加到图中。
  3. 定义节点之间的边(条件边或普通边)。
  4. 编译图,并用一系列测试用例验证整个工作流是否符合预期。特别注意循环和条件分支的路径。

5.6 生产化考量

当核心流程跑通后,就需要考虑生产环境的要求:

  • 错误处理与重试:为 LLM 调用、工具调用(尤其是网络 API)添加重试机制和超时控制。
  • 日志与监控:记录详细的执行日志,包括每轮交互的输入、输出、工具调用详情、耗时等,便于问题排查和性能分析。
  • 速率限制:如果你使用付费 API,需要在应用层面实施速率限制,避免意外超支。
  • 可观测性:考虑集成像 LangSmith 这样的平台,它可以可视化跟踪链、代理和图的所有执行步骤,是调试和优化不可或缺的工具。

最后,回到最初的观点:LangChain 提供的这套概念和组件,本质上是帮你更好地组织和管理与大模型交互的复杂性。它不能替代你对业务逻辑的深刻理解,也不能自动解决所有问题。最有效的学习方式,就是从一个明确的小目标开始,动手搭建,观察每一步的输出,遇到问题就深入排查对应的概念和组件。把这五个核心概念——链、代理、工具、记忆、状态——之间的关系和适用场景想清楚,你就能更自信地设计和实现属于自己的 AI Agent 架构。

🚀 30+款热门AI模型一站整合,DeepSeek/GLM/Qwen 随心用,限时 5 折。 👉 点击领海量免费额度