基于LangChain与LangGraph构建医疗问诊AI智能体实战教程

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

最近在尝试将大语言模型（LLM）应用到具体的垂直业务场景时，比如医疗问诊，我发现一个核心痛点：如何让一个AI助手不仅能理解用户问题，还能可靠地执行一系列复杂的、有状态的流程？例如，从症状收集、到初步分析、再到建议追问、最后生成报告。单纯调用一次LLM API是远远不够的，我们需要一个能编排“思考”和“行动”的框架。

这正是LangChain生态的价值所在。本文将从一个具体的“医疗问诊智能体（Agent）”项目出发，手把手带你构建一个具备多轮对话、状态管理和工具调用能力的AI助手。通过这个实战案例，你将清晰地理解LangChain、LangGraph和LangSmith这三个核心组件各自扮演的角色、如何协同工作，以及它们如何解决AI应用开发中的可靠性、可观测性和工程化难题。无论你是刚接触Agent概念的新手，还是希望将原型落地的开发者，这篇系统化的教程都能提供从零到一的完整路径。

1. 背景与核心概念：为什么需要Agent框架？

在深入代码之前，我们有必要厘清几个关键概念，以及它们所要解决的核心问题。

1.1 什么是AI Agent（智能体）？

简单来说，一个AI Agent是一个能够感知环境、进行决策并执行行动以达到目标的系统。在LLM的语境下，Agent通常指一个能够使用大语言模型进行“思考”，并调用外部工具（如搜索、计算、数据库查询）来“行动”的程序。它不再是简单的问答机器人，而是一个具备自主规划和执行复杂任务能力的智能体。

在我们的“医疗问诊”场景中，一个理想的Agent应该能够：

1. 理解用户描述的症状。
2. 规划问诊流程：可能需要先询问病史，再询问症状细节。
3. 执行行动：调用医学知识库工具进行查询，或调用诊断逻辑进行计算。
4. 维持状态：记住之前对话中已获取的信息（如患者年龄、既往病史）。
5. 迭代：根据工具返回的结果和用户的新输入，决定下一步是继续提问、给出建议还是结束问诊。

1.2 LangChain、LangGraph与LangSmith的角色与关系

这是本文的核心。很多人对这三个工具感到困惑，其实它们构成了一个从快速原型到生产部署的完整技术栈。

• LangChain： 应用构建的“脚手架”和“工具箱”。

  • 它是什么：一个用于开发由LLM驱动的应用程序的框架。它提供了大量的标准化组件（如提示词模板、文档加载器、向量存储接口、链式调用等），让你能像搭积木一样快速组合出功能。
  • 核心价值：降低开发门槛。它封装了与不同LLM提供商（OpenAI, Anthropic等）的交互、文本处理、记忆管理等通用功能，让你专注于业务逻辑。
  • 在问诊Agent中：我们用LangChain来定义与LLM的对话、创建提示词模板、管理对话历史（Memory）、以及封装我们要调用的工具（如症状诊断工具）。

• LangGraph： 智能体工作流的“编排引擎”和“状态机”。

  • 它是什么：一个基于图（Graph）的框架，专门用于构建有状态的、多步骤的、可能循环的Agent工作流。它将工作流定义为节点（Node）和边（Edge）组成的有向图。
  • 核心价值：实现复杂、可靠的流程控制。它解决了传统链式调用难以处理循环、条件分支和持久化状态的问题。LangGraph提供了对Agent推理和行动过程的细粒度控制。
  • 在问诊Agent中：我们用LangGraph来定义问诊的完整流程：开始 -> 分析症状 -> [是否需要更多信息？] -> 是：提问；否：调用诊断工具 -> 生成报告 -> 结束。这个流程可以循环，状态（已收集的信息）在整个图中传递。

