30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度这次我们来看一个医疗问诊 Agent 的完整构建过程。这个项目不是简单的概念介绍而是通过一个具体的应用场景帮你彻底理清 LangChain、LangGraph 和 LangSmith 这三个核心框架在实际开发中各自扮演什么角色、如何协同工作以及如何从零开始搭建一个能稳定运行的智能体。如果你正在学习 AI Agent 开发或者想了解如何将大模型能力整合进业务流程这篇文章可以直接收藏。我们将重点关注如何用 LangChain 组织工具链如何用 LangGraph 编排复杂的决策流程以及如何用 LangSmith 来监控、调试和优化整个 Agent 的运行。整个过程不涉及复杂的硬件部署核心是代码逻辑和架构设计。1. 核心能力速览能力项说明项目类型AI Agent 应用开发具体为医疗问诊场景核心框架LangChain工具与记忆、LangGraph流程编排、LangSmith可观测性主要功能模拟医生问诊流程进行多轮对话、症状收集、初步分析与建议技术门槛中等需熟悉 Python 及大模型基础 API 调用无需特定 GPU启动方式本地 Python 脚本启动或封装为 FastAPI 等 Web 服务是否支持 API是核心 Agent 可作为服务被调用是否支持“批量任务”是可通过异步或队列处理多个并行的问诊会话适合场景学习 Agent 开发、构建专业领域对话系统、理解 LangChain 生态2. 适用场景与使用边界这个医疗问诊 Agent 项目主要适合以下几类读者和场景AI 应用开发者/学习者想通过一个完整案例系统性掌握 LangChain、LangGraph、LangSmith 的使用方法和集成技巧。垂直领域对话系统探索者医疗问诊是一个典型的、流程清晰的领域其设计模式可以迁移到法律咨询、金融客服、技术支持等场景。希望提升 Agent 稳定性的工程师LangSmith 提供的可观测性工具是解决 Agent 运行时“黑盒”问题、调试复杂逻辑的关键。需要明确的使用边界非真实医疗诊断工具本项目仅为技术演示其生成的任何症状分析、健康建议都不能替代专业医生的诊断。严禁将其用于真实的医疗诊断场景。知识依赖大模型Agent 的医学知识完全来源于所连接的大语言模型如 GPT-4、Claude、Qwen 等以及可能集成的外部知识库RAG。其准确性和可靠性受限于模型本身的知识截止日期与准确性。合规与隐私如果处理真实的用户问诊数据必须严格遵守数据隐私法规如 HIPAA、GDPR 等对用户个人信息进行脱敏处理并确保数据传输和存储的安全。场景局限性当前演示的流程相对标准化对于极其复杂、危急或需要多模态检查如影像分析的医疗情况此 Agent 架构需要大幅扩展和接入专业系统。3. 环境准备与前置条件在开始编码之前你需要准备好以下环境。我们的开发将主要在 CPU 环境下进行因为核心是逻辑编排而非模型训练。基础环境清单操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04 推荐)。本文示例基于 Linux/macOS 命令行Windows 用户可使用 WSL 或 Git Bash。Python 版本Python 3.10 或 3.11。这是 LangChain 等库兼容性较好的版本。包管理工具pip或conda。代码编辑器VS Code, PyCharm 等任选。大模型 API 密钥你需要一个可用的 LLM API 服务。我们将以 OpenAI GPT 系列或兼容 OpenAI API 的本地模型为例你需要准备相应的API_KEY和BASE_URL如果使用非官方服务。关键依赖库我们将通过一个requirements.txt文件来管理依赖。核心库包括langchainlangchain-community: 提供基础组件LLM、工具、记忆等。langgraph用于构建有状态的、多步骤的工作流图。langsmith用于跟踪、评估和监控 Agent 运行。openai用于调用 OpenAI 或兼容 API。python-dotenv管理环境变量。4. 安装部署与启动方式首先创建项目目录并安装依赖。# 1. 创建项目目录并进入 mkdir medical-agent-demo cd medical-agent-demo # 2. 创建虚拟环境 (可选但推荐) python -m venv venv # Windows: venv\Scripts\activate # Linux/macOS: source venv/bin/activate # 3. 创建 requirements.txt 文件 cat requirements.txt EOF langchain0.1.0 langchain-community0.0.10 langgraph0.0.20 langsmith0.1.0 openai1.0.0 python-dotenv1.0.0 EOF # 4. 安装依赖 pip install -r requirements.txt接下来设置环境变量。创建一个.env文件来安全存储你的 API 密钥和 LangSmith 配置。# 创建 .env 文件 cat .env EOF # 你的 OpenAI 兼容 API 配置 OPENAI_API_KEYsk-your-openai-api-key-here OPENAI_BASE_URLhttps://api.openai.com/v1 # 若使用其他服务请替换 # LangSmith 配置用于追踪和调试注册后获取 LANGCHAIN_TRACING_V2true LANGCHAIN_ENDPOINThttps://api.smith.langchain.com LANGCHAIN_API_KEYls-your-langsmith-api-key-here LANGCHAIN_PROJECTMedical-Agent-Demo # 在 LangSmith 上看到的项目名 EOF重要你需要去 LangSmith 官网 注册并获取LANGCHAIN_API_KEY。虽然不配置 LangSmith 也能运行 Agent但你将无法获得强大的调试和追踪能力。至此基础环境就搭建完成了。我们的“启动”就是运行一个 Python 脚本。下面进入核心部分构建 Agent。5. 功能测试与效果验证构建医疗问诊 Agent我们将分步骤构建 Agent并在每一步进行验证。最终目标是形成一个可以执行“症状收集 - 初步分析 - 给出建议”流程的智能体。5.1 第一步定义工具 - 医生的“听诊器”在 LangChain 中工具Tools是 Agent 与外界交互的手段。对于问诊 Agent我们可以定义一些工具来获取结构化信息或执行特定操作。我们先创建一个tools.py文件# tools.py from langchain.tools import tool from typing import Optional tool def assess_symptom_severity(symptom: str, duration_hours: int) - str: 根据症状描述和持续时间小时初步评估严重程度。 severity_levels { (0, 24): f“症状‘{symptom}’持续时间较短{duration_hours}小时可能为急性、轻微问题建议观察休息。”, (24, 72): f“症状‘{symptom}’已持续{duration_hours}小时1-3天需保持关注如加重请就医。”, (72, 168): f“症状‘{symptom}’持续超过3天{duration_hours}小时建议安排时间就诊检查。”, (168, float(‘inf’)): f“症状‘{symptom}’为长期问题{duration_hours}小时建议进行专业诊断和管理。” } for (low, high), message in severity_levels.items(): if low duration_hours high: return message return “无法评估严重程度。” tool def query_medical_knowledge_base(keyword: str) - str: 模拟查询医学知识库。在实际应用中这里可以接入真实的 RAG 系统。 # 这里简化为一个模拟的字典查询 knowledge_snippets { “头痛”: “常见原因包括紧张性头痛、偏头痛、感冒、高血压等。需注意是否伴有发烧、呕吐或视力变化。”, “咳嗽”: “分为干咳和湿咳。可能由感冒、流感、支气管炎、过敏等引起。长期咳嗽需排查肺部问题。”, “发烧”: “体温超过37.5°C。是身体对抗感染的信号。需关注体温峰值、伴随症状及持续时间。”, “腹痛”: “部位和性质很重要如绞痛、胀痛。可能原因从消化不良到急腹症如阑尾炎不等。” } return knowledge_snippets.get(keyword, f“关于‘{keyword}’的详细医学信息建议咨询专业资料或医生。”) tool def recommend_next_action(assessed_conditions: list) - str: 根据已评估的症状列表综合推荐下一步行动。 if not assessed_conditions: return “未获取到有效症状进行评估。” recommendations [] for condition in assessed_conditions: if “建议观察休息” in condition: recommendations.append(“可暂时居家观察多喝水注意休息。”) elif “建议安排时间就诊” in condition: recommendations.append(“建议近日前往社区医院或门诊就诊。”) elif “建议进行专业诊断” in condition: recommendations.append(“建议尽快预约专科医生进行详细检查。”) if “尽快” in “ “.join(recommendations): return “综合来看情况可能需要及时医疗介入。建议” “; “.join(set(recommendations)) else: return “综合建议” “; “.join(set(recommendations)) if recommendations else “暂无明确建议。”验证工具在 Python 交互环境中测试工具是否正常工作。# 在 Python shell 中测试 from tools import assess_symptom_severity, query_medical_knowledge_base print(assess_symptom_severity.invoke({“symptom”: “头痛”, “duration_hours”: 48})) # 输出症状‘头痛’已持续48小时1-3天需保持关注如加重请就医。 print(query_medical_knowledge_base.invoke({“keyword”: “咳嗽”})) # 输出分为干咳和湿咳。可能由感冒、流感、支气管炎、过敏等引起。长期咳嗽需排查肺部问题。5.2 第二步创建 Agent 状态与 LangGraph 工作流这是核心部分。我们将使用 LangGraph 来定义问诊的流程。创建一个agent_graph.py文件。# agent_graph.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain.tools import Tool from langchain.agents import create_react_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder import os from dotenv import load_dotenv from tools import assess_symptom_severity, query_medical_knowledge_base, recommend_next_action load_dotenv() # 加载 .env 中的 API 密钥 # 1. 定义 Agent 的状态State class AgentState(TypedDict): “”“图的状态贯穿整个对话流程。”“” messages: Annotated[List, operator.add] # 对话消息历史 symptoms_collected: List[str] # 收集到的症状描述列表 assessments: List[str] # 对每个症状的评估结果列表 current_focus: str # 当前正在询问或处理的症状关键词 # 2. 初始化 LLM 和工具 llm ChatOpenAI( model“gpt-4o-mini”, # 或 “gpt-3.5-turbo”, “claude-3-haiku” 等 temperature0.2, # 较低的温度使输出更稳定 api_keyos.getenv(“OPENAI_API_KEY”), base_urlos.getenv(“OPENAI_BASE_URL”) ) # 将工具包装成 LangChain Tool 对象 tools [ Tool.from_function(assess_symptom_severity), Tool.from_function(query_medical_knowledge_base), Tool.from_function(recommend_next_action) ] # 3. 创建 ReAct Agent 作为我们的“推理大脑” prompt ChatPromptTemplate.from_messages([ (“system”, “你是一位耐心细致的AI医疗助手正在帮助用户进行症状初步问诊。请一步步思考必要时使用工具。每次只深入询问一个核心症状。当用户主动表示没有更多症状或你判断信息足够时就使用 recommend_next_action 工具给出综合建议并结束问诊。”), MessagesPlaceholder(variable_name“messages”), (“user”, “{input}”), MessagesPlaceholder(variable_name“agent_scratchpad”), ]) agent create_react_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) # 4. 定义图的各个节点Nodes def collect_symptoms_node(state: AgentState): “”“节点与用户交互收集症状信息。”“” # 这里简化处理实际应从 state[‘messages’] 中提取最新用户输入 # 我们假设最新一条用户消息是症状描述 if state[‘messages’]: last_message state[‘messages’][-1] if last_message.type “human”: symptom_text last_message.content state[‘symptoms_collected’].append(symptom_text) # 简单提取关键词实际可用更复杂的NLP state[‘current_focus’] symptom_text.split()[0] if symptom_text else “” return state def assess_symptom_node(state: AgentState): “”“节点调用工具评估当前关注的症状。”“” if state[‘current_focus’] and state[‘symptoms_collected’]: # 这里简化用最后一个症状和假设的持续时间进行评估 # 真实场景应从对话中提取持续时间 duration 24 # 假设持续24小时 assessment assess_symptom_severity.invoke({ “symptom”: state[‘current_focus’], “duration_hours”: duration }) state[‘assessments’].append(assessment) return state def query_knowledge_node(state: AgentState): “”“节点查询知识库获取更多背景信息。”“” if state[‘current_focus’]: knowledge query_medical_knowledge_base.invoke({“keyword”: state[‘current_focus’]}) # 可以将知识添加到消息历史中供 Agent 参考 state[‘messages’].append({“role”: “system”, “content”: f“医学知识参考{knowledge}”}) return state def recommend_action_node(state: AgentState): “”“节点调用工具生成综合建议。”“” if state[‘assessments’]: recommendation recommend_next_action.invoke({“assessed_conditions”: state[‘assessments’]}) state[‘messages’].append({“role”: “assistant”, “content”: recommendation}) return state def should_continue(state: AgentState) - str: “”“路由逻辑判断是继续询问症状还是给出建议结束。”“” # 简单的逻辑如果用户最近一次输入包含“结束”、“没了”、“没有更多了”等则结束。 last_user_msg “” for msg in reversed(state[‘messages’]): if msg.type “human”: last_user_msg msg.content.lower() break stop_keywords [“结束”, “没了”, “没有更多了”, “就这样”, “stop”] if any(keyword in last_user_msg for keyword in stop_keywords): return “recommend” # 否则继续收集症状在实际中这里可以加入更复杂的判断比如症状数量、关键症状是否收集全等 return “collect” # 5. 构建图Graph workflow StateGraph(AgentState) # 添加节点 workflow.add_node(“collect”, collect_symptoms_node) workflow.add_node(“assess”, assess_symptom_node) workflow.add_node(“query_knowledge”, query_knowledge_node) workflow.add_node(“recommend”, recommend_action_node) # 设置入口点 workflow.set_entry_point(“collect”) # 添加边Edges和路由 workflow.add_conditional_edges( “collect”, should_continue, # 路由函数 { “collect”: “assess”, # 继续 - 去评估症状 “recommend”: “recommend” # 结束 - 去生成建议 } ) workflow.add_edge(“assess”, “query_knowledge”) workflow.add_edge(“query_knowledge”, “collect”) # 查询知识后回到收集节点准备下一轮 workflow.add_edge(“recommend”, END) # 生成建议后结束图执行 # 编译图 app workflow.compile()验证图结构你可以检查图是否编译成功并打印其结构。# 在 Python shell 中测试 from agent_graph import app print(app.get_graph().draw_ascii()) # 这会在控制台输出一个简单的 ASCII 图显示节点和边的关系。5.3 第三步运行与测试 Agent创建一个run_agent.py脚本来启动一次完整的问诊会话。# run_agent.py from agent_graph import app, AgentState from langchain_core.messages import HumanMessage def run_medical_agent(): “”“运行医疗问诊 Agent 的示例。”“” # 1. 初始化状态 initial_state: AgentState { “messages”: [], “symptoms_collected”: [], “assessments”: [], “current_focus”: “” } print(“ AI 医疗问诊助手 (输入‘结束’以完成问诊) “) # 2. 多轮对话循环 while True: try: user_input input(“\n您有什么症状(或输入‘结束’): “) except KeyboardInterrupt: print(“\n问诊已中断。”) break if user_input.lower() in [“结束”, “exit”, “quit”]: # 触发结束流程 initial_state[‘messages’].append(HumanMessage(content“结束”)) final_state app.invoke(initial_state, config{“recursion_limit”: 50}) print(“\n 问诊总结 ) for msg in final_state[‘messages’]: if msg.type “assistant”: print(f“助手: {msg.content}”) break # 3. 将用户输入加入状态并执行图 initial_state[‘messages’].append(HumanMessage(contentuser_input)) print(f“\n[用户]: {user_input}”) # 执行图的一步或一个循环 # 注意这里我们调用整个图它会根据内部路由运行到下一个‘暂停点’通常是等待下一次用户输入前。 # 更精细的控制可以使用 app.stream。 new_state app.invoke(initial_state, config{“recursion_limit”: 50}) # 4. 显示 AI 的响应来自工具调用或 LLM 生成 # 查找最新的助理消息 for msg in reversed(new_state[‘messages’]): if msg.type “assistant”: print(f“[助手]: {msg.content}”) break # 更新状态准备下一轮 initial_state new_state if __name__ “__main__”: run_medical_agent()运行测试 在终端执行python run_agent.py你将进入一个交互式问诊会话。尝试输入“我有点头痛和咳嗽”然后根据提示继续。输入“结束”后Agent 会调用recommend_next_action工具给出综合建议。预期效果Agent 会识别你提到的症状如“头痛”。它会在后台调用assess_symptom_severity和query_medical_knowledge_base工具。在对话中它可能会根据知识库信息追问细节如“头痛是哪种痛法”这依赖于 LLM 根据提示词和工具信息生成的回复。当你表示结束后它会生成一份综合建议。6. 接口 API 与批量任务6.1 封装为 FastAPI 服务将 Agent 封装成 API 服务便于集成到其他系统。创建api_server.py。# api_server.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Optional from agent_graph import app, AgentState from langchain_core.messages import HumanMessage import asyncio from contextlib import asynccontextmanager # 定义请求/响应模型 class ChatRequest(BaseModel): message: str session_id: Optional[str] None # 用于支持多会话 class ChatResponse(BaseModel): reply: str session_state: dict # 返回当前会话的状态摘要 # 简单的内存会话存储生产环境应使用 Redis、数据库等 session_store {} asynccontextmanager async def lifespan(app: FastAPI): “”“启动和关闭生命周期管理。”“” # 启动时可以做些初始化 yield # 关闭时清理资源 session_store.clear() app FastAPI(lifespanlifespan) app.post(“/chat”, response_modelChatResponse) async def chat_endpoint(request: ChatRequest): “”“处理单轮问诊消息。”“” session_id request.session_id or “default_session” # 获取或初始化会话状态 if session_id not in session_store: session_store[session_id] { “messages”: [], “symptoms_collected”: [], “assessments”: [], “current_focus”: “” } state: AgentState session_store[session_id] # 将用户消息加入状态 state[‘messages’].append(HumanMessage(contentrequest.message)) try: # 执行 Agent 图 new_state await asyncio.to_thread( app.invoke, state, {“recursion_limit”: 50} ) except Exception as e: raise HTTPException(status_code500, detailf“Agent 执行失败: {str(e)}”) # 更新会话存储 session_store[session_id] new_state # 提取最新的助理回复 assistant_reply “” for msg in reversed(new_state[‘messages’]): if msg.type “assistant”: assistant_reply msg.content break if not assistant_reply: assistant_reply “正在思考...” # 返回响应 return ChatResponse( replyassistant_reply, session_state{ “symptoms_collected”: new_state[‘symptoms_collected’], “assessments”: new_state[‘assessments’], “message_count”: len(new_state[‘messages’]) } ) app.get(“/session/{session_id}”) async def get_session_state(session_id: str): “”“获取指定会话的当前状态。”“” if session_id not in session_store: raise HTTPException(status_code404, detail“Session not found”) return session_store[session_id] app.delete(“/session/{session_id}”) async def delete_session(session_id: str): “”“删除会话。”“” if session_id in session_store: del session_store[session_id] return {“message”: f“Session {session_id} deleted”} raise HTTPException(status_code404, detail“Session not found”) if __name__ “__main__”: import uvicorn uvicorn.run(app, host“0.0.0.0”, port8000)启动 API 服务pip install fastapi uvicorn # 如果尚未安装 python api_server.py服务将在http://127.0.0.1:8000启动。你可以使用curl或 Postman 进行测试。API 调用测试# 开始一个新问诊 curl -X POST “http://127.0.0.1:8000/chat \ -H “Content-Type: application/json” \ -d ‘{“message”: “我今天开始头痛”, “session_id”: “user_123”}’ # 继续同一会话 curl -X POST “http://127.0.0.1:8000/chat \ -H “Content-Type: application/json” \ -d ‘{“message”: “还有点咳嗽没有发烧”, “session_id”: “user_123”}’ # 结束问诊 curl -X POST “http://127.0.0.1:8000/chat \ -H “Content-Type: application/json” \ -d ‘{“message”: “结束”, “session_id”: “user_123”}’ # 获取会话状态 curl “http://127.0.0.1:8000/session/user_1236.2 批量任务处理对于批量处理例如分析一批离线描述的病例你可以编写一个脚本循环调用 Agent 或 API。# batch_process.py import asyncio import aiohttp import json from typing import List async def process_one_case(session: aiohttp.ClientSession, case_id: str, symptom_texts: List[str], api_url: str): “”“处理一个病例包含多条症状描述。”“” session_id f“batch_{case_id}” all_responses [] for i, text in enumerate(symptom_texts): payload {“message”: text, “session_id”: session_id} try: async with session.post(f“{api_url}/chat, jsonpayload) as resp: result await resp.json() all_responses.append({“turn”: i, “input”: text, “output”: result[“reply”]}) print(f“Case {case_id}, Turn {i}: {text[:50]}... - {result[‘reply’][:50]}...”) except Exception as e: print(f“Error processing case {case_id}, turn {i}: {e}”) all_responses.append({“turn”: i, “input”: text, “output”: f“ERROR: {e}”}) # 最后发送“结束”信号以获取总结 try: async with session.post(f“{api_url}/chat, json{“message”: “结束”, “session_id”: session_id}) as resp: final_result await resp.json() all_responses.append({“turn”: “final”, “input”: “结束”, “output”: final_result[“reply”]}) except Exception as e: print(f“Error getting final recommendation for case {case_id}: {e}”) return {“case_id”: case_id, “responses”: all_responses} async def batch_process_cases(case_list: List[dict], api_url: str “http://127.0.0.1:8000): “”“批量处理多个病例。”“” connector aiohttp.TCPConnector(limit10) # 控制并发数避免压垮服务 async with aiohttp.ClientSession(connectorconnector) as session: tasks [] for case in case_list: task process_one_case(session, case[“id”], case[“symptoms”], api_url) tasks.append(task) results await asyncio.gather(*tasks, return_exceptionsTrue) # 处理结果 successful [] failed [] for r in results: if isinstance(r, Exception): failed.append(str(r)) else: successful.append(r) print(f“批量处理完成。成功: {len(successful)}, 失败: {len(failed)}”) return successful, failed if __name__ “__main__”: # 模拟批量数据 mock_cases [ {“id”: “001”, “symptoms”: [“头痛太阳穴一阵阵的疼”, “感觉有点恶心”, “没有发烧”]}, {“id”: “002”, “symptoms”: [“咳嗽有黄痰”, “嗓子疼”, “流鼻涕”]}, ] asyncio.run(batch_process_cases(mock_cases))运行批量任务python batch_process.py这个脚本会异步并发地处理多个“病例”每个病例包含多条症状描述模拟连续的问诊对话并最终获取每个病例的建议总结。7. 资源占用与性能观察本项目主要消耗的是网络 I/O调用大模型 API和内存维护会话状态和消息历史。本地没有模型推理因此不消耗 GPU 显存。CPU/内存占用主要来自 Python 运行时、LangChain/LangGraph 框架以及会话状态存储。单个会话内存占用很小KB 到 MB 级别。批量处理时注意管理并发连接数和会话缓存避免内存泄漏。网络延迟性能瓶颈主要在于 LLM API 的响应时间。使用gpt-4o-mini或gpt-3.5-turbo通常能在 2-5 秒内返回结果。批量处理时使用异步请求asyncioaiohttp可以大幅提升吞吐量。LangSmith 开销开启 LangSmith 追踪后每个步骤LLM 调用、工具调用都会产生一个网络请求到 LangSmith 服务器这会增加少量延迟通常几百毫秒但换来了无价的调试信息。在生产环境可以考虑抽样追踪或仅在调试时开启。优化建议缓存对query_medical_knowledge_base这类工具的结果进行缓存。会话管理定期清理长时间不活动的会话状态。超时与重试为 LLM API 调用设置合理的超时和重试机制。限流在 API 服务层面对客户端请求进行限流防止滥用。8. 常见问题与排查方法问题现象可能原因排查方式解决方案导入 LangChain 相关模块失败版本不兼容或未安装pip listgrep langchain 检查版本查看错误信息运行时报错OpenAI API相关API 密钥错误、网络问题或余额不足检查.env文件用curl或简单脚本测试 API 连通性确认OPENAI_API_KEY和OPENAI_BASE_URL正确检查网络代理设置Agent 不调用工具一直闲聊提示词Prompt未清晰指示使用工具工具描述不准确检查agent_graph.py中的prompt检查工具函数的docstring在 System Prompt 中强调“使用工具”完善工具描述使其能被 LLM 正确理解LangGraph 图陷入无限循环路由逻辑should_continue有缺陷始终返回collect在should_continue函数中添加打印日志观察判断逻辑完善结束条件例如基于对话轮数、关键症状是否收集齐全等LangSmith 看不到追踪数据环境变量未正确设置项目名冲突确认.env文件已加载登录 LangSmith 网站查看对应项目重启终端或服务以加载新环境变量在 LangSmith 上创建新项目并更新LANGCHAIN_PROJECTFastAPI 服务启动失败端口被占用端口 8000 已被其他程序使用netstat -angrep 8000(Linux/macOS) 或netstat -ano批量处理时大量请求失败并发数过高导致 API 限流或服务崩溃观察错误日志降低aiohttp.TCPConnector(limit)的值增加请求间隔asyncio.sleep实现指数退避重试服务端增加限流中间件工具调用参数类型错误工具函数定义的参数类型与 Agent 传递的不匹配查看 LangSmith 追踪中工具调用的输入输出确保工具函数参数有明确的类型注解且 Agent 传递的是正确的字典格式9. 最佳实践与使用建议从简单开始逐步复杂化先构建一个只有 1-2 个工具和简单线性流程的 Agent确保它能跑通。然后再引入 LangGraph 处理复杂状态和循环最后集成 LangSmith 进行观察和优化。充分利用 LangSmith 进行调试这是 LangChain 生态最大的优势之一。在开发阶段务必开启 LangSmith。它可以帮助你可视化整个调用链看清每一步是调用了 LLM 还是工具输入输出是什么。检查提示词Prompt效果对比不同 Prompt 下 LLM 的响应。评估工具使用正确性查看工具是否被正确调用参数是否传递准确。性能分析统计每个步骤的耗时找到瓶颈。设计清晰的状态State在 LangGraph 中State 是所有节点共享的内存。设计时要明确哪些数据需要持久化在整个会话中如symptoms_collected哪些是临时变量。保持 State 结构扁平、简洁。为工具编写高质量的文档字符串DocstringLLM 依靠工具的 Docstring 来决定何时以及如何使用它。描述要准确、简洁包含参数说明和返回值的含义。实现健壮的错误处理在工具调用、LLM 调用、图执行步骤中加入try...except。对于可重试的错误如网络超时实现重试机制。关注安全与合规输入过滤对用户输入进行基本的清洗和过滤防止 Prompt 注入。输出审查对于医疗等敏感领域Agent 的最终输出应包含明确的免责声明并建议用户咨询专业人士。数据留存明确告知用户会话数据如何存储、使用和删除遵守相关法规。性能与成本权衡选择性价比合适的 LLM如gpt-4o-mini用于对话gpt-4用于复杂推理。对知识库查询RAG结果进行缓存。考虑对非实时场景的会话进行异步处理。10. 总结与下一步通过这个医疗问诊 Agent 的构建我们清晰地看到了 LangChain、LangGraph 和 LangSmith 是如何各司其职又紧密协作的LangChain提供了构建 Agent 的“砖块”LLM 封装、工具定义、记忆管理和提示词模板。它是基础组件库。LangGraph提供了组装“砖块”并赋予其智能流程的“蓝图”它让你能以图的形式定义复杂、有状态、带循环和条件分支的工作流。它是 Agent 的“大脑”和“调度中心”。LangSmith提供了观察和优化整个系统的“显微镜”和“仪表盘”从链路追踪、提示词管理到评估和监控它让 Agent 的开发从黑盒变成白盒。它是保障 Agent 稳定、可控的运维平台。最值得尝试的点不是简单地调用ChatGPT而是用 LangGraph 编排一个能自主决策、使用工具、管理状态的工作流。这是实现复杂业务逻辑的关键。最先应该验证的功能确保你的工具能被 Agent 正确理解和调用。在 LangSmith 上追踪一个完整的会话查看工具调用的输入输出是否符合预期。最容易踩的坑状态管理混乱在 LangGraph 中错误地修改了 State导致数据流不一致。画一张状态流转图有助于理解。提示词Prompt不够精准导致 LLM 不按预期使用工具或做出判断。多迭代、多测试利用 LangSmith 进行对比。忽略错误处理网络、API 限流、工具异常等情况没有处理导致整个 Agent 崩溃。后续扩展方向集成真实 RAG将query_medical_knowledge_base工具替换为连接真实医学数据库或文档库的检索增强生成管道。多模态输入接入语音识别或图像识别支持用户上传病历图片或语音描述症状。更复杂的决策流程引入更专业的医学分类或分诊逻辑让 Agent 能根据症状紧急程度决定是建议自我护理、预约门诊还是立即急诊。模型微调针对医疗对话数据对基础 LLM 进行微调SFT/LoRA使其术语使用和推理风格更专业。前端界面构建一个 Web 或移动端聊天界面提供更好的用户体验。这个项目就像一个“脚手架”展示了构建一个实用 AI Agent 的核心模式。你可以替换掉“医疗”这个领域套用类似的架构去开发客服、教育、娱乐等任何领域的对话智能体。建议收藏本文在动手实践时反复对照。 30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度