当前位置：首页 > news >正文

LangChain、LangGraph与LangSmith：构建复杂AI智能体的分层架构与实践

news 2026/7/5 11:07:20

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

1. 先搞清楚 LangChain、LangGraph、LangSmith 到底在解决什么问题

如果你正在接触 AI 应用开发，尤其是想做一个能处理多步骤、有状态交互的智能体（Agent），那么 LangChain、LangGraph 和 LangSmith 这三个名字你肯定绕不开。很多人一上来就去看文档、跑示例，但往往越看越迷糊：它们仨到底有什么区别？我该先学哪个？做一个像医疗问诊这样的复杂 Agent，到底需要用到哪一层？

简单来说，你可以这样理解：

LangChain是你的基础工具箱。它提供了连接大语言模型（LLM）、处理文本、调用工具（比如搜索、计算、查数据库）的标准件。你想让 AI 调用一个函数或读取一个文件，LangChain 帮你把轮子造好了。
LangGraph是你的流程设计器。当你的任务不是一问一答，而是像看病一样需要分步骤（挂号、分诊、检查、开药）、有状态（知道上一步做了什么）、甚至能循环或并行执行时，LangGraph 帮你用“图”的结构来清晰定义和控制这个流程。它是建立在 LangChain 之上的。
LangSmith是你的监控调试平台。前面两个工具帮你把应用跑起来了，但为什么效果不好？哪一步出错了？耗时多少？LangSmith 提供一整套的日志、追踪、测试和评估能力，让你能从“黑盒”开发进入“白盒”优化。

所以，一个医疗问诊 Agent 的诞生，恰恰是串联这三者的绝佳案例。它不是一个简单的聊天机器人，而是需要严谨的流程、专业的工具调用和持续的效果优化。下面，我就以一个虚拟的“AI家庭医生”Agent 为例，带你走一遍从零到一的构建思路，重点不是给你一堆代码，而是让你理解在每一步，这三个工具分别扮演什么角色，以及一个从业者会按什么顺序去思考和落地。

2. 需求拆解：医疗问诊 Agent 需要哪些核心能力？

在动手写任何代码之前，我们必须把需求拆解成技术模块。一个合格的医疗问诊 Agent，至少需要这几层能力：

2.1 信息收集与理解层

用户输入通常是模糊的，如“我头疼三天了，还有点发烧”。Agent 需要：

提取关键医学实体：症状（头疼、发烧）、持续时间（三天）、体征（体温？）。
追问澄清：发烧具体多少度？头疼是哪种疼？有没有其他症状？这部分需要 Agent 具备“多轮对话”和“状态记忆”能力。
初步意图分类：用户是寻求诊断建议、用药咨询、就医指导，还是仅仅描述症状？

技术映射：这部分主要依赖 LLM 的基础能力（理解、生成），但LangChain的ChatPromptTemplate、MessagesPlaceholder能帮我们结构化对话历史；OutputParser能帮我们把 LLM 的非结构化回复解析成我们需要的结构化数据（比如一个包含症状列表的 JSON）。

2.2 专业工具调用层

问诊不能光靠 LLM“臆想”，必须能调用可靠的工具（知识）：

医学知识库查询：根据症状，检索可能的疾病、用药禁忌、护理常识。
药品信息查询：查询非处方药（OTC）的用法、用量、副作用。
紧急程度判断逻辑：基于规则（如高热、胸痛、意识模糊）判断是否需要建议立即就医。
本地医院/科室查询（如果集成）：根据病情推荐合适的科室。

技术映射：这是LangChain的核心场景。我们需要用Tool接口将每一个查询功能（比如一个函数或一个 API 封装）定义成 Agent 可用的“工具”。create_react_agent或create_openai_tools_agent这类高阶接口能帮我们快速构建一个能决定“何时、调用何工具”的简单 Agent。

2.3 多步骤决策与流程控制层

简单的工具调用 Agent（如 ReAct 模式）在处理线性任务时还行，但医疗问诊流程复杂：