• LangSmith： 智能体开发的“调试台”和“监控中心”。

  • 它是什么：一个AI应用开发的平台，提供调试、测试、评估和监控功能。
  • 核心价值：提升开发效率与系统可观测性。它让你能可视化地追踪每一次Agent调用的完整生命周期（输入的提示词、LLM的思考过程、调用的工具、返回的结果），方便定位问题、评估效果并进行迭代优化。
  • 在问诊Agent中：我们用LangSmith来记录每一次问诊会话，查看Agent在哪个步骤做出了什么决策，调用工具的参数和结果是什么，从而优化提示词或工作流逻辑。

三者的关系可以类比为：

• LangChain提供了建造房屋的砖瓦和水泥（基础组件）。
• LangGraph提供了施工蓝图和脚手架，告诉你如何把这些材料组装成坚固的、结构复杂的房屋（工作流编排）。
• LangSmith则是工程监理和质检仪，确保建造过程可控，房屋质量达标，并能持续改进（开发运维）。

接下来，我们将通过构建医疗问诊Agent，将这三者串联起来。

2. 环境准备与项目初始化

我们使用Python作为开发语言。请确保你的Python版本在3.8以上。

2.1 创建虚拟环境与安装依赖

首先，创建一个新的项目目录并设置虚拟环境，这是一个好习惯，可以隔离项目依赖。

# 创建项目目录 mkdir medical-chat-agent cd medical-chat-agent # 创建虚拟环境 (以venv为例) python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate

安装核心依赖库。我们将安装langchain,langgraph,langsmith的客户端，以及OpenAI的SDK（因为我们使用GPT模型作为LLM）。同时安装python-dotenv来管理环境变量。

pip install langchain langgraph langsmith openai python-dotenv

2.2 配置API密钥与环境变量

你需要准备以下API密钥：

1. OpenAI API Key：用于访问GPT模型。
2. LangSmith API Key（可选但强烈推荐）：用于将追踪日志发送到LangSmith平台。你可以在 LangSmith官网 注册并获取。

在项目根目录下创建一个.env文件来存储这些敏感信息：

# .env OPENAI_API_KEY=sk-your-openai-api-key-here LANGCHAIN_API_KEY=ls-your-langsmith-api-key-here # LangSmith默认会将数据发送到其云服务，如需自定义，可设置以下变量 # LANGCHAIN_TRACING_V2=true # LANGCHAIN_ENDPOINT="https://api.smith.langchain.com" # LANGCHAIN_PROJECT="medical-agent-tutorial" # 你的项目名

重要：请将.env文件添加到.gitignore中，避免将密钥提交到版本控制系统。

# .gitignore .env venv/ __pycache__/ *.pyc

2.3 项目结构规划

我们的项目结构将如下所示，这有助于代码组织：

medical-chat-agent/ ├── .env # 环境变量（不提交） ├── .gitignore ├── requirements.txt # 依赖列表 ├── config.py # 配置加载 ├── tools/ # 自定义工具目录 │ └── medical_tools.py ├── graph/ # LangGraph定义目录 │ └── medical_agent.py ├── run_agent.py # 主运行脚本 └── utils.py # 工具函数

现在，让我们开始编写代码。首先创建config.py来加载环境变量：

# config.py import os from dotenv import load_dotenv load_dotenv() # 从 .env 文件加载环境变量 OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") LANGCHAIN_API_KEY = os.getenv("LANGCHAIN_API_KEY") # 简单验证 if not OPENAI_API_KEY: raise ValueError("请在 .env 文件中设置 OPENAI_API_KEY") # LANGCHAIN_API_KEY 可选，没有则仅本地运行，无LangSmith追踪

3. 核心组件拆解：工具、模型与记忆

在构建图之前，我们需要用LangChain创建几个基础组件。

3.1 定义医疗问诊工具（Tools）

工具是Agent与外界交互的“手”。这里我们创建两个简单的模拟工具：一个用于根据症状查询可能的疾病，另一个用于计算BMI（身体质量指数）。在实际应用中，这些工具可以连接真实的医学数据库或API。

