🚀 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文件并填入以下内容：

langchain==0.1.0 langchain-community==0.0.10 langchain-openai==0.0.5 langgraph==0.0.26 langsmith==0.0.87 openai==1.6.1 chromadb==0.4.22 tiktoken==0.5.2 fastapi==0.104.1 uvicorn[standard]==0.24.0 pydantic==2.5.0 python-dotenv==1.0.0

通过pip安装：

pip install -r requirements.txt

关键服务与API：

1. OpenAI API：本项目使用gpt-3.5-turbo作为核心LLM。你需要准备一个有效的OpenAI API密钥。
2. LangSmith：前往 LangSmith官网 注册并创建API密钥。它提供免费额度，足够用于开发和调试。
3. 向量数据库：我们使用轻量级的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_KEY=sk-your-openai-key-here LANGSMITH_API_KEY=lsv2-your-langsmith-key-here LANGSMITH_PROJECT=medical-agent-demo # LangSmith中的项目名 LANGCHAIN_TRACING_V2=true LANGCHAIN_ENDPOINT=https://api.smith.langchain.com

然后，实现知识库模块agent/knowledge_base.py：

import 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(model="text-embedding-3-small") self.persist_directory = persist_directory self.vectorstore = None def build_from_file(self, file_path: str): """从文本文件构建知识库""" print("正在加载医学指南文档...") loader = TextLoader(file_path, encoding='utf-8') documents = loader.load() # 文本分割：将长文档切成适合检索的片段 text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, # 每个片段约1000字符 chunk_overlap=200, # 片段间重叠200字符，保持上下文 separators=["\n\n", "\n", "。", "；", "，", " ", ""] ) splits = text_splitter.split_documents(documents) print(f"文档已分割为 {len(splits)} 个片段。") # 创建向量存储并持久化 self.vectorstore = Chroma.from_documents( documents=splits, embedding=self.embeddings, persist_directory=self.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_directory=self.persist_directory, embedding_function=self.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, k=k) 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--- 片段 {i+1} ---") print(doc.page_content[:300] + "...") # 打印前300字符

关键点解析：

• 文档分割：RecursiveCharacterTextSplitter是LangChain提供的智能分割器，能较好地在句末和段落处切割，保持语义完整性。
• 向量化：OpenAIEmbeddings将文本转换为向量，Chroma存储这些向量并执行相似性搜索。
• 检索器（Retriever）：as_retriever()方法返回的对象是LangChain生态中的标准接口，可以无缝接入后续的链（Chain）中。

3.2 第二步：定义智能体的状态（LangGraph State）

LangGraph的核心是有状态图。状态（State）是一个贯穿整个工作流、在各节点间传递和修改的数据结构。我们为医疗问诊Agent定义其状态。

创建agent/graph_state.py：

from 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.py：

from 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(model="gpt-3.5-turbo", temperature=0.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.py：

from 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(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI(title="医疗问诊智能体 API", version="0.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_model=QueryResponse) 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( response=final_state["llm_response"], stage=final_state["stage"], session_id=request.session_id ) except Exception as e: logger.error(f"处理会话 [{request.session_id}] 时出错: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"智能体处理失败: {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, host="0.0.0.0", port=8000)

4. 运行、验证与LangSmith监控

4.1 启动服务并测试

1. 确保你的.env文件已正确配置API密钥。
2. 在终端中运行：

   python main.py

3. 服务启动后，打开浏览器访问http://127.0.0.1:8000/docs，你会看到自动生成的Swagger UI界面。
4. 在/chat端点下，点击“Try it out”，输入JSON请求体，例如：

   { "user_input": "我喉咙痛，有点发烧，是感冒吗？", "session_id": "test_user_1" }

5. 点击“Execute”，观察响应。同时，在运行服务的终端里，你会看到我们打印的节点执行日志。

4.2 使用LangSmith进行追踪和调试

这是体现LangSmith价值的时刻。因为我们在.env中设置了LANGCHAIN_TRACING_V2=true，所以所有通过LangChain/LangGraph的调用都会被自动记录。

1. 登录 LangSmith 网站。
2. 在左侧边栏选择你设置的项目（medical-agent-demo）。
3. 在“Traces”页面，你将看到每次API调用产生的追踪记录。
4. 点击任意一次追踪，你可以看到完整的调用链：
   • 每个节点的输入（Input）和输出（Output）。
   • 每个步骤的耗时和Token使用情况。
   • retrieve节点检索到了哪些文档片段。
   • generate节点接收的提示词（Prompt）和模型的完整回复。
   • 路由函数是如何做出决策的。
5. 你可以在这里直接调试和编辑提示词，重新运行某个步骤，观察输出变化，而无需修改本地代码。这对于优化智能体逻辑至关重要。

5. 常见问题与排查思路

在开发过程中，你可能会遇到以下典型问题：

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 折。 👉 点击领海量免费额度