30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度这次我们来看一个医疗问诊 Agent 的构建过程它不是一个单一的模型而是一个由 LangChain、LangGraph 和 LangSmith 三大组件协同工作的智能系统。这个项目清晰地展示了如何将一个复杂的现实任务医疗问诊拆解、编排并监控最终形成一个可运行的 AI 应用。对于想入门 AI Agent 开发特别是想理解 LangChain 生态如何落地的开发者来说这是一个绝佳的案例。这个项目的核心价值在于它用“医疗问诊”这个具体场景把 LangChain工具链、LangGraph流程编排和 LangSmith监控调试这三个抽象概念串联了起来。你不需要先啃完所有官方文档而是通过一个完整的项目直观地看到每个组件扮演什么角色、如何协作。本文将带你从零开始理解这个 Agent 的架构并动手搭建一个简化的原型验证其核心流程。本文适合对 AI 应用开发感兴趣听说过 LangChain 但不知如何入手的开发者。我们将重点关注1如何用 LangChain 构建“医生工具箱”2如何用 LangGraph 绘制“就诊流程图”3如何用 LangSmith 监控整个问诊过程。读完本文你将能清晰地理解这三个工具的区别与联系并具备搭建一个基础 Agent 的能力。1. 核心能力速览能力项说明项目类型AI Agent 应用开发框架与工具链实践核心组件LangChain (工具与记忆)、LangGraph (流程编排)、LangSmith (监控与评估)主要功能模拟多轮医疗问诊流程包括症状收集、分诊、专科问答、诊断建议生成等技术栈Python, OpenAI API (或其他 LLM), LangChain 生态硬件门槛无特殊要求主要依赖云 LLM API 或本地部署的大语言模型。本地运行仅需普通开发机。启动方式通过 Python 脚本启动或集成到 Web 服务如 FastAPI中提供 API。是否支持 API是可轻松封装为 RESTful API 服务。是否支持批量任务是可通过异步或队列处理多个并行的问诊会话。适合场景学习 LangChain 生态、构建复杂多轮对话 Agent、需要可视化调试 AI 应用、教育演示、智能客服/问诊原型开发。2. 适用场景与使用边界这个医疗问诊 Agent 项目主要适用于以下场景AI Agent 学习与教学它是理解 LangChain、LangGraph、LangSmith 三大件协同工作的最佳实践案例。复杂流程自动化原型对于需要多步骤决策、状态管理和工具调用的业务逻辑如客服、审批、诊断本项目提供了可参考的架构。AI 应用可观测性实践通过 LangSmith开发者可以学习如何跟踪、调试和评估一个非确定性的 LLM 应用。使用边界与重要提醒非真实医疗诊断工具本项目仅为技术演示其生成的任何医疗建议都不可作为真实的诊断依据。AI 模型存在幻觉和知识截止问题绝对不能在真实医疗环境中使用。依赖外部 LLM核心智能依赖于 OpenAI GPT 或类似的大语言模型需要相应的 API 密钥或本地模型部署能力。数据隐私与合规如果处理真实用户问诊数据必须严格遵守相关数据隐私法规如 HIPAA, GDPR确保数据加密、匿名化并获得用户明确授权。本项目示例数据应为虚构。知识库局限性Agent 的医学知识完全来源于其连接的 LLM 和工具如医学数据库查询其准确性和时效性需要持续维护和验证。3. 环境准备与前置条件在开始构建之前请确保你的开发环境满足以下要求操作系统Windows 10/11, macOS, 或 Linux (如 Ubuntu 20.04)。Python 版本推荐 Python 3.10 或 3.11。避免使用 Python 3.12 等过新版本以防某些库兼容性问题。包管理工具使用pip或conda进行依赖管理。推荐创建独立的虚拟环境。核心依赖langchainlangchain-community: 提供基础链、工具、记忆等组件。langgraph: 用于构建有状态、多参与者的 Agent 工作流。langsmith(可选但强烈推荐): 用于跟踪、调试和评估 Agent 运行。大语言模型 SDK: 如openai(用于 GPT 系列) 或ollama(用于本地模型)。LLM 访问权限你需要一个可用的 LLM。最简单的方式是注册 OpenAI 并获取 API Key。如果你想本地运行需要部署如 Llama 3、Qwen 等开源模型并通过ollama或vllm等方式提供 API。LangSmith 账户 (可选)为了获得完整的可观测性体验建议注册 LangSmith 账户并获取 API Key。它提供免费的额度供学习和测试。4. 安装部署与启动方式我们将通过一个简化的步骤来搭建这个医疗问诊 Agent 的核心骨架。首先创建项目目录并初始化虚拟环境# 创建项目目录 mkdir medical-agent-demo cd medical-agent-demo # 创建虚拟环境 (以 venv 为例) python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/macOS: source venv/bin/activate接下来安装核心依赖pip install langchain langgraph langchain-openai langsmith # 如果你使用其他模型提供商安装对应的 langchain-xxx 包例如 # pip install langchain-community langchain-ollama然后创建一个名为medical_agent.py的主文件开始构建 Agent。我们先从最核心的 LangChain 部分开始。5. 功能测试与效果验证我们将分步构建并验证 Agent 的各个组件。整个问诊流程可以抽象为患者描述症状 - 分诊 Agent 判断科室 - 专科 Agent 深入问询 - 生成初步建议。5.1 构建“医生工具箱” (LangChain)LangChain 在这里的作用是封装各种“能力”比如与 LLM 对话、查询知识库、记录历史等。首先设置环境变量在实际项目中应使用.env文件管理import os from langchain_openai import ChatOpenAI # 设置你的 OpenAI API Key os.environ[OPENAI_API_KEY] your-openai-api-key-here # 设置 LangSmith 用于追踪 (可选) os.environ[LANGCHAIN_TRACING_V2] true os.environ[LANGCHAIN_ENDPOINT] https://api.smith.langchain.com os.environ[LANGCHAIN_API_KEY] your-langsmith-api-key-here os.environ[LANGCHAIN_PROJECT] Medical-Agent-Demo # 初始化 LLM llm ChatOpenAI(modelgpt-4o-mini, temperature0.1) # 使用一个较小、较稳定的模型接着我们定义几个关键的“工具”。在真实场景中这些工具可能是查询医学数据库、计算 BMI、获取药品信息等。这里我们用简单的模拟函数代替from langchain.tools import tool from typing import Optional tool def search_medical_knowledge_base(query: str) - str: 根据症状或疾病名称查询医学知识库。这是一个模拟函数。 # 模拟返回一些知识 knowledge_map { 头痛: 常见原因包括紧张性头痛、偏头痛、感冒或流感、高血压等。需关注是否伴有发热、呕吐、视力变化。, 发烧: 体温超过37.5°C。可能由感染病毒/细菌、炎症、中暑等引起。儿童高热需警惕热性惊厥。, 咳嗽: 可分为干咳和湿咳有痰。常见于呼吸道感染、过敏、哮喘等。长期咳嗽需排查慢性病。 } return knowledge_map.get(query.lower(), f已收到关于{query}的查询建议结合具体症状和病史进一步分析。) tool def calculate_bmi(weight_kg: float, height_m: float) - dict: 计算身体质量指数 (BMI)。 bmi weight_kg / (height_m ** 2) category 体重过轻 if bmi 18.5 else 正常范围 if bmi 25 else 超重 if bmi 30 else 肥胖 return {bmi: round(bmi, 2), category: category} # 将工具打包 tools [search_medical_knowledge_base, calculate_bmi]现在我们可以创建一个基础的“全科医生”Agent它能够使用这些工具from langchain.agents import create_react_agent, AgentExecutor from langchain import hub # 从 LangChain Hub 拉取一个标准的 ReAct 提示词模板 prompt hub.pull(hwchase17/react) # 创建 Agent agent create_react_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) # 测试一下这个基础 Agent result agent_executor.invoke({input: 我头痛还有点发烧体温大概38度这是什么情况}) print(result[output])运行这段代码你会看到 Agent 逐步思考verboseTrue可能会调用search_medical_knowledge_base工具并最终给出一个综合的回答。这验证了 LangChain 在组织工具和 LLM 协作方面的能力。5.2 绘制“就诊流程图” (LangGraph)单一的 Agent 无法处理复杂的多步骤、多角色问诊。LangGraph 允许我们定义一个有状态的图Graph其中节点是函数或 Agent边是控制流。我们来构建一个简化的两层问诊图分诊节点 (Triage): 根据用户主诉判断需要哪个“专科”处理或直接给出简单建议。专科问诊节点 (Specialist): 接收分诊结果进行更深入的、可能多轮的问答。首先定义图的状态。状态是一个字典随着流程推进不断更新。from typing import TypedDict, Annotated, Union from langgraph.graph import StateGraph, END import operator # 定义图的状态结构 class AgentState(TypedDict): messages: Annotated[list, operator.add] # 对话消息历史 department: Optional[str] # 分诊决定的科室 needs_specialist: bool # 是否需要专科处理 final_advice: Optional[str] # 最终建议 # 初始化图构建器 graph_builder StateGraph(AgentState)然后添加节点。每个节点是一个函数接收当前状态返回更新后的状态。def triage_node(state: AgentState) - dict: 分诊节点判断病情紧急程度和科室。 from langchain_core.messages import HumanMessage, SystemMessage # 构建系统提示词 system_prompt 你是一个经验丰富的分诊护士。请根据患者的描述完成以下任务 1. 判断病情的紧急程度紧急/非紧急。 2. 判断可能涉及的科室如内科、外科、儿科、耳鼻喉科等。如果问题非常轻微或属于常识健康咨询可以标记为“general”全科。 3. 决定是否需要转给专科医生进行深入问诊。 请以 JSON 格式回答包含字段urgency, department, needs_specialist (true/false)。 # 获取最新的用户消息 user_input state[messages][-1].content if state[messages] else 患者未描述症状。 messages [ SystemMessage(contentsystem_prompt), HumanMessage(contentf患者主诉{user_input}) ] # 调用 LLM 进行分析 response llm.invoke(messages) # 这里简化处理实际应解析 JSON import json try: triage_result json.loads(response.content) except: # 如果解析失败使用默认值 triage_result {urgency: non-urgent, department: general, needs_specialist: False} # 更新状态 new_state { department: triage_result.get(department), needs_specialist: triage_result.get(needs_specialist, False) } # 如果不需要专科直接生成初步建议 if not new_state[needs_specialist]: advice_prompt f基于以下分诊结果给患者一个简单、安全的初步健康建议。分诊结果{triage_result}。患者主诉{user_input} advice_msg llm.invoke([HumanMessage(contentadvice_prompt)]) new_state[final_advice] advice_msg.content return new_state def specialist_node(state: AgentState) - dict: 专科问诊节点进行深入交互。 # 这里可以集成前面创建的 agent_executor或者一个更复杂的专科 Agent # 例如一个专门处理“头痛”的 Agent会询问疼痛性质、部位、持续时间等。 # 为简化演示我们模拟一个询问过程。 department state.get(department, general) # 模拟专科问题 specialist_questions { internal: 为了进一步明确您的头痛原因请问是胀痛、刺痛还是搏动性疼痛持续多久了, pediatrics: 孩子多大了发烧时精神状态和食欲怎么样有没有出疹子, general: 您的症状出现多久了有没有自行服用过什么药物 } question specialist_questions.get(department, 能否再详细描述一下您的症状) # 在实际中这里应该等待用户的下一次输入并更新消息历史。 # 本例中我们模拟用户回答并生成最终建议。 simulated_answer 是搏动性的疼大概一天了。 # 模拟生成专科建议 advice_prompt f你是一名{department}科医生。患者主诉{state[messages][-1].content}。你的追问{question}。患者回答{simulated_answer}。请给出专业的初步分析和行动建议如休息、观察、非处方药或尽快就医。 final_advice llm.invoke([HumanMessage(contentadvice_prompt)]).content return {final_advice: final_advice} # 将节点添加到图中 graph_builder.add_node(triage, triage_node) graph_builder.add_node(specialist, specialist_node)接下来定义边路由逻辑控制流程的走向。# 设置入口点 graph_builder.set_entry_point(triage) # 定义路由函数根据分诊结果决定下一步 def route_after_triage(state: AgentState) - str: if state.get(needs_specialist): return specialist # 需要专科跳转到专科节点 else: return END # 不需要专科直接结束 # 添加条件边 graph_builder.add_conditional_edges( triage, route_after_triage, { specialist: specialist, END: END } ) # 专科节点结束后流程终止 graph_builder.add_edge(specialist, END) # 编译图 graph graph_builder.compile()现在我们可以运行这个图来模拟一次问诊# 初始化状态 initial_state: AgentState { messages: [HumanMessage(content我从今天早上开始头痛感觉有点发烧。)], department: None, needs_specialist: None, final_advice: None } # 执行图 final_state graph.invoke(initial_state) print(\n 分诊科室 ) print(final_state.get(department)) print(\n 最终建议 ) print(final_state.get(final_advice))运行后你将看到流程从triage节点开始根据 LLM 的判断可能跳转到specialist节点最终生成建议。这验证了 LangGraph 在编排复杂、有条件分支的工作流上的能力。5.3 监控“问诊全过程” (LangSmith)LangSmith 为上述所有步骤提供了可观测性。如果你正确设置了环境变量上述agent_executor.invoke()和graph.invoke()的调用都会被自动记录到 LangSmith 平台。登录 LangSmith访问 https://smith.langchain.com 并登录。查看轨迹 (Traces)在项目 (Medical-Agent-Demo) 下你会看到每次调用的记录。点击进入一条 Trace你可以看到完整的输入和输出。每一步的耗时和 Token 使用量。Agent 的思考过程 (ReAct)包括工具调用的输入输出。LangGraph 图的执行路径清晰展示走了triage-specialist还是triage-END。调试与优化你可以检查 LLM 的输入提示词是否合理工具返回的结果是否被正确利用以及图的路由逻辑是否符合预期。这是开发复杂 Agent 时不可或缺的“调试器”它能帮你快速定位是哪个环节工具、LLM、路由导致了不符合预期的输出。6. 接口 API 与批量任务要将这个 Agent 投入实用需要将其封装成服务并支持批量或并发处理。6.1 封装为 FastAPI 服务我们可以创建一个简单的 HTTP API接收用户问诊请求返回 Agent 的处理结果。# 文件app.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List from medical_agent import graph, AgentState # 假设之前的代码在 medical_agent.py 中 from langchain_core.messages import HumanMessage app FastAPI(titleMedical Consultation Agent API) class ConsultationRequest(BaseModel): user_input: str session_id: str None # 用于支持多轮对话的会话ID class ConsultationResponse(BaseModel): session_id: str department: str None final_advice: str trace_url: str None # LangSmith 跟踪链接 app.post(/consult, response_modelConsultationResponse) async def consult(request: ConsultationRequest): try: # 准备初始状态 initial_state: AgentState { messages: [HumanMessage(contentrequest.user_input)], department: None, needs_specialist: None, final_advice: None } # 执行问诊图 final_state graph.invoke(initial_state) # 构建响应 # 注意实际中需要从 LangSmith SDK 获取本次运行的 trace URL response ConsultationResponse( session_idrequest.session_id or new_session, departmentfinal_state.get(department), final_advicefinal_state.get(final_advice, 未能生成建议。), trace_urlNone # 此处可填入从 langsmith 客户端获取的 URL ) return response except Exception as e: raise HTTPException(status_code500, detailfAgent execution failed: {str(e)}) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)使用uvicorn app:app --reload启动服务后就可以通过POST /consult接口进行问诊。6.2 批量任务处理对于批量处理问诊记录例如处理一批离线用户反馈可以使用异步任务队列如 Celery Redis或简单的线程池。# 文件batch_processor.py import asyncio import aiohttp import pandas as pd from typing import List async def process_single_consultation(session: aiohttp.ClientSession, user_input: str, idx: int): 异步处理单个问诊请求。 api_url http://localhost:8000/consult payload {user_input: user_input, session_id: fbatch_{idx}} try: async with session.post(api_url, jsonpayload) as resp: result await resp.json() return result except Exception as e: return {error: str(e), input: user_input} async def batch_process(inputs: List[str]): 批量处理问诊列表。 async with aiohttp.ClientSession() as session: tasks [process_single_consultation(session, inp, i) for i, inp in enumerate(inputs)] results await asyncio.gather(*tasks, return_exceptionsTrue) return results # 示例从CSV文件读取批量问诊 df pd.read_csv(batch_consultations.csv) user_inputs df[symptom_description].tolist() # 运行批量处理 loop asyncio.get_event_loop() batch_results loop.run_until_complete(batch_process(user_inputs[:10])) # 先测试10条 for res in batch_results: print(res)7. 资源占用与性能观察由于本项目的核心计算负载在于大语言模型LLM的推理因此资源占用主要取决于你选择的 LLM 服务方式使用云 API (如 OpenAI)本地资源占用极低CPU、内存和显存占用主要来自轻量级的 Web 框架如 FastAPI和 LangChain/LangGraph 框架本身通常可以忽略不计。性能瓶颈在网络和 API 限流响应时间取决于网络延迟和云 API 的响应速度。需要处理 API 调用失败、重试和速率限制。成本可控按 Token 用量计费易于估算。本地部署开源 LLM显存是主要瓶颈例如运行一个 7B 参数的模型进行 4-bit 量化推理可能需要 6-8 GB 显存。模型越大显存需求越高。CPU/内存要求如果使用 CPU 推理需要足够的内存通常是模型大小的 1.5-2 倍和较强的 CPU。启动方式通常通过ollama run或启动vllm/text-generation-webui的 API 服务。性能观察使用nvidia-smi(GPU) 或任务管理器监控显存/内存占用。关注每个请求的首次 Token 生成时间 (TTFT) 和吞吐量。LangGraph 性能提示图编译graph_builder.compile()是一次性开销。状态序列化如果状态很大例如包含长对话历史可能会影响性能。考虑使用外部存储如 Redis管理对话状态。并发LangGraph 本身是框架并发能力取决于你如何部署它如多进程 FastAPI。确保你的 LLM 后端本地或云端能支持预期的并发量。8. 常见问题与排查方法问题现象可能原因排查方式解决方案导入 LangChain 相关模块失败1. 未安装对应包。2. Python 版本不兼容。3. 虚拟环境未激活。1. 运行pip list | grep langchain检查。2. 确认 Python 版本为 3.10/3.11。3. 检查命令行前缀是否有(venv)。1. 使用pip install langchain langgraph ...安装。2. 使用conda或pyenv管理版本。3. 重新激活虚拟环境。运行时报错OPENAI_API_KEY未设置环境变量未正确设置。1. 在代码中print(os.environ.get(“OPENAI_API_KEY”))检查。2. 检查.env文件是否加载。1. 在代码开头直接设置os.environ[“OPENAI_API_KEY”] “sk-...”。2. 使用python-dotenv加载.env文件。Agent 调用工具时解析错误1. LLM 返回格式不符合工具调用规范。2. 工具函数定义参数、文档字符串不清晰。1. 在 LangSmith 中查看 LLM 的原始输出。2. 检查工具函数的tool装饰器和文档字符串。1. 优化提示词明确要求 LLM 以指定格式如 JSON输出。2. 使用handle_parsing_errorsTrue参数让 AgentExecutor 更健壮。3. 简化工具定义。LangGraph 图执行后未得到预期结果1. 路由逻辑 (conditional_edges) 有误。2. 节点函数修改状态的方式不对。3. LLM 在分诊节点判断不准。1. 在 LangSmith 中查看图的执行轨迹检查每一步的输入输出和状态变化。2. 在路由函数中增加print调试。3. 检查分诊节点的系统提示词是否清晰。1. 仔细检查add_conditional_edges的映射字典。2. 确保节点函数返回的字典键名与State定义匹配。3. 优化分诊提示词加入少量示例Few-shot。LangSmith 上看不到追踪记录1. 环境变量未设置或错误。2. 项目名称 (LANGCHAIN_PROJECT) 不存在。3. 网络问题。1. 确认所有LANGCHAIN_*环境变量已设置。2. 登录 LangSmith 网站查看项目列表。3. 尝试在代码中初始化Client并手动发送追踪。1. 确保环境变量在代码运行前已设置。2. 在 LangSmith 上创建同名项目。3. 检查代理或防火墙设置。FastAPI 服务调用超时或失败1. Agent 内部处理时间过长。2. LLM API 调用失败。3. 服务未启动或端口冲突。1. 查看服务日志。2. 单独测试 Agent 核心函数耗时。3. 使用curl或httpie测试 API 端点。1. 为 API 端点设置合理的超时时间。2. 在 Agent 调用层添加重试机制。3. 使用uvicorn app:app --host 0.0.0.0 --port 8000启动并确认无报错。批量处理时大量失败1. API 速率限制。2. 网络不稳定。3. 输入数据格式异常。1. 查看失败请求的返回信息。2. 监控 LLM 服务提供商的控制台。3. 对输入数据做预处理和清洗。1. 在批量任务中添加指数退避重试。2. 限制并发请求数。3. 增加输入验证和错误处理。9. 最佳实践与使用建议从简单开始逐步复杂化不要一开始就设计庞大的图。先实现一个能跑通的单节点 Agent再逐步添加工具、路由和新的节点。充分利用 LangSmith在开发阶段始终开启 LangSmith 追踪。它是你理解 Agent 内部决策过程、调试提示词和优化流程的最强工具。为关键节点如分诊、工具调用添加自定义标签或元数据便于筛选和分析。提示词工程是关键Agent 的智能程度很大程度上取决于提示词。为每个节点分诊、专科设计清晰、具体、带有角色和任务的系统提示词。使用Few-shot提供示例可以显著提高输出的稳定性和准确性。状态设计要精简LangGraph 的State会在每个节点间传递。只存储必要的信息避免状态过大影响性能和序列化。对于长对话历史考虑使用外部存储如数据库在状态中只保留会话 ID。错误处理与健壮性在工具调用、LLM 响应解析、路由判断等环节都要加入try...except。使用handle_parsing_errors等参数让 Agent 更稳定。为 API 服务设计友好的错误响应格式。测试与评估除了功能测试还要建立评估体系。可以定义一些测试用例输入-期望输出对利用 LangSmith 的评估功能自动运行并评分持续监控 Agent 的质量变化。安全与合规前置如前所述医疗、金融、法律等领域的 Agent 必须严格考虑合规性。所有输出都应包含免责声明。对用户输入进行过滤防止 Prompt 注入攻击。10. 总结与下一步通过这个医疗问诊 Agent 的构建之旅我们清晰地看到了 LangChain、LangGraph 和 LangSmith 如何各司其职LangChain提供了构建 Agent 所需的“砖块”LLM、提示词模板、工具、记忆。LangGraph提供了将这些“砖块”组装成复杂“机器”的蓝图有状态、可循环、多参与者的工作流。LangSmith提供了观察这台“机器”如何运行的仪表盘追踪、调试、评估。最值得尝试的点亲手运行一遍从单一 Agent 到多节点工作流的代码并在 LangSmith 上观察完整的执行轨迹。这种直观的感受比阅读文档有效得多。最先应该验证的功能确保你的基础环境API Key、依赖包没问题然后跑通一个最简单的AgentExecutor调用工具的例子。这是所有复杂应用的基石。最容易踩的坑提示词设计不佳导致路由混乱或工具调用失败状态管理不当导致信息丢失忽略了错误处理使得服务不稳定。后续扩展方向集成真实工具将模拟的search_medical_knowledge_base替换为连接真实医学数据库需授权或权威医学网站 API 的工具。实现多轮对话改造 LangGraph 的状态和路由使其能根据专科 Agent 的追问循环等待用户输入实现真正的交互式问诊。加入记忆机制使用 LangChain 的ConversationBufferMemory或ConversationSummaryMemory让 Agent 记住整个会话历史。构建前端界面使用 Gradio 或 Streamlit 快速搭建一个 Web 界面提供更友好的交互体验。探索更复杂的图结构例如引入“检查检验建议”节点、“用药咨询”节点甚至“紧急情况直连人工”节点让流程更贴近现实。这个项目就像一个功能完备的“样板间”展示了用 LangChain 生态构建复杂 AI 应用的完整方法论。你可以基于此将其改造成客服、教育、游戏、自动化办公等任何领域的智能助手。 30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度
)
