pip install langchain openai python-dotenv tiktoken

OPENAI_API_KEY=your_api_key_here

from langchain.chains import LLMChain
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
from dotenv import load_dotenv
import os

# 加载环境变量
load_dotenv()

# 初始化LLM
llm = OpenAI(temperature=0.3)  # temperature控制回答的创造性

# 定义医疗提示模板
medical_template = """
你是一位专业的医疗助手，请根据患者的描述提供专业但易懂的建议。
注意：你的回答不应替代专业医生的诊断。

患者问题: {question}

医疗建议:
"""

prompt = PromptTemplate(
    input_variables=["question"],
    template=medical_template,
)

# 创建问答链
medical_chain = LLMChain(llm=llm, prompt=prompt)

# 测试问答
question = "我最近三天持续头痛，伴有轻微发烧，应该怎么办？"
response = medical_chain.run(question=question)
print(response)

from langchain.chains import SequentialChain

# 症状分类器模板
classifier_template = """
分析以下患者描述的症状属于哪个科室范畴:
1. 内科  2. 外科  3. 儿科  4. 妇科  5. 急诊科

症状描述: {symptoms}
科室编号:
"""

classifier_prompt = PromptTemplate(
    input_variables=["symptoms"],
    template=classifier_template,
)

classifier_chain = LLMChain(
    llm=llm,
    prompt=classifier_prompt,
    output_key="department"
)

# 组合链式流程
overall_chain = SequentialChain(
    chains=[classifier_chain, medical_chain],
    input_variables=["question"],
    output_variables=["department", "text"],
    verbose=True
)

response = overall_chain.run("我怀孕6周，最近有轻微腹痛")
print(f"科室建议: {response['department']}")
print(f"医疗建议: {response['text']}")

Q:感冒有哪些症状?
A:常见症状包括鼻塞、流涕、咳嗽、喉咙痛、发热等...

Q:高血压的标准是什么?
A:收缩压≥140mmHg和/或舒张压≥90mmHg...

from langchain.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS

# 加载本地知识库
loader = TextLoader("medical_knowledge.txt")
documents = loader.load()

# 分割文本
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
docs = text_splitter.split_documents(documents)

# 创建向量数据库存储知识库内容并实现语义搜索功能  
embeddings = OpenAIEmbeddings()
db = FAISS.from_documents(docs, embeddings)

def get_similar_knowledge(query):
    docs = db.similarity_search(query)
    return "\n".join([d.page_content for d in docs])

# Augment the original prompt with retrieved knowledge  
augmented_template = """根据以下医学知识和患者问题提供建议:

相关医学知识:
{knowledge}

患者问题: {question}

医疗建议:
"""

augmented_prompt = PromptTemplate(
    input_variables=["knowledge", "question"],
    template=augmented_template,
)

knowledge_chain = LLMChain(llm=llm, prompt=augmented_prompt)

def enhanced_answer(question):
    knowledge = get_similar_knowledge(question)
    return knowledge_chain.run(knowledge=knowledge, question=question)

print(enhanced_answer("高血压的标准值是多少?"))

from typing import List  
import re  

def validate_input(text: str) -> bool:  
    """验证用户输入是否合法"""  

     #检查是否有明显的药物滥用或自残倾向表述  
     danger_keywords = ["自杀","过量服用","自残"]  
     if any(keyword in text for keyword in danger_keywords):  
         return False  

     #检查是否有明确的个人信息泄露风险    
     personal_info_patterns = [  
         r"\d{18}|\d{17}X", #身份证号   
         r"1[3-9]\d{9}", #手机号   
         r"\d{6,8}" #可能包含密码   
     ]  

     if any(re.search(pattern, text) for pattern in personal_info_patterns):  
         return False  

     return True  

def safety_filter(response: str) -> str:  
     """对输出进行安全过滤"""  

     免责声明="\n\n注意：本建议仅供参考，具体诊疗请遵医嘱"  

      if "癌症" in response or "肿瘤" in response:  
          response += "\n[重要]疑似重大疾病建议立即就医检查" +免责声明 

      elif any(word in response for word in ["手术","住院"]):  
          response += "\n[提示]需要专业医疗机构评估"+免责声明 

      return response  

#整合到主流程中   
def safe_medical_query(question):  
     if not validate_input(question):  
         return "您的问题涉及敏感内容，建议直接咨询专业医生"  

     response = enhanced_answer(question) 
     return safety_filter(response)  

print(safe_medical_query("我的身份证是110101199003076666")) #测试隐私保护   
print(safe_medical_query("我最近咳血怎么办")) #测试重大疾病提醒   

from langchain.cache import InMemoryCache   
import langchain   

#启用内存缓存    
langchain.llm_cache=InMemoryCache()    

#首次查询会调用API    
response1=medical_chain.run("感冒吃什么药?") 

#相同问题直接从缓存读取    
response2=medical_chain.run("感冒吃什么药?")    

import asyncio   

async def async_query(questions:List[str]):    
   tasks=[medical_chain.arun(q)for q in questions]    
   return await asyncio.gather(*tasks)    

questions=[
   "头痛怎么办",
   "腹泻吃什么药",
   "血糖高要注意什么"
]    

results=asyncio.run(async_query(questions))    
for q,r in zip(questions,results):    
   print(f"Q:{q}\nA:{r}\n")