# tools/medical_tools.py from langchain.tools import tool from typing import Dict, List import json @tool def search_medical_knowledge(symptoms: str) -> str: """ 根据输入的症状描述，查询医学知识库，返回可能的疾病列表及相关信息。 参数: symptoms: 患者描述的症状，如“头痛、发烧、咳嗽”。 返回: 一个格式化的字符串，包含可能的疾病和建议。 """ # 这里是模拟逻辑。真实场景应接入医学知识图谱或数据库。 symptom_keywords = symptoms.lower() knowledge_base = { "headache,fever,cough": [ {"disease": "Common Cold", "confidence": "Medium", "advice": "Rest, hydrate, over-the-counter cold medicine."}, {"disease": "Influenza (Flu)", "confidence": "Medium", "advice": "Rest, antiviral medication if early, see doctor if severe."}, ], "chest pain,shortness of breath": [ {"disease": "Angina", "confidence": "High", "advice": "**This is a medical emergency. Seek immediate medical attention.**"}, ], "fatigue,weight loss,increased thirst": [ {"disease": "Diabetes", "confidence": "Medium", "advice": "Consult a doctor for blood sugar tests."}, ] } # 简单匹配逻辑 for key, value in knowledge_base.items(): if all(kw in symptom_keywords for kw in key.split(',')): return json.dumps(value, indent=2, ensure_ascii=False) return json.dumps([{"disease": "Symptom pattern not specifically recognized.", "confidence": "Low", "advice": "Please consult a healthcare professional for accurate diagnosis."}]) @tool def calculate_bmi(weight_kg: float, height_m: float) -> Dict: """ 计算身体质量指数 (BMI)。 参数: weight_kg: 体重，单位千克。 height_m: 身高，单位米。 返回: 包含BMI值和分类的字典。 """ bmi = weight_kg / (height_m ** 2) category = "" if bmi < 18.5: category = "Underweight" elif bmi < 25: category = "Normal weight" elif bmi < 30: category = "Overweight" else: category = "Obesity" return { "bmi": round(bmi, 2), "category": category, "interpretation": f"A BMI of {round(bmi,2)} is considered {category}." } # 将工具放入列表，方便后续使用 MEDICAL_TOOLS = [search_medical_knowledge, calculate_bmi]

3.2 初始化LLM与提示词

我们使用LangChain的ChatOpenAI来封装与GPT的交互，并创建一个系统提示词来设定Agent的角色和行为准则。

# graph/medical_agent.py (部分代码) from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.schema import SystemMessage from config import OPENAI_API_KEY # 1. 初始化LLM。我们使用gpt-3.5-turbo，成本较低，适合演示。 llm = ChatOpenAI( model="gpt-3.5-turbo", temperature=0.1, # 温度调低，使输出更稳定、更专业 openai_api_key=OPENAI_API_KEY, streaming=True, # 启用流式输出，提升用户体验 ) # 2. 定义系统提示词，赋予Agent“医生助手”的角色 system_prompt = SystemMessage(content="""你是一个专业的医疗问诊助手。你的目标是帮助用户初步分析症状，收集必要信息，并提供基于医学知识的可能性参考和建议。 请遵循以下准则： 1. 保持专业、共情和谨慎的态度。你并非替代医生，不能提供最终诊断。 2. 主动、有条理地收集信息。例如，当用户说“我头疼”，你应该询问疼痛部位、性质、持续时间、伴随症状等。 3. 在信息足够时，可以调用工具来查询医学知识或进行计算。 4. 所有涉及严重疾病（如胸痛、呼吸困难）的建议，必须强调立即寻求专业医疗帮助。 5. 每次回复应清晰、结构化，并引导对话进行下去。 """) # 3. 创建提示词模板，预留对话历史的位置 prompt_template = ChatPromptTemplate.from_messages([ system_prompt, MessagesPlaceholder(variable_name="messages"), # 历史消息 MessagesPlaceholder(variable_name="agent_scratchpad"), # Agent的思考过程 ])

