医学 Agent：自带医学深度调研 deep research，优化治疗策略+药物参考

医学 Agent：自带医学深度调研deep research，优化治疗策略+药物参考

• • 第一版代码
  • • 输出结果，居然连不上网
    • 运行结果梳理
  • 第二版
  • • 结果测试

 

---

第一版代码

医疗顾问AI系统 - 基于Qwen API 的智能医疗助手

最终目标：构建一个能够查询疾病治疗方案和药物知识的AI医疗顾问系统

层层分解：

1. 环境配置模块 - 配置API密钥和环境变量
2. 思维链追踪模块 - 展示AI的思考过程
3. 医疗工具模块 - 定义疾病治疗和药物查询工具
4. AI模型初始化模块 - 配置Qwen语言模型
5. Agent代理模块 - 创建智能代理和执行器
6. 主控制模块 - 领导式规划，协调整个系统运行

# ================================
# 1. 依赖库导入模块 - 为什么要导入这些库？
# ================================import os  # 操作系统接口库 - 用于设置环境变量，管理系统配置
from dotenv import load_dotenv  # 环境变量加载库 - 从.env文件加载API密钥等敏感配置
from langchain_openai import ChatOpenAI  # OpenAI兼容接口库 - 与Qwen API通信的桥梁
from langchain.tools import tool  # 工具装饰器 - 将普通函数转换为LangChain可用的工具
from langchain.agents import AgentExecutor, create_react_agent  # 智能代理库 - 创建能够使用工具的AI代理
from langchain.prompts import PromptTemplate  # 提示模板库 - 标准化AI的输入格式
from langchain.memory import ConversationBufferMemory  # 记忆库 - 保存对话历史，实现上下文连续性
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler  # 流式输出库 - 实时显示AI思考过程
from langchain_community.tools import DuckDuckGoSearchRun  # 搜索工具库 - 提供网络搜索能力
import json  # JSON处理库 - 处理结构化数据
import datetime  # 时间处理库 - 处理时间相关操作
from typing import List, Union, Dict, Any  # 类型注解库 - 提供代码类型提示，增强可读性# ================================
# 2. 环境配置模块 - 系统基础设置
# ================================def setup_environment():"""环境配置函数 - 领导级别的环境准备工作功能：加载环境变量，设置API密钥和基础URL"""# 加载.env文件中的环境变量 - 保护敏感信息不被硬编码load_dotenv()# 设置Qwen API密钥 - 使用环境变量方式保护API密钥安全os.environ["OPENAI_API_KEY"] = "sk-025a6...."# 设置Qwen API的基础URL - 使用阿里云DashScope的OpenAI兼容接口QWEN_BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"return QWEN_BASE_URL# ================================
# 3. 思维链追踪模块 - AI思考过程可视化
# ================================class ThoughtProcessCallbackHandler(StreamingStdOutCallbackHandler):"""思维链回调处理器 - 捕获和展示AI的思考过程为什么需要这个模块？- AI的决策过程对用户来说是黑盒，需要透明化- 医疗场景需要可解释的AI，展示推理逻辑- 帮助调试和优化AI的表现核心功能：- 实时捕获AI输出的每个token（词元）- 识别思考标记，分离思考过程和最终答案- 格式化展示完整的推理链条"""def __init__(self):"""初始化思维链收集器"""super().__init__()self.thoughts = []  # 存储所有思考步骤的列表self.current_thought = ""  # 当前正在收集的思考内容self.is_thinking = False  # 标记是否正在思考阶段def on_llm_new_token(self, token: str, **kwargs) -> None:"""处理AI输出的新token参数：token - AI输出的单个词元"""# 实时输出token，让用户看到AI"思考"的过程print(token, end="", flush=True)# 检测思考开始的标记 - 支持中英文标记if "思考:" in token or "Thought:" in token:self.is_thinking = True  # 进入思考收集模式self.current_thought = token  # 开始收集当前思考# 如果正在思考阶段，继续收集内容elif self.is_thinking:self.current_thought += token# 检测到动作标记时，结束当前思考收集if "Action:" in token:self.thoughts.append(self.current_thought)self.is_thinking = Falsedef on_llm_end(self, response, **kwargs) -> None:"""AI输出结束时的处理保存最后一个思考并展示完整思维链"""# 保存最后一个思考（如果存在）if self.current_thought:self.thoughts.append(self.current_thought)self.current_thought = ""# 格式化展示完整的思维链 - 让用户理解AI的推理过程if self.thoughts:print("\n\n===== 完整思维链 =====")for i, thought in enumerate(self.thoughts):print(f"思考 {i+1}: {thought}\n")# ================================
# 4. 医疗工具模块 - 核心业务功能实现
# ================================@tool
def search_disease_treatment(disease_name: str) -> str:"""疾病治疗方案搜索工具功能：搜索特定疾病的治疗方法和临床策略参数：disease_name - 疾病名称（如"2型糖尿病"）返回：治疗方案的详细信息为什么需要这个工具？- 医疗知识更新快，需要获取最新的治疗指南- 不同疾病有不同的治疗策略，需要专门查询"""try:# 创建DuckDuckGo搜索实例 - 免费的搜索引擎APIsearch = DuckDuckGoSearchRun()# 执行搜索 - 使用专业医疗关键词提高搜索精度result = search.run(f"{disease_name} 治疗方法 临床指南")return resultexcept Exception as e:# 搜索失败时的备用医疗知识库if "糖尿病" in disease_name:return """2型糖尿病治疗策略包括：
1. 生活方式干预：健康饮食、规律运动、体重管理
2. 药物治疗：二甲双胍为一线用药，必要时联合其他降糖药
3. 血糖监测：定期检测血糖、糖化血红蛋白
4. 并发症预防：控制血压、血脂，保护肾脏和心血管
5. 患者教育：糖尿病自我管理知识
注：搜索服务暂时不可用，以上为基础医疗知识参考"""return f"搜索服务暂时不可用，无法获取{disease_name}的最新治疗信息"@tool
def search_disease_medications(disease_name: str) -> str:"""疾病用药方案搜索工具功能：搜索特定疾病的推荐用药和处方信息参数：disease_name - 疾病名称返回：药物处方和用药指导"""try:search = DuckDuckGoSearchRun()# 搜索药物相关信息 - 关注处方和用药指南result = search.run(f"{disease_name} 推荐药物 处方 用药指南")return resultexcept Exception as e:# 糖尿病用药的备用知识库if "糖尿病" in disease_name:return """2型糖尿病常用药物：
1. 二甲双胍：一线用药，改善胰岛素敏感性
2. 磺脲类：格列齐特、格列美脲等，促进胰岛素分泌
3. DPP-4抑制剂：西格列汀、维格列汀等
4. SGLT-2抑制剂：达格列净、恩格列净等
5. GLP-1受体激动剂：利拉鲁肽、司美格鲁肽等
6. 胰岛素：根据病情需要使用
注：搜索服务暂时不可用，以上为基础用药知识参考"""return f"搜索服务暂时不可用，无法获取{disease_name}的最新用药信息"@tool
def search_medication_details(medication_name: str) -> str:"""药物详细信息查询工具功能：查询特定药物的详细信息、用法用量和副作用参数：medication_name - 药物名称（如"二甲双胍"）返回：药物的详细使用说明"""try:search = DuckDuckGoSearchRun()# 搜索药物详细信息 - 关注用法用量和安全性result = search.run(f"{medication_name} 用法用量 副作用 禁忌症")return resultexcept Exception as e:# 常见药物的详细信息备用库medication_info = {"二甲双胍": """二甲双胍详细信息：
用法用量：起始剂量500mg，每日2-3次，餐时或餐后服用，最大剂量2000-2500mg/日
作用机制：减少肝糖输出，改善胰岛素敏感性
副作用：胃肠道反应（恶心、腹泻、腹胀）、乳酸酸中毒（罕见）
禁忌症：严重肾功能不全、肝功能不全、心衰、酗酒
注意事项：定期监测肾功能""","格列齐特": """格列齐特详细信息：
用法用量：30-120mg/日，分1-2次服用，餐前30分钟
作用机制：刺激胰岛β细胞分泌胰岛素
副作用：低血糖、体重增加、皮疹
禁忌症：1型糖尿病、糖尿病酮症酸中毒、严重肝肾功能不全
注意事项：注意低血糖反应，老年患者慎用""",}# 查找匹配的药物信息for med, info in medication_info.items():if med in medication_name:return info + "\n注：搜索服务暂时不可用，以上为基础药物知识参考"return f"搜索服务暂时不可用，无法获取{medication_name}的详细信息"@tool
def calculator(expression: str) -> str:"""数学计算工具功能：执行数学计算，用于药物剂量计算等参数：expression - 数学表达式字符串返回：计算结果使用场景：- 根据患者体重计算药物剂量- 计算血糖控制指标- 其他医疗相关的数值计算"""try:# 安全的数学表达式计算return str(eval(expression))except Exception as e:return f"计算错误: {e}"# ================================
# 5. AI模型初始化模块 - 大脑配置
# ================================def initialize_llm(base_url: str):"""初始化大语言模型功能：配置Qwen模型的各项参数参数：base_url - API基础URL返回：配置好的LLM实例模型配置说明：- model: 选择qwen-max获得最佳性能- temperature: 0.1保证输出的一致性和准确性- max_tokens: 2000确保能输出完整的医疗建议"""# 创建思维链回调处理器实例thought_callback = ThoughtProcessCallbackHandler()# 初始化Qwen模型 - 通过OpenAI兼容接口llm = ChatOpenAI(model="qwen-max",  # 使用Qwen的最强模型base_url=base_url,  # 阿里云DashScope接口streaming=True,  # 开启流式输出，实时显示思考过程callbacks=[thought_callback],  # 添加思维链追踪temperature=0.1,  # 较低温度确保医疗建议的准确性max_tokens=2000,  # 足够的token数量输出完整建议timeout=30  # 30秒超时防止请求卡死)return llm, thought_callback# ================================
# 6. Agent代理模块 - 智能决策中心
# ================================def create_medical_agent(llm, tools):"""创建医疗顾问AI代理功能：整合LLM和工具，创建能够自主决策的AI代理参数：- llm: 大语言模型实例- tools: 可用工具列表返回：配置好的Agent执行器"""# 创建对话记忆 - 保持上下文连续性memory = ConversationBufferMemory(memory_key="chat_history",  # 记忆键名return_messages=True  # 返回消息格式)# 创建Agent提示模板 - AI的行为指导方针template = """你是一个专业的医疗顾问AI助手，具备查询疾病治疗方案和药物知识的能力。可用工具：
{tools}工具名称: {tool_names}对话历史:
{chat_history}当前问题: {input}工作流程规范：
当需要使用工具时，请严格遵循以下格式，一次只能执行一个动作:
思考: 详细说明你的分析过程，包括问题理解、解决策略和工具选择理由
Action: 工具名称
Action Input: 工具输入参数
Observation: 工具返回的结果当完成所有必要的工具调用后，使用以下格式给出最终答案:
思考: 综合分析所有搜索结果，说明如何得出结论
Final Answer: 完整的医疗建议医疗咨询标准流程：
1. 首先搜索疾病的基本治疗策略和临床指南
2. 然后查询该疾病的推荐用药和处方信息  
3. 对重要药物进一步查询详细的用法用量和副作用
4. 整合所有信息，提供全面的治疗建议
5. 必须声明：本建议仅供参考，不能替代专业医生的诊断和治疗重要提醒：
- 每次回应只能包含一个Action或一个Final Answer
- 必须基于搜索结果提供建议，不得编造信息
- 确保建议的准确性、全面性和安全性{agent_scratchpad}
"""# 创建提示模板实例prompt = PromptTemplate.from_template(template)# 创建ReAct类型的Agent - 能够推理和行动的智能代理agent = create_react_agent(llm, tools, prompt)# 创建Agent执行器 - 负责执行Agent的决策agent_executor = AgentExecutor(agent=agent,  # 智能代理实例tools=tools,  # 可用工具列表memory=memory,  # 对话记忆verbose=True,  # 详细输出执行过程return_intermediate_steps=True,  # 返回中间步骤max_iterations=10,  # 最大迭代次数防止无限循环early_stopping_method="generate",  # 早停策略handle_parsing_errors=True  # 处理解析错误)return agent_executor# ================================
# 7. 主控制模块 - 领导式系统协调
# ================================def run_medical_consultation(query: str, agent_executor, thought_callback):"""执行医疗咨询 - 主控制函数功能：协调整个医疗咨询流程的执行参数：- query: 用户的医疗咨询问题- agent_executor: AI代理执行器- thought_callback: 思维链回调处理器返回：咨询结果领导式规划：1. 接收用户问题2. 重置思维链收集器3. 执行AI代理咨询4. 展示详细的执行步骤5. 输出最终的医疗建议"""print(f"\n===== 用户咨询: {query} =====\n")# 重置思维链收集器 - 确保每次咨询的独立性thought_callback.thoughts = []try:# 执行AI代理咨询 - 核心业务逻辑result = agent_executor.invoke({"input": query})# 提取并展示执行步骤 - 增强系统透明度intermediate_steps = result.get("intermediate_steps", [])print("\n===== 执行步骤详情 =====")for i, step in enumerate(intermediate_steps):action, observation = stepprint(f"\n步骤 {i+1}:")print(f"使用工具: {action.tool}")print(f"输入参数: {action.tool_input}")print(f"执行结果: {observation}")# 输出最终的医疗建议print(f"\n===== 最终医疗建议 =====\n{result['output']}\n")return resultexcept Exception as e:print(f"咨询过程中出现错误: {e}")return None# ================================
# 8. 系统启动模块 - 主函数入口
# ================================def main():"""主函数 - 系统启动入口领导式规划：整个医疗顾问系统的启动和运行流程1. 环境配置2. 模型初始化  3. 工具准备4. 代理创建5. 系统启动6. 执行咨询"""print("\n========== 医疗顾问AI系统启动 ==========")print("基于Qwen API的智能医疗助手")print("功能：疾病治疗方案查询 + 药物处方建议")# 步骤1：环境配置print("\n[1/5] 正在配置系统环境...")base_url = setup_environment()# 步骤2：初始化AI模型print("[2/5] 正在初始化Qwen模型...")llm, thought_callback = initialize_llm(base_url)# 步骤3：准备医疗工具集print("[3/5] 正在准备医疗工具...")tools = [search_disease_treatment,   # 疾病治疗搜索工具search_disease_medications, # 药物处方搜索工具search_medication_details,  # 药物详情查询工具calculator                  # 数学计算工具]# 步骤4：创建医疗AI代理print("[4/5] 正在创建医疗AI代理...")agent_executor = create_medical_agent(llm, tools)# 步骤5：系统就绪，开始咨询print("[5/5] 系统启动完成，开始医疗咨询...\n")# 执行示例咨询sample_query = "患者有2型糖尿病，该如何治疗，用什么药物处方"result = run_medical_consultation(sample_query, agent_executor, thought_callback)return result# ================================
# 9. 程序入口点
# ================================if __name__ == "__main__":main()

输出结果，居然连不上网

========== 医疗顾问AI系统启动 ==========
基于Qwen API的智能医疗助手
功能：疾病治疗方案查询 + 药物处方建议

[1/5] 正在配置系统环境…
[2/5] 正在初始化Qwen模型…
[3/5] 正在准备医疗工具…
[4/5] 正在创建医疗AI代理…
[5/5] 系统启动完成，开始医疗咨询…

===== 用户咨询: 患者有2型糖尿病，该如何治疗，用什么药物处方 =====

Entering new AgentExecutor chain…
思考: 对于2型糖尿病的治疗，首先需要了解当前最新的临床指南和推荐的治疗方法。这将帮助我们理解基础的治疗原则以及可能涉及的生活方式调整等非药物干预措施。因此，我将首先使用search_disease_treatment工具来获取这些信息。
Action: search_disease_treatment
Action Input: 2型糖尿病思考: 对于2型糖尿病的治疗，首先需要了解当前最新的临床指南和推荐的治疗方法。这将帮助我们理解基础的治疗原则以及可能涉及的生活方式调整等非药物干预措施。因此，我将首先使用search_disease_treatment工具来获取这些信息。
Action: search_disease_treatment
Action Input: 2型糖尿病2型糖尿病治疗策略包括：

1. 生活方式干预：健康饮食、规律运动、体重管理
2. 药物治疗：二甲双胍为一线用药，必要时联合其他降糖药
3. 血糖监测：定期检测血糖、糖化血红蛋白
4. 并发症预防：控制血压、血脂，保护肾脏和心血管
5. 患者教育：糖尿病自我管理知识
   注：搜索服务暂时不可用，以上为基础医疗知识参考基于观察到的信息，2型糖尿病的治疗是一个综合性的过程，包括生活方式的调整和药物治疗等多方面。接下来，为了给患者提供具体的药物处方建议，我将使用search_disease_medications工具来获取针对2型糖尿病推荐使用的药物信息。
   Action: search_disease_medications
   Action Input: 2型糖尿病基于观察到的信息，2型糖尿病的治疗是一个综合性的过程，包括生活方式的调整和药物治疗等多方面。接下来，为了给患者提供具体的药物处方建议，我将使用search_disease_medications工具来获取针对2型糖尿病推荐使用的药物信息。
   Action: search_disease_medications
   Action Input: 2型糖尿病2型糖尿病常用药物：
6. 二甲双胍：一线用药，改善胰岛素敏感性
7. 磺脲类：格列齐特、格列美脲等，促进胰岛素分泌
8. DPP-4抑制剂：西格列汀、维格列汀等
9. SGLT-2抑制剂：达格列净、恩格列净等
10. GLP-1受体激动剂：利拉鲁肽、司美格鲁肽等
11. 胰岛素：根据病情需要使用
    注：搜索服务暂时不可用，以上为基础用药知识参考根据观察到的信息，对于2型糖尿病的药物治疗方案有了一个大致了解。二甲双胍作为一线用药，在大多数情况下是首选；而其他类型的降糖药如磺脲类、DPP-4抑制剂等则可以根据患者的具体情况（比如血糖控制效果、是否存在特定并发症）来选择性地添加使用。接下来，考虑到二甲双胍的重要性及其广泛的应用，我认为有必要进一步查询这种药物的具体用法用量及可能产生的副作用，以便为患者提供更加详尽且安全的用药指导。
    Action: search_medication_details
    Action Input: 二甲双胍根据观察到的信息，对于2型糖尿病的药物治疗方案有了一个大致了解。二甲双胍作为一线用药，在大多数情况下是首选；而其他类型的降糖药如磺脲类、DPP-4抑制剂等则可以根据患者的具体情况（比如血糖控制效果、是否存在特定并发症）来选择性地添加使用。接下来，考虑到二甲双胍的重要性及其广泛的应用，我认为有必要进一步查询这种药物的具体用法用量及可能产生的副作用，以便为患者提供更加详尽且安全的用药指导。
    Action: search_medication_details
    Action Input: 二甲双胍二甲双胍详细信息：
    用法用量：起始剂量500mg，每日2-3次，餐时或餐后服用，最大剂量2000-2500mg/日
    作用机制：减少肝糖输出，改善胰岛素敏感性
    副作用：胃肠道反应（恶心、腹泻、腹胀）、乳酸酸中毒（罕见）
    禁忌症：严重肾功能不全、肝功能不全、心衰、酗酒
    注意事项：定期监测肾功能
    注：搜索服务暂时不可用，以上为基础药物知识参考综合以上收集到的信息，我们可以为2型糖尿病患者提供一个较为全面的治疗建议。首先，生活方式的调整是基础，包括健康饮食、规律运动以及体重管理等非药物干预措施。其次，在药物治疗方面，二甲双胍作为一线用药被广泛推荐使用，其主要通过减少肝脏糖分输出并提高身体对胰岛素的敏感度来发挥作用。对于部分患者来说，可能还需要联合其他类型的降糖药物以达到更好的血糖控制效果。

关于二甲双胍的具体使用，通常从较低剂量开始（如500mg/次），每日服用2-3次，并且最好是在餐时或餐后服用，以减轻胃肠道不适的症状。随着治疗进程的发展，医生可能会根据患者的反应逐渐增加剂量至最大推荐量（约2000-2500mg/日）。需要注意的是，虽然二甲双胍相对安全有效，但仍有可能引起一些副作用，比如轻微的胃肠道不适；极少数情况下可能发生严重的乳酸酸中毒。因此，在使用过程中应密切关注任何不良反应，并定期检查肾功能状态。

此外，还有多种其他类型的降糖药物可供选择，例如磺脲类药物可以刺激胰腺产生更多胰岛素；DPP-4抑制剂和SGLT-2抑制剂则分别通过不同的机制帮助降低血糖水平；GLP-1受体激动剂不仅有助于血糖控制还能促进减重；当口服药物不足以控制血糖时，则需要考虑使用胰岛素治疗。

Final Answer: 对于2型糖尿病患者而言，建议采取综合性治疗方法，结合生活方式调整与合理用药。二甲双胍通常是首选药物，但具体治疗方案需由专业医师根据个人情况制定。请务必遵循医嘱服药，并注意监测可能出现的副作用。本建议仅供参考，不能替代专业医生的诊断和治疗。综合以上收集到的信息，我们可以为2型糖尿病患者提供一个较为全面的治疗建议。首先，生活方式的调整是基础，包括健康饮食、规律运动以及体重管理等非药物干预措施。其次，在药物治疗方面，二甲双胍作为一线用药被广泛推荐使用，其主要通过减少肝脏糖分输出并提高身体对胰岛素的敏感度来发挥作用。对于部分患者来说，可能还需要联合其他类型的降糖药物以达到更好的血糖控制效果。

关于二甲双胍的具体使用，通常从较低剂量开始（如500mg/次），每日服用2-3次，并且最好是在餐时或餐后服用，以减轻胃肠道不适的症状。随着治疗进程的发展，医生可能会根据患者的反应逐渐增加剂量至最大推荐量（约2000-2500mg/日）。需要注意的是，虽然二甲双胍相对安全有效，但仍有可能引起一些副作用，比如轻微的胃肠道不适；极少数情况下可能发生严重的乳酸酸中毒。因此，在使用过程中应密切关注任何不良反应，并定期检查肾功能状态。

此外，还有多种其他类型的降糖药物可供选择，例如磺脲类药物可以刺激胰腺产生更多胰岛素；DPP-4抑制剂和SGLT-2抑制剂则分别通过不同的机制帮助降低血糖水平；GLP-1受体激动剂不仅有助于血糖控制还能促进减重；当口服药物不足以控制血糖时，则需要考虑使用胰岛素治疗。

Final Answer: 对于2型糖尿病患者而言，建议采取综合性治疗方法，结合生活方式调整与合理用药。二甲双胍通常是首选药物，但具体治疗方案需由专业医师根据个人情况制定。请务必遵循医嘱服药，并注意监测可能出现的副作用。本建议仅供参考，不能替代专业医生的诊断和治疗。

Finished chain.

===== 执行步骤详情 =====

步骤 1:
使用工具: search_disease_treatment
输入参数: 2型糖尿病
执行结果: 2型糖尿病治疗策略包括：

1. 生活方式干预：健康饮食、规律运动、体重管理
2. 药物治疗：二甲双胍为一线用药，必要时联合其他降糖药
3. 血糖监测：定期检测血糖、糖化血红蛋白
4. 并发症预防：控制血压、血脂，保护肾脏和心血管
5. 患者教育：糖尿病自我管理知识
   注：搜索服务暂时不可用，以上为基础医疗知识参考

步骤 2:
使用工具: search_disease_medications
输入参数: 2型糖尿病
执行结果: 2型糖尿病常用药物：

1. 二甲双胍：一线用药，改善胰岛素敏感性
2. 磺脲类：格列齐特、格列美脲等，促进胰岛素分泌
3. DPP-4抑制剂：西格列汀、维格列汀等
4. SGLT-2抑制剂：达格列净、恩格列净等
5. GLP-1受体激动剂：利拉鲁肽、司美格鲁肽等
6. 胰岛素：根据病情需要使用
   注：搜索服务暂时不可用，以上为基础用药知识参考

步骤 3:
使用工具: search_medication_details
输入参数: 二甲双胍
执行结果: 二甲双胍详细信息：
用法用量：起始剂量500mg，每日2-3次，餐时或餐后服用，最大剂量2000-2500mg/日
作用机制：减少肝糖输出，改善胰岛素敏感性
副作用：胃肠道反应（恶心、腹泻、腹胀）、乳酸酸中毒（罕见）
禁忌症：严重肾功能不全、肝功能不全、心衰、酗酒
注意事项：定期监测肾功能
注：搜索服务暂时不可用，以上为基础药物知识参考

===== 最终医疗建议 =====
对于2型糖尿病患者而言，建议采取综合性治疗方法，结合生活方式调整与合理用药。二甲双胍通常是首选药物，但具体治疗方案需由专业医师根据个人情况制定。请务必遵循医嘱服药，并注意监测可能出现的副作用。本建议仅供参考，不能替代专业医生的诊断和治疗。

运行结果梳理

第二版

"""
深度网络调研医疗顾问AI系统
专注于在线信息获取和深度调研分析核心特性：
1. 强化网络搜索能力 - 解决DuckDuckGo连接问题
2. 多角度深度调研 - 医学文献、临床指南、药物数据库
3. 搜索结果智能分析 - 提取关键信息，去除噪音
4. 实时信息获取 - 最新治疗方案和药物信息
5. 详细搜索诊断 - 精确定位网络问题
"""# ================================
# 1. 依赖库导入和配置
# ================================import os
import time
import json
import logging
import requests
from typing import List, Dict, Any, Optional
from urllib.parse import quote
import urllib3
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry# 禁用SSL警告（调试时使用）
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)# LangChain核心库
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain.agents import AgentExecutor, create_react_agent
from langchain.prompts import PromptTemplate
from langchain.memory import ConversationBufferMemory
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler# 搜索相关
from langchain_community.tools import DuckDuckGoSearchRun
from duckduckgo_search import DDGS# ================================
# 2. 日志和调试配置
# ================================def setup_logging():"""设置详细的调试日志"""logging.basicConfig(level=logging.DEBUG,  # 设置为DEBUG级别format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',handlers=[logging.FileHandler('medical_research.log', encoding='utf-8'),logging.StreamHandler()])# 设置第三方库的日志级别logging.getLogger('urllib3').setLevel(logging.DEBUG)logging.getLogger('requests').setLevel(logging.DEBUG)return logging.getLogger(__name__)logger = setup_logging()# ================================
# 3. 网络连接诊断工具
# ================================class NetworkDiagnostic:"""网络连接诊断工具"""@staticmethoddef test_basic_connectivity():"""测试基础网络连接"""test_urls = ["https://www.google.com","https://www.bing.com", "https://duckduckgo.com","https://www.baidu.com"]print("🔍 开始网络连接诊断...")results = {}for url in test_urls:try:response = requests.get(url, timeout=10)results[url] = {'status': '✅ 连接成功','status_code': response.status_code,'response_time': f"{response.elapsed.total_seconds():.2f}s"}print(f"   {url}: ✅ {response.status_code} ({response.elapsed.total_seconds():.2f}s)")except requests.exceptions.ConnectTimeout:results[url] = {'status': '❌ 连接超时', 'error': 'Timeout'}print(f"   {url}: ❌ 连接超时")except requests.exceptions.ConnectionError as e:results[url] = {'status': '❌ 连接错误', 'error': str(e)}print(f"   {url}: ❌ 连接错误 - {e}")except Exception as e:results[url] = {'status': '❌ 其他错误', 'error': str(e)}print(f"   {url}: ❌ 其他错误 - {e}")return results@staticmethoddef test_proxy_settings():"""检测代理设置"""print("🔍 检查代理设置...")proxy_vars = ['HTTP_PROXY', 'HTTPS_PROXY', 'http_proxy', 'https_proxy']for var in proxy_vars:value = os.environ.get(var)if value:print(f"   发现代理设置: {var} = {value}")else:print(f"   {var}: 未设置")@staticmethoddef test_dns_resolution():"""测试DNS解析"""import socketprint("🔍 测试DNS解析...")hosts = ['duckduckgo.com', 'google.com', 'bing.com']for host in hosts:try:ip = socket.gethostbyname(host)print(f"   {host} -> {ip} ✅")except socket.gaierror as e:print(f"   {host} -> DNS解析失败: {e} ❌")# ================================
# 4. 强化搜索引擎类
# ================================class EnhancedSearchEngine:"""增强版搜索引擎 - 专注解决连接问题"""def __init__(self):self.session = self._create_robust_session()self.ddgs = Noneself._init_ddgs()def _create_robust_session(self):"""创建健壮的requests会话"""session = requests.Session()# 配置重试策略retry_strategy = Retry(total=3,backoff_factor=1,status_forcelist=[429, 500, 502, 503, 504],allowed_methods=["HEAD", "GET", "OPTIONS"])adapter = HTTPAdapter(max_retries=retry_strategy)session.mount("http://", adapter)session.mount("https://", adapter)# 设置请求头session.headers.update({'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36','Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8','Accept-Language': 'en-US,en;q=0.5','Accept-Encoding': 'gzip, deflate','Connection': 'keep-alive',})# 超时设置session.timeout = 30return sessiondef _init_ddgs(self):"""初始化DuckDuckGo搜索"""try:logger.info("🦆 初始化DuckDuckGo搜索引擎...")self.ddgs = DDGS(headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'},timeout=30)logger.info("✅ DuckDuckGo初始化成功")except Exception as e:logger.error(f"❌ DuckDuckGo初始化失败: {e}")self.ddgs = Nonedef search_duckduckgo_direct(self, query: str, max_results: int = 10) -> List[Dict]:"""直接使用duckduckgo-search库"""logger.info(f"🦆 使用duckduckgo-search直接搜索: {query}")try:if self.ddgs is None:self._init_ddgs()if self.ddgs is None:raise Exception("DuckDuckGo搜索引擎初始化失败")# 执行搜索results = list(self.ddgs.text(keywords=query,max_results=max_results,region='us-en',safesearch='moderate'))logger.info(f"✅ DuckDuckGo搜索成功，获得 {len(results)} 个结果")return resultsexcept Exception as e:logger.error(f"❌ DuckDuckGo直接搜索失败: {e}")return []def search_with_fallback(self, query: str) -> str:"""带降级策略的搜索"""logger.info(f"🔍 开始搜索: {query}")# 方法1: 直接使用duckduckgo-searchresults = self.search_duckduckgo_direct(query)if results:return self._format_search_results(results, "DuckDuckGo Direct")# 方法2: 使用LangChain的DuckDuckGoSearchRuntry:logger.info("🔄 尝试LangChain DuckDuckGoSearchRun...")search_tool = DuckDuckGoSearchRun()result = search_tool.run(query)if result and "No results found" not in result:logger.info("✅ LangChain DuckDuckGo搜索成功")return f"📋 搜索结果 (LangChain DuckDuckGo):\n{result}"except Exception as e:logger.error(f"❌ LangChain DuckDuckGo搜索失败: {e}")# 方法3: 使用自定义HTTP请求try:logger.info("🔄 尝试自定义HTTP搜索...")return self._custom_search_request(query)except Exception as e:logger.error(f"❌ 自定义搜索失败: {e}")# 所有方法都失败logger.error("❌ 所有搜索方法均失败")return "❌ 网络搜索暂时不可用，请检查网络连接设置。建议：\n1. 检查防火墙设置\n2. 确认代理配置\n3. 测试DNS解析\n4. 联系网络管理员"def _format_search_results(self, results: List[Dict], source: str) -> str:"""格式化搜索结果"""if not results:return "未找到相关搜索结果"formatted = f"📋 搜索结果 ({source}):\n\n"for i, result in enumerate(results[:5], 1):  # 只显示前5个结果title = result.get('title', '无标题')body = result.get('body', result.get('snippet', '无摘要'))href = result.get('href', '无链接')formatted += f"{i}. **{title}**\n"formatted += f"   {body[:200]}...\n"formatted += f"   🔗 {href}\n\n"return formatteddef _custom_search_request(self, query: str) -> str:"""自定义HTTP搜索请求"""logger.info("🔄 执行自定义HTTP搜索请求...")# 构建搜索URLencoded_query = quote(query)search_url = f"https://duckduckgo.com/html/?q={encoded_query}"try:response = self.session.get(search_url, timeout=30)response.raise_for_status()# 简单的HTML解析（这里可以用BeautifulSoup进一步优化）content = response.textif "No results found" in content:return "未找到相关搜索结果"logger.info("✅ 自定义HTTP搜索成功")return f"📋 搜索成功 (自定义HTTP):\n搜索查询: {query}\n状态: 已获取搜索页面\n建议: 请查看完整搜索结果以获取详细信息"except Exception as e:logger.error(f"❌ 自定义HTTP搜索失败: {e}")raise# 全局搜索引擎实例
search_engine = EnhancedSearchEngine()# ================================
# 5. 深度调研工具集
# ================================@tool
def deep_medical_research(topic: str) -> str:"""深度医疗调研工具针对医疗主题进行多角度深度调研包括：临床指南、最新研究、治疗方案、药物信息"""logger.info(f"🔬 开始深度医疗调研: {topic}")# 构建多个专业搜索查询research_queries = [f"{topic} 临床治疗指南 2024 2025",f"{topic} 最新研究 clinical trial",f"{topic} 药物治疗 medication treatment", f"{topic} 诊疗规范 medical guidelines",f"{topic} 循证医学 evidence based medicine"]all_results = []for i, query in enumerate(research_queries, 1):logger.info(f"📍 执行第 {i} 个调研查询: {query}")result = search_engine.search_with_fallback(query)if result and "❌" not in result:all_results.append(f"🔍 调研角度 {i}: {query}\n{result}\n")logger.info(f"✅ 第 {i} 个查询成功")else:logger.warning(f"⚠️ 第 {i} 个查询失败")# 添加延迟避免频率限制time.sleep(2)if all_results:combined_result = f"📊 {topic} 深度调研报告:\n\n" + "\n".join(all_results)logger.info(f"✅ 深度调研完成，共获得 {len(all_results)} 份有效结果")return combined_resultelse:logger.error("❌ 深度调研失败")return f"❌ 无法完成 {topic} 的深度调研，请检查网络连接"@tool  
def search_latest_medical_guidelines(disease: str) -> str:"""搜索最新医疗指南和临床规范"""logger.info(f"📚 搜索最新医疗指南: {disease}")queries = [f"{disease} clinical guidelines 2024 2025",f"{disease} treatment protocol latest",f"{disease} 诊疗指南 国家标准",f"{disease} WHO guidelines recommendations"]for query in queries:result = search_engine.search_with_fallback(query)if result and "❌" not in result:return f"📚 {disease} 最新医疗指南:\n{result}"time.sleep(1)return f"❌ 无法获取 {disease} 的最新医疗指南"@tool
def search_drug_information(drug_name: str) -> str:"""搜索药物详细信息和最新资料"""logger.info(f"💊 搜索药物信息: {drug_name}")queries = [f"{drug_name} 药物说明书 用法用量",f"{drug_name} side effects contraindications",f"{drug_name} clinical pharmacology",f"{drug_name} drug interactions warnings"]for query in queries:result = search_engine.search_with_fallback(query)if result and "❌" not in result:return f"💊 {drug_name} 详细药物信息:\n{result}"time.sleep(1)return f"❌ 无法获取 {drug_name} 的详细信息"@tool
def search_medical_research(topic: str) -> str:"""搜索最新医学研究和临床试验"""logger.info(f"🔬 搜索医学研究: {topic}")queries = [f"{topic} clinical trial 2024 2025",f"{topic} latest research pubmed",f"{topic} systematic review meta-analysis",f"{topic} 最新研究 临床试验"]for query in queries:result = search_engine.search_with_fallback(query)if result and "❌" not in result:return f"🔬 {topic} 最新医学研究:\n{result}"time.sleep(1)return f"❌ 无法获取 {topic} 的最新研究信息"@tool
def network_diagnostic_tool() -> str:"""网络诊断工具"""logger.info("🔧 执行网络诊断...")print("\n" + "="*50)print("🔧 网络连接诊断报告")print("="*50)# 基础连接测试connectivity_results = NetworkDiagnostic.test_basic_connectivity()# 代理设置检查NetworkDiagnostic.test_proxy_settings()# DNS解析测试NetworkDiagnostic.test_dns_resolution()# 搜索引擎测试print("\n🔍 搜索引擎连接测试...")test_query = "医学测试"result = search_engine.search_with_fallback(test_query)if "❌" not in result:print("   搜索引擎: ✅ 工作正常")status = "✅ 网络连接正常，搜索功能可用"else:print("   搜索引擎: ❌ 连接失败")status = "❌ 网络连接存在问题，搜索功能不可用"print("="*50)return status# ================================
# 6. 环境配置
# ================================def setup_environment():"""环境配置"""logger.info("🔧 配置系统环境...")load_dotenv()# API配置api_key = os.getenv("QWEN_API_KEY", "sk-025a68f2bf98489c9e4fce2fc0c9f0c2")os.environ["OPENAI_API_KEY"] = api_keybase_url = os.getenv("QWEN_BASE_URL", "https://dashscope.aliyuncs.com/compatible-mode/v1")logger.info("✅ 环境配置完成")return base_url# ================================
# 7. AI模型和Agent
# ================================class MedicalResearchCallbackHandler(StreamingStdOutCallbackHandler):"""医疗调研回调处理器"""def __init__(self):super().__init__()self.search_count = 0def on_llm_new_token(self, token: str, **kwargs) -> None:print(token, end="", flush=True)if "搜索" in token or "search" in token.lower():self.search_count += 1def initialize_llm(base_url: str):"""初始化LLM"""logger.info("🤖 初始化Qwen模型...")callback = MedicalResearchCallbackHandler()llm = ChatOpenAI(model="qwen-max",base_url=base_url,streaming=True,callbacks=[callback],temperature=0.1,max_tokens=4000,timeout=120)logger.info("✅ Qwen模型初始化成功")return llm, callbackdef create_research_agent(llm, tools):"""创建深度调研Agent"""logger.info("🤖 创建深度调研Agent...")memory = ConversationBufferMemory(memory_key="chat_history",return_messages=True)# 专注于深度调研的提示模板template = """你是一个专业的医疗信息调研AI，专门进行深度的在线医疗信息搜索和分析。🛠️ 可用工具：
{tools}工具名称: {tool_names}📝 对话历史:
{chat_history}❓ 调研任务: {input}🔍 深度调研标准流程：
当收到医疗咨询任务时，你必须按以下步骤执行深度在线调研：1. 🔬 **深度医疗调研**: 使用deep_medical_research工具进行多角度调研
2. 📚 **最新指南搜索**: 使用search_latest_medical_guidelines获取权威指南
3. 💊 **药物信息搜索**: 使用search_drug_information获取详细药物资料
4. 🔬 **最新研究搜索**: 使用search_medical_research获取前沿研究信息
5. 📊 **综合分析**: 整合所有在线搜索结果，提供基于最新信息的专业建议🎯 调研质量要求：
- 必须基于在线搜索的实时信息
- 不得使用任何预设的本地知识
- 每个工具都要使用，确保信息全面性
- 如果搜索失败，使用network_diagnostic_tool诊断问题
- 提供信息来源和时效性说明⚠️ 重要原则：
- 所有建议必须基于搜索到的最新在线信息
- 强调信息的时效性和来源可靠性
- 建议用户咨询专业医生进行个性化诊疗使用ReAct格式，每次只执行一个动作：
思考: [分析当前情况，选择下一步工具]
Action: [工具名称]
Action Input: [具体参数]
Observation: [工具返回结果]最终必须用以下格式提供综合分析：
Final Answer: [基于所有搜索结果的综合医疗建议]{agent_scratchpad}
"""prompt = PromptTemplate.from_template(template)agent = create_react_agent(llm, tools, prompt)agent_executor = AgentExecutor(agent=agent,tools=tools,memory=memory,verbose=True,return_intermediate_steps=True,max_iterations=20,  # 增加迭代次数支持深度调研early_stopping_method="generate",handle_parsing_errors=True,max_execution_time=600  # 10分钟超时)logger.info("✅ 深度调研Agent创建成功")return agent_executor# ================================
# 8. 主控制函数
# ================================def run_deep_medical_research(query: str, agent_executor, callback):"""执行深度医疗调研"""logger.info(f"🏥 开始深度医疗调研: {query}")print(f"\n===== 🔬 深度医疗调研任务: {query} =====\n")# 重置计数器callback.search_count = 0start_time = time.time()try:# 执行深度调研result = agent_executor.invoke({"input": query})execution_time = time.time() - start_time# 显示调研统计intermediate_steps = result.get("intermediate_steps", [])print("\n===== 📊 调研执行统计 =====")print(f"🔧 执行工具数量: {len(intermediate_steps)}")print(f"🔍 搜索操作次数: {callback.search_count}")print(f"⏱️ 总调研时间: {execution_time:.2f} 秒")# 显示执行步骤print("\n===== 🔍 调研步骤详情 =====")for i, step in enumerate(intermediate_steps):action, observation = stepprint(f"\n🔧 步骤 {i+1}: {action.tool}")print(f"   📥 输入: {action.tool_input}")# 只显示观察结果的前200字符避免输出过长obs_preview = str(observation)[:200] + "..." if len(str(observation)) > 200 else str(observation)print(f"   📤 结果: {obs_preview}")# 最终调研报告print(f"\n===== 🏥 深度调研报告 =====")print(result.get('output', '未能生成调研报告'))logger.info("✅ 深度医疗调研完成")return resultexcept Exception as e:execution_time = time.time() - start_timelogger.error(f"❌ 调研失败: {e}")print(f"\n❌ 调研过程出现错误: {e}")print(f"⏱️ 错误前耗时: {execution_time:.2f} 秒")return Nonedef main():"""主函数"""print("""
🔬 ============================================深度网络调研医疗顾问AI系统 v3.0
🔬 ============================================
🌐 专注在线信息获取和实时医疗调研
🔍 解决网络搜索问题，提供最新医疗信息
🏥 支持多角度深度调研和专业分析""")try:# 系统初始化print("🔧 [1/5] 配置系统环境...")base_url = setup_environment()print("🔍 [2/5] 执行网络诊断...")NetworkDiagnostic.test_basic_connectivity()print("🤖 [3/5] 初始化AI模型...")llm, callback = initialize_llm(base_url)print("🛠️ [4/5] 准备调研工具...")tools = [deep_medical_research,search_latest_medical_guidelines,search_drug_information,search_medical_research,network_diagnostic_tool]print("🤖 [5/5] 创建调研Agent...")agent_executor = create_research_agent(llm, tools)print("\n✅ 系统启动完成！开始深度调研演示...")# 执行深度调研示例sample_query = "2型糖尿病的最新治疗方案和药物选择，需要最新的临床指南和研究证据"result = run_deep_medical_research(sample_query, agent_executor, callback)return resultexcept Exception as e:logger.error(f"系统启动失败: {e}")print(f"❌ 系统启动失败: {e}")return Noneif __name__ == "__main__":main()

结果测试

这个版本去掉了知识库，只使用联网搜索，但调用联网搜索模块还是不成功。

为什么，autodl 上不能调用联网工具，不知道怎么回事。