流程分支：用户描述症状 -> Agent 判断信息是否足够 -> 不足则循环追问 -> 足够则进入分析阶段。
子流程嵌套：分析阶段可能并行触发“疾病知识查询”和“药品信息查询”，然后综合结果。
状态持久化：整个对话过程中，用户的症状列表、已询问的问题、已查询的结果都需要作为一个状态被维护和传递，而不是每次交互都从头开始。

技术映射：当流程出现循环、条件分支、并行或复杂状态管理时，简单的 LangChain Agent 就显得力不从心了。这正是LangGraph的用武之地。它允许我们用节点（Node）和边（Edge）来绘制一个清晰的“问诊流程图”，每个节点可以是一个 LLM 调用、一个工具调用或一个条件判断，状态在整个图中流转。

2.4 效果监控与迭代优化层

开发完成，部署测试，问题来了：

追踪困难：用户说“效果不好”，到底是哪一步没做好？是症状提取错了，还是知识库没查到，还是最终建议生成得不好？
评估量化：如何系统性评估 Agent 回复的安全性、准确性和有用性？
版本对比：我调整了提示词（Prompt）或流程，新版本比旧版本到底好在哪里？

技术映射：这就是LangSmith的价值。它在你的 LangChain/LangGraph 应用运行时自动记录每一次 LLM 调用、工具调用、中间状态，形成完整的追踪链（Trace）。你可以在一个可视化面板上回放整个问诊过程，精准定位问题。还可以用它来管理测试数据集，批量运行不同版本的 Agent 进行对比评估。

3. 环境准备与核心工具选型

理解了架构，我们来看落地需要准备什么。我的建议是：分阶段验证，不要一开始就追求大而全。

3.1 基础环境与依赖

首先，你需要一个 Python 环境（建议 3.8+）。核心安装包如下：

pip install langchain langchain-community langgraph langsmith

注意：langchain是核心库，langchain-community包含大量第三方工具和集成，langgraph是流程库，langsmith是 SDK 和客户端。

其次，你需要一个 LLM 的 API 密钥。对于学习和原型开发，OpenAI 的 GPT 系列或 Anthropic 的 Claude 系列是很好的起点，因为它们遵循了较好的工具调用（Function Calling）协议，与 LangChain 集成最顺畅。国内开发者也可以使用兼容 OpenAI API 格式的国内模型服务。

# 环境变量中设置你的 API Key import os os.environ["OPENAI_API_KEY"] = "your-api-key-here" # 如果需要 LangSmith 追踪，还需要 os.environ["LANGSMITH_API_KEY"] = "your-langsmith-api-key" os.environ["LANGSMITH_TRACING"] = "true" # 开启自动追踪 os.environ["LANGSMITH_PROJECT"] = "medical-agent-demo" # 设置项目名

3.2 工具（Tool）的实现策略

对于医疗问诊 Demo，我们实现几个简单的工具来模拟：

症状知识库查询工具：可以连接一个本地的医学症状-疾病关联 CSV 文件，或者调用一个模拟的 API。
药品信息查询工具：类似地，连接一个药品数据库。
紧急程度判断工具：一个简单的规则函数，输入症状字典，输出紧急等级。

这里的关键是，先用最简单的模拟数据把工具调用的链路跑通。例如，你的药品查询工具可以暂时只是一个返回固定文本的函数。验证核心流程比纠结于工具本身的准确性更重要。

from langchain.tools import tool @tool def search_medical_knowledge(symptom: str) -> str: """根据症状查询可能的医学知识和注意事项。""" # 模拟一个简单的字典查询 knowledge_base = { "头痛发烧": "可能是感冒或流感，建议多休息、多喝水，若体温超过38.5℃或持续不退，请就医。", "腹痛腹泻": "可能是急性肠胃炎，注意补充水分和电解质，避免油腻食物。若便血或剧烈疼痛，立即就医。" } return knowledge_base.get(symptom, "未找到相关医学知识，建议咨询专业医生。") # 类似地定义其他工具...

3.3 流程设计（Graph）的初步构思

在纸上或白板上画出你的问诊流程图。一个极简的版本可以是：