3.3 构建基础的Agent执行器（非LangGraph版本）

为了理解LangGraph带来的价值，我们先看看只用LangChain如何构建一个简单的、具有工具调用能力的Agent。这通常使用create_react_agent或类似高阶函数。

# graph/medical_agent.py (继续添加) from langchain.agents import create_react_agent, AgentExecutor from langchain.memory import ConversationBufferMemory from tools.medical_tools import MEDICAL_TOOLS # 1. 创建记忆，保存对话历史 memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True) # 2. 使用LangChain的ReAct范式创建Agent # 注意：`create_react_agent` 在较新版本中可能有所变化，这里展示核心逻辑。 # 实际中，我们更推荐使用LangGraph，但此例用于对比。 from langchain.agents import AgentExecutor, create_openai_tools_agent # 创建Agent agent = create_openai_tools_agent(llm, MEDICAL_TOOLS, prompt_template) # 创建执行器 agent_executor_simple = AgentExecutor( agent=agent, tools=MEDICAL_TOOLS, memory=memory, verbose=True, # 打印详细执行过程 handle_parsing_errors=True, # 处理解析错误 ) # 这个执行器可以运行，但控制流是黑盒，难以自定义复杂循环和状态。

这个agent_executor_simple可以处理工具调用，但其内部的工作流（何时思考、何时调用工具、何时结束）是固定的、不透明的。当我们需要实现“先问三个问题，再调用工具，然后根据结果决定是否继续问”这样的定制流程时，它就力不从心了。这就是我们需要LangGraph的原因。

4. 使用LangGraph构建可控的问诊工作流

现在，我们来用LangGraph重新定义问诊Agent。我们将工作流建模为一个有向图。

4.1 定义状态（State）

首先，我们需要定义在整个工作流中传递的“状态”。状态是一个字典，包含了对话所需的所有信息。

# graph/medical_agent.py (引入新部分) from typing import TypedDict, Annotated, List, Union from langgraph.graph.message import AnyMessage, add_messages import operator # 使用TypedDict定义状态的结构，这有助于类型检查和代码清晰度。 class AgentState(TypedDict): # 消息列表：包含用户输入、AI回复、工具结果等所有消息 messages: Annotated[List[AnyMessage], add_messages] # 从对话中提取的关键信息（结构化数据） extracted_info: dict # 一个标志位，指示Agent下一步应该做什么。例如：'ask_question', 'call_tool', 'finalize' next_action: str # 记录已询问的问题，避免重复 questions_asked: List[str]

• Annotated[List[AnyMessage], add_messages]：这是一个LangGraph的“缩减器”（reducer）。add_messages是一个函数，它确保新的消息被正确地追加到messages列表中，而不是覆盖。这是管理对话历史的最佳实践。
• extracted_info：我们将把从对话中提取的年龄、症状描述等信息结构化地存储在这里。
• next_action：这是控制工作流走向的关键。
• questions_asked：用于实现简单的记忆，避免重复提问。

4.2 创建图节点（Nodes）

节点是图中的函数，它们接收状态，执行操作，并返回更新后的状态。

节点1：路由节点（Router）这个节点是图的大脑，它根据当前状态决定下一步该去哪个节点。

# graph/medical_agent.py from langchain.schema import HumanMessage, AIMessage def router_node(state: AgentState) -> str: """ 根据当前状态，决定下一步动作。 返回一个字符串，代表下一个节点的名称。 """ messages = state['messages'] last_message = messages[-1] # 如果是用户的第一次输入，或者我们刚问完问题，需要分析 if isinstance(last_message, HumanMessage): # 这里可以加入更复杂的逻辑，比如分析信息是否足够 # 简单起见，我们设定：如果extracted_info里还没有`symptoms`，就去提问 if not state.get('extracted_info', {}).get('symptoms'): return "ask_questions" else: # 如果有症状信息了，就去调用工具分析 return "call_tool" # 如果是AI的回复（可能是提问或最终答案），并且我们设定了下一步动作 elif isinstance(last_message, AIMessage) and state.get('next_action'): return state['next_action'] # 默认情况，结束对话 return "__end__"

