LangChain、LangGraph与LangSmith实战：构建医疗问诊智能体全流程

30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度大家好我是专注于AI应用开发的技术博主。在探索大模型落地的过程中你是否也遇到过这样的困惑看了一堆LangChain、LangGraph、LangSmith的教程概念都懂但一到实际项目还是不知道如何将它们串联起来构建一个真正能跑起来的智能体Agent特别是当业务场景复杂需要多步骤决策和状态管理时更是无从下手。本文将以一个医疗问诊智能体Medical Consultation Agent的完整构建过程为例手把手带你从零到一串联起LangChain、LangGraph和LangSmith三大核心工具。你将不仅理解它们各自的作用更能掌握如何将它们有机组合打造一个具备多轮对话、知识检索、逻辑判断和工具调用能力的实用AI应用。无论你是刚接触Agent开发的新手还是希望深化工程实践的开发者这篇系统化的实战指南都能让你获得可直接复用的代码和经验。1. 背景与核心概念为什么需要这套组合拳在深入代码之前我们有必要厘清这三个核心工具的角色与关系。它们并非相互替代而是各司其职共同构成了现代AI Agent开发的“铁三角”。LangChain应用构建的“脚手架”你可以把LangChain理解为一套高度封装的乐高积木。它提供了大量预构建的模块如提示词模板、文档加载器、文本分割器、向量存储接口、各种链Chain和代理Agent极大地简化了与大模型交互、处理非结构化数据、构建对话流程的复杂度。它的核心价值在于标准化和提速让你不用从HTTP请求开始写起。LangGraph复杂工作流的“调度中心”当你的智能体逻辑变得复杂不再是简单的“一问一答”或线性链式调用时就需要LangGraph。它基于有状态图StateGraph的概念允许你定义包含条件分支、循环、并行执行节点的复杂工作流。在医疗问诊场景中患者描述症状后Agent可能需要先查询知识库然后根据结果决定是直接给出建议还是继续追问更多细节这个动态的、带状态的决策过程正是LangGraph的用武之地。简言之LangChain帮你造零件LangGraph帮你设计并运行整个自动化流水线。LangSmith开发与运维的“监控仪表盘”这是整个开发流程的“眼睛”。LangSmith是一个平台用于跟踪、调试、测试和评估基于LangChain/LangGraph构建的应用。它能记录每一次链或图的调用详情输入、输出、中间步骤、耗时、Token消耗帮助你直观地分析问题、优化提示词、评估效果。没有它调试一个多步Agent就像在黑暗中修车。我们的项目目标利用LangChain快速搭建基础能力如RAG检索利用LangGraph编排复杂的问诊决策流程并最终使用LangSmith对整个系统进行监控、调试和优化打造一个原型级医疗问诊Agent。2. 环境准备与版本说明工欲善其事必先利其器。以下是构建本项目所需的环境。请注意AI领域工具迭代迅速以下版本为撰写时的稳定版本核心逻辑具有普适性。基础环境操作系统Windows 10/11, macOS 或 Linux (Ubuntu 20.04) 均可。Python版本 3.10。强烈建议使用虚拟环境如venv或conda。核心库安装在项目根目录下创建requirements.txt文件并填入以下内容langchain0.1.0 langchain-community0.0.10 langchain-openai0.0.5 langgraph0.0.26 langsmith0.0.87 openai1.6.1 chromadb0.4.22 tiktoken0.5.2 fastapi0.104.1 uvicorn[standard]0.24.0 pydantic2.5.0 python-dotenv1.0.0通过pip安装pip install -r requirements.txt关键服务与APIOpenAI API本项目使用gpt-3.5-turbo作为核心LLM。你需要准备一个有效的OpenAI API密钥。LangSmith前往 LangSmith官网 注册并创建API密钥。它提供免费额度足够用于开发和调试。向量数据库我们使用轻量级的ChromaDB作为本地向量数据库来存储医学知识无需额外部署。项目结构预览在开始前我们先规划好项目目录这有助于理解代码组织。medical_agent_project/ ├── .env # 存储敏感密钥如OPENAI_API_KEY, LANGSMITH_API_KEY ├── requirements.txt # 项目依赖 ├── main.py # 应用主入口FastAPI服务 ├── agent/ # 智能体核心模块 │ ├── __init__.py │ ├── knowledge_base.py # 知识库构建与检索RAG │ ├── graph_state.py # 定义LangGraph的状态State │ └── workflow.py # 定义LangGraph的工作流节点与图 ├── data/ # 存放原始知识文档 │ └── medical_guidelines.txt └── logs/ # 可选日志目录接下来我们将从最基础的知识库构建开始逐步向上搭建整个智能体。3. 核心模块拆解与实现3.1 第一步构建医学知识库RAG with LangChain智能体不能只靠大模型的通用知识必须结合专业领域信息。我们使用LangChain快速实现一个RAG检索增强生成系统。首先设置环境变量。创建.env文件OPENAI_API_KEYsk-your-openai-key-here LANGSMITH_API_KEYlsv2-your-langsmith-key-here LANGSMITH_PROJECTmedical-agent-demo # LangSmith中的项目名 LANGCHAIN_TRACING_V2true LANGCHAIN_ENDPOINThttps://api.smith.langchain.com然后实现知识库模块agent/knowledge_base.pyimport os from typing import List from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma from langchain.schema import Document from dotenv import load_dotenv load_dotenv() # 加载.env文件中的环境变量 class MedicalKnowledgeBase: 医学知识库类负责文档加载、分割、向量化存储与检索 def __init__(self, persist_directory: str ./chroma_db_medical): # 使用OpenAI的嵌入模型 self.embeddings OpenAIEmbeddings(modeltext-embedding-3-small) self.persist_directory persist_directory self.vectorstore None def build_from_file(self, file_path: str): 从文本文件构建知识库 print(正在加载医学指南文档...) loader TextLoader(file_path, encodingutf-8) documents loader.load() # 文本分割将长文档切成适合检索的片段 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, # 每个片段约1000字符 chunk_overlap200, # 片段间重叠200字符保持上下文 separators[\n\n, \n, 。, , , , ] ) splits text_splitter.split_documents(documents) print(f文档已分割为 {len(splits)} 个片段。) # 创建向量存储并持久化 self.vectorstore Chroma.from_documents( documentssplits, embeddingself.embeddings, persist_directoryself.persist_directory ) self.vectorstore.persist() print(f知识库已构建并保存至 {self.persist_directory}) def load_existing(self): 加载已构建的知识库 if os.path.exists(self.persist_directory): self.vectorstore Chroma( persist_directoryself.persist_directory, embedding_functionself.embeddings ) print(已加载现有知识库。) return True else: print(未找到已存在的知识库请先构建。) return False def query(self, question: str, k: int 3) - List[Document]: 检索与问题最相关的k个知识片段 if not self.vectorstore: raise ValueError(知识库未初始化请先构建或加载。) # similarity_search 是 LangChain 封装的检索方法 docs self.vectorstore.similarity_search(question, kk) return docs def get_retriever(self): 返回一个LangChain Retriever对象便于集成到链中 if not self.vectorstore: raise ValueError(知识库未初始化。) return self.vectorstore.as_retriever(search_kwargs{k: 3}) # 示例如何构建知识库 if __name__ __main__: kb MedicalKnowledgeBase() # 首次运行构建知识库假设有 data/medical_guidelines.txt # kb.build_from_file(./data/medical_guidelines.txt) # 后续运行直接加载 kb.load_existing() # 测试检索 test_question 普通感冒有哪些典型症状 results kb.query(test_question) print(f\n针对问题『{test_question}』检索到以下内容) for i, doc in enumerate(results): print(f\n--- 片段 {i1} ---) print(doc.page_content[:300] ...) # 打印前300字符关键点解析文档分割RecursiveCharacterTextSplitter是LangChain提供的智能分割器能较好地在句末和段落处切割保持语义完整性。向量化OpenAIEmbeddings将文本转换为向量Chroma存储这些向量并执行相似性搜索。检索器Retrieveras_retriever()方法返回的对象是LangChain生态中的标准接口可以无缝接入后续的链Chain中。3.2 第二步定义智能体的状态LangGraph StateLangGraph的核心是有状态图。状态State是一个贯穿整个工作流、在各节点间传递和修改的数据结构。我们为医疗问诊Agent定义其状态。创建agent/graph_state.pyfrom typing import TypedDict, List, Annotated, Optional import operator class GraphState(TypedDict): 定义医疗问诊智能体的工作流状态。 这是一个类型字典明确了每个字段的类型。 # 用户输入的问题或对话轮次 user_input: str # 从知识库检索到的相关文档 retrieved_docs: List[str] # 大模型生成的回复或分析结果 llm_response: str # 对话历史用于让模型记住上下文 conversation_history: Annotated[List[str], operator.add] # 当前问诊阶段用于控制流程分支 stage: str # 例如initial_symptom, follow_up, final_advice, need_human # 模型决定下一步要做什么工具调用或直接回复 next_action: Optional[str]状态字段说明user_input当前用户的最新问题如“我头痛发烧”。retrieved_docs根据当前输入检索到的医学知识。llm_response模型生成的文本回复。conversation_history这是一个累加式字段。Annotated[List[str], operator.add]是LangGraph的语法意味着每个节点都可以向这个列表中添加内容而不会覆盖之前的内容。这完美实现了对话历史的累积。stage一个关键的控制变量。例如初始阶段initial_symptom可能触发知识检索follow_up阶段可能决定是否需要追问更多症状final_advice阶段生成最终建议need_human阶段则转人工。next_action由模型决定用于指导图的下一个节点例如“answer”或“retrieve”。3.3 第三步构建工作流节点与图LangGraph这是最核心的部分我们将把问诊流程拆解成多个节点并用图连接起来。创建agent/workflow.pyfrom langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from .graph_state import GraphState from .knowledge_base import MedicalKnowledgeBase import json # 初始化大模型和知识库 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.2) # temperature调低使输出更稳定 kb MedicalKnowledgeBase() kb.load_existing() # 假设知识库已存在 def retrieve(state: GraphState): 节点1检索节点。从知识库中获取相关信息。 print(f[检索节点] 正在检索与用户输入相关的信息...) docs kb.query(state[user_input]) # 将Document对象转换为纯文本列表 state[retrieved_docs] [doc.page_content for doc in docs] print(f[检索节点] 检索到 {len(state[retrieved_docs])} 条相关信息。) return state def generate_initial_response(state: GraphState): 节点2生成初步回复/分析节点。结合检索结果和用户输入生成初步分析。 print(f[生成节点] 基于检索信息生成初步分析...) # 构建提示词模板 prompt_template ChatPromptTemplate.from_messages([ (system, 你是一位专业的AI医疗助手。请基于提供的医学知识片段和用户描述的症状进行初步分析。 知识片段 {context} 你的任务是 1. 判断用户症状是否在知识片段中有明确对应或相关指南。 2. 生成一段初步的、谨慎的分析。 3. 决定下一步是【直接给出建议】还是【需要追问更多细节】。 请以JSON格式回复包含两个字段 - “analysis”: 你的初步分析文本。 - “next_action”: “advice” 或 “question”。), (human, 用户症状描述{symptom}) ]) # 构建链 chain prompt_template | llm | StrOutputParser() # 执行 context_text \n\n---\n\n.join(state[retrieved_docs]) result chain.invoke({ context: context_text, symptom: state[user_input] }) # 解析大模型的输出假设是JSON字符串 try: parsed json.loads(result) state[llm_response] parsed.get(analysis, 分析生成失败。) state[next_action] parsed.get(next_action, advice) state[stage] initial_analysis except json.JSONDecodeError: # 如果模型没返回标准JSON做降级处理 state[llm_response] result state[next_action] advice state[stage] initial_analysis # 将本轮交互加入历史 state[conversation_history].append(f用户: {state[user_input]}) state[conversation_history].append(f助手: {state[llm_response]}) print(f[生成节点] 分析完成。下一步动作: {state[next_action]}) return state def give_final_advice(state: GraphState): 节点3给出最终建议节点。 print(f[建议节点] 正在生成最终建议...) prompt ChatPromptTemplate.from_messages([ (system, 你是医疗助手。请基于之前的对话历史和知识给用户清晰、安全、负责任的最终建议。 务必强调你的建议不能替代专业医生诊断如果症状严重或持续请立即就医。 对话历史 {history} 请生成最终建议), ]) chain prompt | llm | StrOutputParser() history_text \n.join(state[conversation_history][-6:]) # 取最近3轮对话 advice chain.invoke({history: history_text}) state[llm_response] advice state[stage] final_advice print(f[建议节点] 最终建议已生成。) return state def ask_follow_up(state: GraphState): 节点4追问细节节点。 print(f[追问节点] 正在生成追问问题...) prompt ChatPromptTemplate.from_messages([ (system, 为了更准确地评估情况你需要追问一些关键信息。请生成一个自然、关切的问题。 对话历史 {history} 请生成一个追问问题), ]) chain prompt | llm | StrOutputParser() history_text \n.join(state[conversation_history][-4:]) question chain.invoke({history: history_text}) state[llm_response] question state[stage] follow_up state[next_action] wait_for_user # 等待用户回答 print(f[追问节点] 追问问题已生成。) return state def route_after_analysis(state: GraphState): 路由函数根据next_action决定下一步走向哪个节点。 print(f[路由] 当前下一步动作: {state.get(next_action)}) if state.get(next_action) advice: return give_final_advice elif state.get(next_action) question: return ask_follow_up else: # 默认走向建议节点 return give_final_advice # 构建图 def create_workflow(): 创建并编译LangGraph工作流 workflow StateGraph(GraphState) # 添加节点 workflow.add_node(retrieve, retrieve) workflow.add_node(generate, generate_initial_response) workflow.add_node(give_advice, give_final_advice) workflow.add_node(ask_question, ask_follow_up) # 设置入口点 workflow.set_entry_point(retrieve) # 添加边连接节点 workflow.add_edge(retrieve, generate) # 在generate节点之后根据路由函数的结果决定去向 workflow.add_conditional_edges( generate, route_after_analysis, { give_final_advice: give_advice, ask_follow_up: ask_question, } ) workflow.add_edge(give_advice, END) workflow.add_edge(ask_question, END) # 注意追问后图结束等待下一次用户输入启动新图。 # 编译图 app workflow.compile() return app # 获取图应用实例 graph_app create_workflow()工作流图解文字描述开始 | v [retrieve] 检索节点根据用户输入查询知识库。 | v [generate] 生成节点结合检索结果分析并决定下一步。 | | |--(next_action “advice”)-- [give_advice] 建议节点生成最终建议 -- 结束 | |--(next_action “question”)-- [ask_question] 追问节点生成追问问题 -- 结束这个图实现了一个简单的两轮决策流程。在实际更复杂的Agent中你可以设计循环让“追问-回答”可以多次进行。3.4 第四步集成与对外服务FastAPI现在我们将编译好的LangGraph应用封装成一个Web服务方便调用。创建main.pyfrom fastapi import FastAPI, HTTPException from pydantic import BaseModel from agent.workflow import graph_app from agent.graph_state import GraphState import uvicorn import logging # 配置日志方便查看运行过程 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) app FastAPI(title医疗问诊智能体 API, version0.1.0) class QueryRequest(BaseModel): API请求体模型 user_input: str session_id: str default_session # 用于区分不同对话会话 class QueryResponse(BaseModel): API响应体模型 response: str stage: str session_id: str app.post(/chat, response_modelQueryResponse) async def chat_with_agent(request: QueryRequest): 与医疗问诊智能体对话的端点。 每次调用都会运行一次LangGraph工作流。 logger.info(f收到会话 [{request.session_id}] 的请求: {request.user_input}) # 初始化图的状态 initial_state: GraphState { user_input: request.user_input, retrieved_docs: [], llm_response: , conversation_history: [], stage: start, next_action: None } try: # 执行图 final_state graph_app.invoke(initial_state) logger.info(f会话 [{request.session_id}] 处理完成阶段: {final_state[stage]}) return QueryResponse( responsefinal_state[llm_response], stagefinal_state[stage], session_idrequest.session_id ) except Exception as e: logger.error(f处理会话 [{request.session_id}] 时出错: {e}, exc_infoTrue) raise HTTPException(status_code500, detailf智能体处理失败: {str(e)}) app.get(/health) async def health_check(): 健康检查端点 return {status: healthy} if __name__ __main__: # 启动服务访问 http://127.0.0.1:8000/docs 查看交互式API文档 uvicorn.run(app, host0.0.0.0, port8000)4. 运行、验证与LangSmith监控4.1 启动服务并测试确保你的.env文件已正确配置API密钥。在终端中运行python main.py服务启动后打开浏览器访问http://127.0.0.1:8000/docs你会看到自动生成的Swagger UI界面。在/chat端点下点击“Try it out”输入JSON请求体例如{ user_input: 我喉咙痛有点发烧是感冒吗, session_id: test_user_1 }点击“Execute”观察响应。同时在运行服务的终端里你会看到我们打印的节点执行日志。4.2 使用LangSmith进行追踪和调试这是体现LangSmith价值的时刻。因为我们在.env中设置了LANGCHAIN_TRACING_V2true所以所有通过LangChain/LangGraph的调用都会被自动记录。登录 LangSmith 网站。在左侧边栏选择你设置的项目medical-agent-demo。在“Traces”页面你将看到每次API调用产生的追踪记录。点击任意一次追踪你可以看到完整的调用链每个节点的输入Input和输出Output。每个步骤的耗时和Token使用情况。retrieve节点检索到了哪些文档片段。generate节点接收的提示词Prompt和模型的完整回复。路由函数是如何做出决策的。你可以在这里直接调试和编辑提示词重新运行某个步骤观察输出变化而无需修改本地代码。这对于优化智能体逻辑至关重要。5. 常见问题与排查思路在开发过程中你可能会遇到以下典型问题问题现象可能原因排查思路与解决方案启动服务时报错ModuleNotFoundError依赖未安装或虚拟环境未激活。1. 确认已激活虚拟环境。2. 运行pip install -r requirements.txt。调用/chatAPI 返回“智能体处理失败”1. OpenAI API密钥无效或余额不足。2. 知识库文件路径错误或未构建。1. 检查.env文件中的OPENAI_API_KEY。2. 运行knowledge_base.py中的构建脚本确保chroma_db_medical目录存在且非空。LangSmith 网站上看不到追踪记录1. LangSmith API密钥错误。2. 环境变量未生效。3. 项目名不匹配。1. 检查.env中的LANGSMITH_API_KEY。2. 重启服务确保环境变量加载。3. 确认LangSmith网站上创建的项目名与.env中LANGSMITH_PROJECT一致。智能体回复质量不高检索不到相关内容1. 知识库文档质量差或分割不合理。2. 检索数量k设置不当。3. 提示词设计不佳。1. 优化原始知识文档确保信息准确、结构化。2. 调整knowledge_base.py中的chunk_size和检索的k值。3. 利用LangSmith的追踪功能查看检索到的具体内容并优化提示词。图Graph的逻辑不符合预期路由错误节点函数对state的修改有误或路由函数route_after_analysis的逻辑判断错误。1. 在节点函数中增加详细打印检查state的中间值。2. 在LangSmith中查看generate节点的输出确认next_action字段是否正确生成。对话历史没有累积GraphState中conversation_history字段未正确定义为累加式。确保其类型为Annotated[List[str], operator.add]这是LangGraph实现状态累积的关键。6. 最佳实践与工程建议基于本项目我们可以提炼出一些通用的AI Agent开发最佳实践1. 状态设计要精简而充分精简状态字段只保留工作流必需的数据。过多的字段会使图变得难以理解和调试。充分确保状态包含所有控制逻辑需要的变量如stage,next_action和需要累积的信息如conversation_history。2. 节点功能保持单一职责每个节点应只做一件事并且做好。例如retrieve节点只负责检索generate节点只负责调用LLM并解析结果。这提高了代码的可测试性和可复用性。3. 充分利用LangSmith进行迭代提示词工程直接在LangSmith界面上修改提示词并重新运行Playground快速验证效果再将最优提示词更新到代码中。性能分析通过追踪记录分析耗时瓶颈是检索慢还是LLM调用慢针对性优化。评估与测试利用LangSmith的测试数据集Dataset和评估功能对智能体进行批量测试和量化评估。4. 生产环境考量错误处理与降级如图中generate节点对JSON解析失败做了降级处理。生产环境中需要对网络超时、API限流、无效输入等有更完备的异常处理。会话管理本例中简单的session_id并未实现真正的多轮对话状态持久化。生产环境需要将会话状态即GraphState存储到数据库如Redis并在每次请求时加载。异步优化对于IO密集型操作如检索、LLM调用可以考虑使用LangChain的异步接口或asyncio提升并发性能。安全与合规对于医疗等敏感领域必须在回复中添加免责声明并设置严格的输入过滤和输出审查机制避免产生医疗建议。5. 扩展方向更复杂的图引入循环边实现真正的多轮交互。例如从ask_question节点不是直接到END而是可以返回到retrieve或generate节点基于用户的新回答继续流程。集成工具让Agent能调用外部API例如查询药品信息、预约挂号接口需严格授权和沙箱测试。多模态结合LangChain的多模态处理能力处理用户上传的图片如皮疹照片进行分析。通过这个从零开始的医疗问诊Agent项目我们完整走通了使用LangChain构建基础能力、LangGraph编排复杂逻辑、LangSmith监控优化这一标准开发流程。这套方法论可以迁移到客服、金融、教育等任何需要复杂决策和状态管理的AI应用场景。记住构建强大Agent的关键不在于堆砌最复杂的模型而在于设计清晰、可控、可调试的工作流。 30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度