开始 | v [接收用户输入] -> [提取症状实体] -> [信息是否足够？] | | | (否) v (是) | [调用知识库工具] v | [生成追问] <---[更新状态]--- [调用药品工具] | | | v `-------> [综合分析与生成建议] -> 结束

这个图会直接指导你后续用 LangGraph 编写代码。

4. 用 LangChain 构建基础工具与简单 Agent

我们先从最基础的 LangChain 组件开始，搭建一个能进行单轮工具调用的 Agent。这是验证你环境、LLM 连接和工具定义是否正确的最快方式。

4.1 定义提示词（Prompt）模板

好的提示词是 Agent 的“灵魂”。对于医疗场景，安全性和准确性提示至关重要。

from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder system_prompt = """你是一位专业的AI家庭医生助手。你的职责是： 1. 仔细倾听用户的症状描述。 2. 在信息不足时，进行有礼貌且专业的追问。 3. 利用你拥有的工具查询可靠的医学知识。 4. 基于查询结果，给出初步的健康建议，并明确强调以下免责声明： - 你的建议不能替代专业医生的诊断。 - 如果出现[高热、胸痛、呼吸困难、严重外伤、意识障碍]等紧急症状，必须立即就医或拨打急救电话。 5. 你的回答应清晰、有条理、充满同理心。 请严格遵循以上原则。""" prompt = ChatPromptTemplate.from_messages([ ("system", system_prompt), MessagesPlaceholder(variable_name="chat_history"), # 预留位置存放对话历史 ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad"), # 预留位置存放Agent的思考过程 ])

这里MessagesPlaceholder是关键，它使得我们的对话能保持多轮记忆。

4.2 组装工具并创建 Agent

我们将之前定义的几个工具组装起来，并使用 LangChain 提供的高阶助手来创建 Agent。

from langchain_openai import ChatOpenAI # 以 OpenAI 为例 from langchain.agents import create_openai_tools_agent, AgentExecutor # 1. 初始化 LLM llm = ChatOpenAI(model="gpt-4-turbo-preview", temperature=0) # 医疗场景，温度调低以保证稳定性 # 2. 准备工具列表 tools = [search_medical_knowledge, search_drug_info, judge_emergency] # 假设都已定义 # 3. 创建 Agent agent = create_openai_tools_agent(llm, tools, prompt) # 4. 创建执行器 agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True)

verbose=True会在控制台打印详细的执行步骤，非常适合调试。handle_parsing_errors=True能避免因为 LLM 输出格式偶尔不符合预期而导致整个程序崩溃。

4.3 运行与验证

现在，我们可以运行这个基础 Agent 了。

# 单轮对话 result = agent_executor.invoke({"input": "我头痛还发烧，应该吃什么药？", "chat_history": []}) print(result["output"]) # 多轮对话（需要手动维护历史） history = [] question1 = "我肚子疼。" result1 = agent_executor.invoke({"input": question1, "chat_history": history}) history.extend([HumanMessage(content=question1), AIMessage(content=result1["output"])]) question2 = "从昨天开始的。" result2 = agent_executor.invoke({"input": question2, "chat_history": history}) print(result2["output"])

此时你会观察到什么？这个基础 Agent 已经能根据你的问题，自动判断是否需要调用工具、调用哪个工具，并整合工具结果生成回复。控制台的verbose日志会显示它的思考过程（“我要用哪个工具？”“工具返回了这个结果，所以我应该这样回答”）。

但它的局限也很明显：

流程是线性的、一次性的。它很难实现“信息不足 -> 自动追问 -> 再判断”的循环。
状态管理（聊天历史）需要你手动维护和传入，比较麻烦。
对于复杂的、有固定步骤的流程（如必须先收集完所有关键症状才能进行查询），控制力不够强。

这就引出了我们对更强大流程控制能力的需求——LangGraph。

5. 用 LangGraph 设计并实现可控的问诊流程

当你的 Agent 需要“流程图”式的精确控制时，就该 LangGraph 上场了。我们将把之前画的草图转化为代码。

5.1 定义状态（State）

首先，我们需要定义在整个流程中流转的“状态”是什么。它应该包含所有必要的信息。

from typing import TypedDict, Annotated, List import operator class AgentState(TypedDict): # 用户输入 input: str # 对话历史 chat_history: List # 已收集到的症状（结构化） collected_symptoms: Annotated[List[str], operator.add] # 关键！这表示该字段会累积 # 知识库查询结果 knowledge_result: str # 药品查询结果 drug_result: str # 紧急程度 emergency_level: str # Agent 的最终输出 output: str

这里Annotated[List[str], operator.add]是 LangGraph 的一个妙处，它声明collected_symptoms这个字段在流程中是可追加（add）的，非常适合用来累积信息。

5.2 创建节点（Nodes）

节点是流程中的步骤，每个节点是一个函数，它读取状态，执行操作，并返回更新后的状态。

节点1：症状提取与收集节点

from langchain.schema import HumanMessage, AIMessage from langgraph.graph import END def collect_symptoms(state: AgentState): """从用户输入和对话历史中提取症状实体，并更新到状态中。""" messages = state["chat_history"] + [HumanMessage(content=state["input"])] # 这里可以调用一个专门的 LLM 或函数来提取症状实体 # 为了简化，我们假设从 input 中提取关键词（实际应用需更复杂） symptoms = extract_symptoms_from_text(state["input"]) # 假设的提取函数 return {"collected_symptoms": symptoms}

节点2：信息充分性判断节点

def should_ask_more(state: AgentState): """判断当前收集的症状是否足够进行下一步分析。""" # 简单的规则：例如，至少需要两个症状，或者包含了关键症状 collected = state.get("collected_symptoms", []) if len(collected) >= 2 or any(s in collected for s in ["发烧", "胸痛", "呼吸困难"]): # 信息足够，前往“分析”节点 return "enough_info" else: # 信息不足，前往“追问”节点 return "need_more_info"

这个节点不修改状态，只做路由判断。

节点3：生成追问节点

def generate_followup(state: AgentState): """根据已收集的症状，生成一个澄清性问题。""" collected = state.get("collected_symptoms", []) # 基于已收集症状，决定追问什么 if "头痛" in collected and "发烧" not in collected: question = "请问您有发烧的症状吗？体温大概多少？" elif "腹痛" in collected: question = "腹痛的具体位置在哪里？是绞痛、胀痛还是隐痛？" else: question = "您能再详细描述一下其他不适吗？" # 将追问作为 Agent 的输出，并更新对话历史 new_history = state["chat_history"] + [ HumanMessage(content=state["input"]), AIMessage(content=question) ] return {"output": question, "chat_history": new_history}

节点4：专业分析节点（调用工具）

def professional_analysis(state: AgentState): """调用知识库和药品工具进行分析。""" symptoms_text = ", ".join(state["collected_symptoms"]) # 调用工具 knowledge = search_medical_knowledge.invoke(symptoms_text) drug_info = search_drug_info.invoke(symptoms_text) emergency = judge_emergency.invoke({"symptoms": state["collected_symptoms"]}) # 综合结果，生成最终建议（这里可以再调用一次 LLM） final_advice = llm.invoke(f""" 基于以下信息生成最终建议： 症状：{symptoms_text} 医学知识：{knowledge} 药品信息：{drug_info} 紧急程度：{emergency} 请生成包含安全免责声明的友好建议。 """).content new_history = state["chat_history"] + [ HumanMessage(content=state["input"]), AIMessage(content=final_advice) ] return { "knowledge_result": knowledge, "drug_result": drug_info, "emergency_level": emergency, "output": final_advice, "chat_history": new_history }

5.3 构建图（Graph）并编译

将节点和路由逻辑连接起来。

from langgraph.graph import StateGraph, START # 创建图 workflow = StateGraph(AgentState) # 添加节点 workflow.add_node("collect", collect_symptoms) workflow.add_node("judge", should_ask_more) # 注意：这是一个“条件”节点，特殊处理 workflow.add_node("ask_more", generate_followup) workflow.add_node("analyze", professional_analysis) # 设置边（路由） workflow.add_edge(START, "collect") # 从开始到收集症状 workflow.add_edge("collect", "judge") # 收集后进入判断 # 从判断节点出发的条件边 workflow.add_conditional_edges( "judge", lambda state: state, # 路由函数，这里我们复用 should_ask_more 节点的逻辑 { "enough_info": "analyze", "need_more_info": "ask_more" } ) workflow.add_edge("ask_more", END) # 追问后，本轮结束，等待用户下一次输入 workflow.add_edge("analyze", END) # 分析并给出建议后，本轮结束 # 编译图 app = workflow.compile()

5.4 运行 LangGraph App

现在，你可以像运行一个状态机一样运行这个问诊流程。

# 初始化状态 initial_state = { "input": "医生，我头痛。", "chat_history": [], "collected_symptoms": [], "knowledge_result": "", "drug_result": "", "emergency_level": "", "output": "" } # 执行图 final_state = app.invoke(initial_state) print(final_state["output"]) # 可能是追问，也可能是最终建议 # 如果上一步是追问，我们需要进行下一轮 # 手动模拟用户回答 next_state = { "input": "不发烧，就是一阵一阵的疼。", "chat_history": final_state["chat_history"], # 携带历史 "collected_symptoms": final_state["collected_symptoms"], # 携带已收集症状 "knowledge_result": "", "drug_result": "", "emergency_level": "", "output": "" } final_state2 = app.invoke(next_state) print(final_state2["output"])

LangGraph 的优势在这里体现得淋漓尽致：

流程可视化：app.get_graph().draw_mermaid_png()可以生成流程图，一目了然。
状态自动管理：collected_symptoms的累积、chat_history的更新，都由框架根据你的定义自动处理。
复杂逻辑封装：循环（追问）、条件分支（是否足够信息）被清晰地建模在图中，代码结构非常干净。
易于调试：因为流程是固定的，你可以精确知道当前执行到了哪个节点，状态是什么。

6. 接入 LangSmith：为你的 Agent 装上“眼睛”和“仪表盘”

当你的 Agent 在 LangGraph 中顺畅运行后，你会进入下一个阶段：它真的可靠吗？效果如何优化？这时，光靠打印日志 (verbose=True) 已经不够了。你需要 LangSmith。

6.1 配置与基础追踪

如果你在环境变量中正确设置了LANGSMITH_API_KEY和LANGSMITH_TRACING=true，那么你的 LangChain 和 LangGraph 调用会自动将追踪数据发送到 LangSmith 平台。

验证是否成功：运行一次你的 Agent 或 Graph，然后登录 LangSmith 网站，在对应的项目中，你应该能看到一次完整的“Trace”。点击进去，你可以看到一棵详细的调用树：

最外层是你的agent_executor.invoke()或app.invoke()。
展开后，能看到里面每一次对 LLM 的调用（输入、输出、耗时、Token 用量）。
继续展开，能看到每一次工具调用（输入参数、返回结果）。
对于 LangGraph，你甚至能看到每个节点的执行顺序和状态快照。

6.2 利用追踪进行调试

假设一个用户反馈：“我问头痛，Agent 却去查了腹痛的药。” 没有 LangSmith，你只能靠猜。有了 LangSmith，你可以：

在 LangSmith 中找到这次出错的 Trace。
查看 “collect_symptoms” 节点（或你的症状提取 LLM 调用）的输入和输出。是不是 LLM 提取症状时出错了？
查看 “judge” 节点的路由决策。它基于什么做出了错误判断？
查看 “analyze” 节点中工具调用的输入。是不是把错误症状传给了工具？

通过这个可视化的调用链，你能在几分钟内定位到问题根源，可能是提示词不清晰、LLM 温度太高、或者工具输入格式不对。

6.3 创建数据集与评估

当你想系统性地提升 Agent 质量时，需要科学评估。

创建数据集：在 LangSmith 中，创建一个“数据集”，录入几十条典型的用户问诊输入，并人工标注上你认为理想的输出（或至少标注关键点，如“必须提及就医建议”）。
配置评估器：LangSmith 提供了一些开箱即用的评估器（如与参考答案的相似度、LLM 辅助评估等），你也可以自定义评估函数（例如，检查回复中是否包含安全免责声明）。
批量测试与对比：将你的 Agent（或 Graph）作为一个“运行器”配置到 LangSmith。然后针对这个数据集运行测试。LangSmith 会自动为每条测试用例生成 Trace 和评估分数。
迭代优化：你修改了提示词，或者调整了 Graph 的流程，可以创建一个新版本（Version）的运行器，再次对同一数据集运行测试。LangSmith 会提供清晰的对比报告，告诉你新版在各项指标上是进步了还是退步了。

6.4 生产环境监控

当 Agent 部署上线后，LangSmith 可以持续收集生产环境的追踪数据。你可以设置警报，例如当某工具调用失败率超过 5%，或平均响应时间超过 10 秒时，触发通知。这让你能从“救火”转向“预防”。

7. 实战避坑与进阶思考

走完从 LangChain 到 LangGraph 再到 LangSmith 的流程，你已经掌握了构建复杂 Agent 的核心方法论。最后，分享几个从实战中总结的关键点。

7.1 工具设计的可靠性优先

工具要“傻”：工具函数应该专注于完成单一、明确、可靠的任务。不要在一个工具里做太多逻辑。比如，search_medical_knowledge就只负责查询，不要在里面做症状过滤或结果排序。
做好错误处理：工具调用可能失败（网络、API 限制）。在你的工具函数或 Agent 执行器中，必须有完善的try...except，并返回结构化的错误信息，让 LLM 或 Graph 能理解并做出反应（如“知识库暂时不可用，请稍后再试”）。
输入验证：对工具输入进行清洗和验证。防止用户输入或 LLM 生成的错误参数导致工具崩溃。

7.2 LangGraph 状态管理的艺术

状态字段要精简：只把真正需要在节点间传递的数据放进 State。过大的状态会影响性能和清晰度。
善用Annotated修饰符：除了operator.add（追加），还有operator.set（覆盖）等，明确字段的更新语义，避免意外。
规划好“循环”的出口：像“追问”循环，必须有终止条件（如最多追问 3 次，或用户明确说“没有更多了”），否则会陷入死循环。这个条件判断可以放在should_ask_more节点中。

7.3 提示词工程与安全

系统提示词是护栏：医疗、法律、金融等领域的 Agent，系统提示词必须反复打磨，强制加入安全声明和职责限制。可以考虑使用 LangChain 的Hub来管理和版本化你的提示词。
少样本（Few-shot）示例：在提示词中提供几个正确调用工具和规范回答的例子，能显著提升 Agent 行为的稳定性和安全性。
输出解析（Output Parser）：强制 LLM 的输出符合特定格式（如 JSON），能极大简化后续处理逻辑，减少解析错误。

7.4 从 Demo 到生产的关键跨越

会话（Session）管理：上述 Demo 中，我们手动传递chat_history。在生产中，你需要一个会话管理系统（如使用数据库或 Redis）来关联用户 ID 和其对应的对话状态（对于 LangGraph，就是那个AgentState字典）。
异步与并发：使用ainvoke替代invoke来支持异步处理，应对高并发请求。LangChain/LangGraph 对异步有良好支持。
可观测性全覆盖：LangSmith 追踪了 LLM 和工具调用，但你自己的业务逻辑、数据库查询等也需要加入日志和监控，形成完整的可观测性链条。
成本与延迟优化：在 LangSmith 中密切关注 Token 消耗和延迟。考虑对常用知识进行 Embedding 并做向量检索（RAG），而不是每次都让 LLM 生成；对工具调用结果进行缓存。

构建一个像医疗问诊这样的复杂 Agent，本质上是在构建一个软件系统。LangChain 提供了砖瓦，LangGraph 提供了蓝图和施工流程，而 LangSmith 提供了工程监理和质量检测。我的建议是，先从 LangChain 的单轮工具调用 Agent 开始，确保基础链路畅通；当流程逻辑变得复杂时，引入 LangGraph 来重构；而在整个开发和迭代过程中，尽早接入 LangSmith，让它帮你看清每一步，让优化过程从“凭感觉”变成“看数据”。