节点2：提问节点（Ask Questions）这个节点负责生成追问，以收集更多信息。

def question_node(state: AgentState) -> AgentState: """生成一个追问，以收集更多症状信息。""" from langchain_core.prompts import ChatPromptTemplate # 构建一个专注于信息收集的提示词 prompt = ChatPromptTemplate.from_messages([ SystemMessage(content="你是一个细心的医生助手。根据当前对话历史，提出一个最相关的问题来收集更多关于患者症状的信息。只问一个问题。"), MessagesPlaceholder(variable_name="messages"), ]) chain = prompt | llm # 调用LLM生成问题 question = chain.invoke({"messages": state['messages']}).content # 更新状态：将AI生成的问题添加到消息历史，并记录已问过的问题 new_messages = state['messages'] + [AIMessage(content=question)] new_questions = state.get('questions_asked', []) + [question] return { "messages": new_messages, "extracted_info": state['extracted_info'], "next_action": "wait_for_user", # 设置下一步：等待用户回答 "questions_asked": new_questions, }

节点3：调用工具节点（Call Tool）当信息收集得差不多时，这个节点会调用我们之前定义的医学工具。

def tool_node(state: AgentState) -> AgentState: """调用医学工具进行分析。""" # 从状态中提取症状信息 symptoms = state['extracted_info'].get('symptoms', '') if not symptoms: # 如果没有症状，回到提问节点 return {**state, "next_action": "ask_questions"} # 1. 准备调用工具的提示词（让LLM决定使用哪个工具及参数） from langchain.agents.format_scratchpad.openai_tools import format_to_openai_tool_messages from langchain.agents.output_parsers.openai_tools import OpenAIToolsAgentOutputParser # 将工具绑定到LLM llm_with_tools = llm.bind_tools(MEDICAL_TOOLS) # 构建Agent提示词 agent_prompt = ChatPromptTemplate.from_messages([ system_prompt, MessagesPlaceholder(variable_name="messages"), ("user", "根据上述对话历史，特别是患者的症状描述：'{symptoms}'，请调用合适的工具进行分析。如果信息不足，请说明需要补充什么。"), MessagesPlaceholder(variable_name="agent_scratchpad"), ]) # 创建Agent链 agent = agent_prompt | llm_with_tools | OpenAIToolsAgentOutputParser() # 2. 执行Agent（这里简化了，实际LangGraph有更优雅的方式集成） # 为了清晰，我们直接模拟调用 search_medical_knowledge 工具 tool_input = symptoms tool_result = search_medical_knowledge.invoke(tool_input) # 3. 将工具结果添加到消息历史 tool_message = AIMessage(content=f"我查询了医学知识库，结果如下：\n{tool_result}") new_messages = state['messages'] + [tool_message] # 4. 更新状态，准备生成最终建议 return { "messages": new_messages, "extracted_info": state['extracted_info'], "next_action": "generate_final_advice", "questions_asked": state['questions_asked'], }

节点4：生成最终建议节点 & 等待用户节点我们还需要节点来处理最终输出和等待用户输入。

def final_advice_node(state: AgentState) -> AgentState: """根据工具结果和对话历史，生成最终的建议和总结。""" prompt = ChatPromptTemplate.from_messages([ SystemMessage(content="你是一名医生助手。请根据工具查询的结果和整个对话历史，为用户生成一份清晰、谨慎的初步评估和建议总结。务必提醒用户这不能替代专业医疗诊断。"), MessagesPlaceholder(variable_name="messages"), ]) chain = prompt | llm final_advice = chain.invoke({"messages": state['messages']}).content new_messages = state['messages'] + [AIMessage(content=final_advice)] # 对话结束，next_action 设为结束 return { "messages": new_messages, "extracted_info": state['extracted_info'], "next_action": "__end__", "questions_asked": state['questions_asked'], } def wait_for_user_node(state: AgentState) -> AgentState: """这是一个特殊节点，它不执行任何操作，只是将控制权交还给用户输入。 在实际的LangGraph流中，我们通常通过外部调用来注入用户消息。 这个节点可以简单地将next_action设置为'router'，让路由节点处理新消息。 """ return {**state, "next_action": "router"}

4.3 组装图（Graph）并编译

现在，我们将所有节点和边组合起来，形成一个完整的工作流。

# graph/medical_agent.py from langgraph.graph import StateGraph, END # 1. 创建图构建器 workflow = StateGraph(AgentState) # 2. 添加节点 workflow.add_node("router", router_node) workflow.add_node("ask_questions", question_node) workflow.add_node("call_tool", tool_node) workflow.add_node("generate_final_advice", final_advice_node) workflow.add_node("wait_for_user", wait_for_user_node) # 用于等待外部输入 # 3. 设置入口点 workflow.set_entry_point("router") # 4. 添加边（定义节点间的流转逻辑） workflow.add_conditional_edges( "router", router_node, # 路由函数本身也决定下一个节点 { "ask_questions": "ask_questions", "call_tool": "call_tool", "__end__": END, } ) workflow.add_edge("ask_questions", "wait_for_user") workflow.add_edge("wait_for_user", "router") # 用户输入后，返回路由节点 workflow.add_edge("call_tool", "generate_final_advice") workflow.add_edge("generate_final_advice", END) # 5. 编译图，得到可执行的对象 medical_agent_graph = workflow.compile()

这个图定义了一个清晰的流程：router-> (ask_questions->wait_for_user->router) 循环 ->call_tool->generate_final_advice-> 结束。

4.4 创建信息提取节点（关键步骤）

我们漏掉了一个重要环节：如何将用户非结构化的描述，提取成结构化的extracted_info？我们需要一个专门的节点来做这件事。

def extract_info_node(state: AgentState) -> AgentState: """从最新的用户消息中提取结构化信息（如症状）。""" messages = state['messages'] last_message = messages[-1] if not isinstance(last_message, HumanMessage): return state # 如果不是用户消息，直接返回 user_input = last_message.content # 使用一个简单的提示词让LLM提取信息 extraction_prompt = ChatPromptTemplate.from_messages([ SystemMessage(content="你是一个信息提取助手。从用户的输入中，提取与医疗问诊相关的结构化信息。只返回一个JSON对象，包含以下可能的字段：'symptoms'（症状描述字符串）、'duration'（持续时间）、'severity'（严重程度）。如果某个字段无法提取，就设为空字符串。"), HumanMessage(content=user_input), ]) extraction_chain = extraction_prompt | llm # 注意：这里需要LLM支持JSON模式，或者使用Pydantic输出解析器更可靠。 # 为简化，我们假设LLM返回了JSON字符串。 try: import json # 在实际中，应使用更健壮的解析方式，例如： # from langchain.output_parsers import PydanticOutputParser # from pydantic import BaseModel extracted_json = json.loads(extraction_chain.invoke({}).content) except: extracted_json = {"symptoms": user_input, "duration": "", "severity": ""} # 降级方案 # 更新状态中的 extracted_info current_info = state.get('extracted_info', {}) updated_info = {**current_info, **extracted_json} return { "messages": state['messages'], "extracted_info": updated_info, "next_action": state['next_action'], "questions_asked": state['questions_asked'], }

然后，我们需要修改图，在router节点之后、决定下一步之前，加入这个信息提取节点。更优雅的方式是修改router_node的逻辑，或者在wait_for_user到router之间插入extract_info_node。为了流程清晰，我们调整图结构：

# 修改后的图组装逻辑（替代之前的add_node和add_edge部分） workflow.add_node("router", router_node) workflow.add_node("extract_info", extract_info_node) # 新增节点 workflow.add_node("ask_questions", question_node) workflow.add_node("call_tool", tool_node) workflow.add_node("generate_final_advice", final_advice_node) workflow.add_node("wait_for_user", wait_for_user_node) workflow.set_entry_point("router") # 新的边逻辑 workflow.add_conditional_edges( "router", router_node, { "ask_questions": "extract_info", # 先去提取信息 "call_tool": "extract_info", # 调用工具前也确保信息最新 "__end__": END, } ) workflow.add_edge("extract_info", "ask_questions") # 提取信息后提问 # 注意：这里需要根据router的结果动态决定extract_info后的走向，上述简化了。 # 更正确的做法是让router返回后，经过extract_info，再根据一个条件跳转到ask_questions或call_tool。 # 为了教程清晰，我们采用一个简化但可工作的版本： def router_after_extract(state: AgentState) -> str: """在提取信息后，再次路由。""" # 简单逻辑：如果已有症状，就去调用工具；否则提问。 if state['extracted_info'].get('symptoms'): return "call_tool" else: return "ask_questions" workflow.add_conditional_edges( "extract_info", router_after_extract, { "ask_questions": "ask_questions", "call_tool": "call_tool", } ) workflow.add_edge("ask_questions", "wait_for_user") workflow.add_edge("wait_for_user", "router") # 用户输入后，回到主路由 workflow.add_edge("call_tool", "generate_final_advice") workflow.add_edge("generate_final_advice", END) medical_agent_graph = workflow.compile()

现在，我们的图变得更加智能：router->extract_info-> (ask_questions或call_tool) -> ...。

5. 运行Agent并与LangSmith集成

5.1 编写主运行脚本

创建一个脚本来与编译好的图进行交互。

# run_agent.py import asyncio from langchain.schema import HumanMessage from graph.medical_agent import medical_agent_graph from config import LANGCHAIN_API_KEY import os # 如果设置了LangSmith API Key，则启用追踪 if LANGCHAIN_API_KEY: os.environ["LANGCHAIN_TRACING_V2"] = "true" os.environ["LANGCHAIN_PROJECT"] = "Medical-Agent-Tutorial" print("LangSmith tracing enabled.") else: print("LangSmith tracing disabled. Set LANGCHAIN_API_KEY in .env to enable.") async def chat_with_agent(): """与医疗问诊Agent进行异步对话。""" print("=== 医疗问诊智能助手 ===") print("描述你的症状或健康问题。输入 'quit' 或 '退出' 结束对话。\n") # 初始化图的状态 initial_state = { "messages": [], "extracted_info": {}, "next_action": "router", "questions_asked": [], } current_state = initial_state while True: # 获取用户输入 user_input = input("\n你: ").strip() if user_input.lower() in ['quit', '退出', 'exit']: print("助手: 感谢使用，请保重身体！") break # 将用户输入添加到消息历史 human_message = HumanMessage(content=user_input) # 注意：我们需要将新消息“注入”到当前状态中。 # LangGraph的`stream`或`ainvoke`方法可以接受一个包含新消息的配置。 # 这里我们模拟一个更新状态并调用图的过程。 new_messages = current_state['messages'] + [human_message] config = {"configurable": {"thread_id": "user_session_1"}} # 用于会话标识 # 使用 astream_events 进行流式输出（推荐） print("\n助手: ", end="", flush=True) full_response = "" async for event in medical_agent_graph.astream_events( {"messages": new_messages, **{k:v for k,v in current_state.items() if k != 'messages'}}, config=config, version="v1" ): # 这里可以处理不同类型的事件，例如节点开始、结束，流式token等。 # 为了简化，我们只打印AI生成的内容。 if event["event"] == "on_chat_model_stream" and event["name"] == "ChatOpenAI": content = event["data"]["chunk"].content if content: print(content, end="", flush=True) full_response += content print() # 换行 # 更新当前状态（在实际中，需要从图的最终输出获取完整状态） # 更简单的方式是直接调用`ainvoke`获取最终状态 final_state = await medical_agent_graph.ainvoke( {"messages": new_messages, **{k:v for k,v in current_state.items() if k != 'messages'}}, config=config ) current_state = final_state if __name__ == "__main__": asyncio.run(chat_with_agent())

5.2 在LangSmith中查看追踪

运行脚本前，确保你的.env文件中的LANGCHAIN_API_KEY已设置。运行后，打开 LangSmith网站 ，你会在指定的项目（Medical-Agent-Tutorial）下看到这次会话的完整追踪记录。

在LangSmith的追踪界面中，你可以：

1. 可视化工作流：看到图（Graph）的每个节点是如何被执行的。
2. 检查输入输出：点击每个节点，查看输入的状态、LLM的提示词、生成的响应、调用的工具及其结果。
3. 调试问题：如果Agent行为异常，你可以精确地定位是哪个节点的逻辑或LLM的回复出了问题。
4. 评估与迭代：你可以对多次运行的结果进行比较，评估Agent的表现，并据此优化提示词或图的结构。

6. 常见问题与排查思路

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

7. 最佳实践与工程建议

将原型Agent转化为可生产部署的系统，需要考虑更多因素：

1. 状态管理的持久化：目前的AgentState存在于内存中，会话结束即消失。在生产环境中，你需要将状态（尤其是messages和extracted_info）持久化到数据库（如Redis、PostgreSQL）中，并通过configurable的thread_id来检索和恢复会话状态。

2. 更健壮的路由与错误处理：

   • 在图中添加错误处理节点（Error Handling Node），用于捕获工具调用失败、LLM响应解析失败等异常，并引导工作流到恢复路径或优雅失败。
   • 为路由逻辑设置超时和最大步数限制，防止Agent陷入无限循环。

3. 工具的可扩展性与安全性：

   • 将工具注册到中央仓库，方便管理和复用。
   • 对于执行写操作或访问敏感数据的工具（如预约挂号），必须实现严格的权限检查和用户认证。
   • 考虑为工具调用添加“人工审核”节点（Human-in-the-loop），在高风险决策前介入。

4. 提示词工程与管理：

   • 不要将提示词硬编码在代码中。使用LangChain的PromptTemplate并从外部文件（如YAML、JSON）或配置中心加载。
   • 为不同的节点（提问、信息提取、总结）设计专用的、优化的提示词。
   • 利用LangSmith的“Playground”功能迭代和测试你的提示词。

5. 性能与成本优化：

   • 对于复杂的图，考虑异步执行节点以提高吞吐量。
   • 使用更便宜的模型（如gpt-3.5-turbo）进行简单的路由和分类任务，仅在需要深度分析和生成时使用更强大的模型（如gpt-4）。
   • 利用LangSmith的追踪数据来分析耗时和成本，优化工作流。

6. 测试与评估：

   • 使用LangSmith的**数据集（Datasets）和评估（Evaluation）**功能，构建一组测试用例（例如，不同的症状描述），并定义评估标准（如“是否询问了关键信息”、“最终建议是否包含安全警告”）。
   • 定期运行评估，监控Agent性能的回归情况。

通过这个从零构建医疗问诊Agent的旅程，你应该对LangChain、LangGraph和LangSmith的定位与协作有了深刻的理解。LangChain提供了构建AI应用的基础模块，LangGraph赋予了我们对复杂、有状态工作流进行精细编排的能力，而LangSmith则为我们提供了洞察整个系统运行状况的眼睛，是迭代和优化过程中不可或缺的利器。这套组合拳，正是开发现代、可靠、可维护的AI Agent应用的最佳实践。

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