FastAPI + Gemini 构建生产级医疗 AI 助手

完整指南：从零到生产

为什么这个项目很重要

喀麦隆的问题

• 医疗服务获取有限 — 许多农村地区缺乏医生
• 语言障碍 — 英语和法语使用者需要支持
• 医学素养 — 人们难以理解医学报告
• 成本问题 — 医疗咨询费用昂贵

我们的解决方案

一个免费的、AI 驱动的医疗助手，可以离线工作（部署后），理解英语和法语，简化医学术语解释，提供基于证据的信息，并尊重隐私。

LangChain 帮助我们链接多个 AI 操作，使用模板实现一致的提示词，获得结构化输出（JSON），并优雅地处理重试和错误。

项目结构

backend/
├── app/
│   ├── __init__.py
│   ├── main.py                 # FastAPI 应用入口
│   ├── config.py               # 环境变量与设置
│   ├── routes/
│   │   ├── health.py           # 健康检查端点
│   │   ├── analysis.py         # 医学分析端点
│   │   └── research.py         # 研究端点
│   ├── services/
│   │   ├── gemini_service.py   # Gemini Vision 操作
│   │   └── tavily_service.py   # 医学研究
│   ├── chains/
│   │   ├── chat_chain.py       # LangChain 聊天流
│   │   └── analysis_chain.py   # LangChain 分析流
│   └── models/
│       └── schemas.py          # Pydantic 数据模型
├── requirements.txt
├── .env.example
└── README.md

前置要求

开始前，确保你已安装：

• Python 3.11+
• pip（Python 包管理器）
• 文本编辑器（推荐 VS Code）
• API 密钥（均免费）：
  • Google Gemini API 密钥
  • Tavily API 密钥

第 1 步：创建项目结构

mkdir medicare-ai
cd medicare-ai
mkdir backend
cd backend
mkdir -p app/routes app/services app/chains app/models

第 2 步：创建虚拟环境

python -m venv venv

# 激活虚拟环境
# Windows:
venv\Scripts\activate
# Mac/Linux:
source venv/bin/activate

第 3 步：安装依赖

创建 requirements.txt：

fastapi
uvicorn
python-dotenv
python-multipart
pydantic
pydantic-settings
langchain
langchain-core
langchain-google-genai
tavily-python
Pillow

pip install --upgrade pip
pip install -r requirements.txt

理解环境变量

环境变量将敏感信息（API 密钥）与代码分离。这对以下方面至关重要：

• 安全性 — API 密钥永远不会进入版本控制
• 灵活性 — 无需修改代码即可更改设置
• 部署 — 为开发/测试/生产环境提供不同的配置

我们使用 Pydantic Settings 而不是 os.getenv()，因为它提供：

• 自动类型验证
• 内置默认值
• 字段描述
• 完整的 IDE 自动完成支持
• 详细的错误信息

在 backend/ 文件夹中创建 .env：

GOOGLE_API_KEY=AIzaSy...your_key_here
TAVILY_API_KEY=tvly-...your_key_here
HOST=0.0.0.0
PORT=8000
CORS_ORIGINS=http://localhost:3000,http://127.0.0.1:3000
GEMINI_MODEL=gemini-2.0-flash-exp
TEMPERATURE=0.7
MAX_TOKENS=2048

重要：将 .env 添加到你的 .gitignore：

echo ".env" >> .gitignore

创建 .env.example（用于文档）：

GOOGLE_API_KEY=your_gemini_api_key_here
TAVILY_API_KEY=your_tavily_api_key_here
HOST=0.0.0.0
PORT=8000
CORS_ORIGINS=http://localhost:3000
GEMINI_MODEL=gemini-2.0-flash-exp
TEMPERATURE=0.7
MAX_TOKENS=2048

配置管理

创建 app/config.py：

该文件使用 Pydantic Settings 管理所有设置并创建 AI 模型实例，以便进行自动验证和类型安全。

import os
from pydantic_settings import BaseSettings
from pydantic import Field
from langchain_google_genai import ChatGoogleGenerativeAI
from functools import lru_cache

class Settings(BaseSettings):
    google_api_key: str = Field(..., description="Google Gemini API key")
    tavily_api_key: str = Field(..., description="Tavily API key")
    host: str = Field(default="0.0.0.0")
    port: int = Field(default=8000)
    cors_origins: str = Field(default="http://localhost:3000")
    gemini_model: str = Field(default="gemini-2.0-flash-exp")
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)
    max_tokens: int = Field(default=2048, ge=100, le=8192)
    max_file_size: int = Field(default=10 * 1024 * 1024)

    class Config:
        env_file = ".env"
        case_sensitive = False

    @property
    def cors_origins_list(self):
        return [origin.strip() for origin in self.cors_origins.split(",")]

settings = Settings()

@lru_cache()
def load_google_llm():
    return ChatGoogleGenerativeAI(
        model=settings.gemini_model,
        google_api_key=settings.google_api_key,
        temperature=settings.temperature,
        max_output_tokens=settings.max_tokens,
        convert_system_message_to_human=True
    )

@lru_cache()
def load_google_vision_llm():
    return ChatGoogleGenerativeAI(
        model=settings.gemini_model,
        google_api_key=settings.google_api_key,
        temperature=0.5,
        max_output_tokens=settings.max_tokens,
        convert_system_message_to_human=True
    )

这里发生了什么：Pydantic Settings 自动读取 .env 文件，验证字段类型和范围，而 @lru_cache 装饰器创建模型一次并重复使用以获得更好的性能。我们有两个 LLM 函数：一个用于聊天，温度较高（更具创意），一个用于视觉，温度较低（文本提取更一致）。

使用 Pydantic 定义数据模型

创建 app/models/schemas.py：

这些模型定义了传入请求和传出响应的数据形状。FastAPI 使用这些模型进行自动验证和文档生成。

from pydantic import BaseModel, Field
from datetime import datetime

class HealthCheckResponse(BaseModel):
    status: str
    timestamp: datetime
    message: str

class ChatRequest(BaseModel):
    message: str = Field(..., min_length=1, max_length=1000)
    language: str = Field(default="en", pattern="^(en|fr)$")

class ChatResponse(BaseModel):
    response: str
    language: str
    timestamp: datetime

class AnalysisRequest(BaseModel):
    text: str = Field(..., min_length=1)
    context: str = Field(default="")
    language: str = Field(default="en")

class MedicalAnalysis(BaseModel):
    summary: str
    key_findings: list[str]
    recommendations: list[str]
    next_steps: list[str]

class AnalysisResponse(BaseModel):
    summary: str
    key_findings: list[str]
    recommendations: list[str]
    next_steps: list[str]
    disclaimer: str
    language: str
    timestamp: datetime

class ImageAnalysisResponse(BaseModel):
    extracted_text: str
    analysis: AnalysisResponse

class ResearchRequest(BaseModel):
    query: str = Field(..., min_length=3, max_length=200)
    max_results: int = Field(default=5, ge=1, le=10)
    language: str = Field(default="en")

class ResearchResult(BaseModel):
    title: str
    url: str
    content: str
    score: float

class ResearchResponse(BaseModel):
    query: str
    results: list[ResearchResult]
    summary: str
    timestamp: datetime

关键点：

• Field 验证（min_length、max_length、pattern）由 FastAPI 自动检查
• MedicalAnalysis 模型被 LangChain 的 PydanticOutputParser 使用，以强制从 AI 获得结构化 JSON 输出
• 嵌套模型（如包含 AnalysisResponse 的 ImageAnalysisResponse）允许复杂的数据结构

创建聊天链

创建 app/chains/chat_chain.py：

这实现了使用 LangChain Expression Language（LCEL）的聊天链。该链以清晰、可读的方式连接提示词模板、LLM 和输出解析器。

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from app.config import load_google_llm

def create_chat_chain(language: str = "en"):
    llm = load_google_llm()

    if language == "fr":
        system_message = """Vous êtes MediCare AI, un assistant médical IA pour le Cameroun.

Vos responsabilités:
- Fournir des informations médicales précises et basées sur des preuves
- Expliquer les concepts médicaux en termes simples
- Toujours recommander de consulter un professionnel de santé qualifié
- Être culturellement sensible au contexte camerounais

IMPORTANT: Vous n'êtes PAS un médecin. Ne donnez jamais de diagnostic définitif."""
    else:
        system_message = """You are MediCare AI, a medical AI assistant for Cameroon.

Your responsibilities:
- Provide accurate, evidence-based medical information
- Explain medical concepts in simple terms
- Always recommend consulting a qualified healthcare professional
- Be culturally sensitive to Cameroon's context

IMPORTANT: You are NOT a doctor. Never give definitive diagnoses."""

    prompt = ChatPromptTemplate.from_messages([
        ("system", system_message),
        ("human", "{message}")
    ])

    chain = prompt | llm | StrOutputParser()
    return chain

您的职责：

• 提供准确、循证的医学信息
• 以简明易懂的方式解释医学概念
• 始终建议咨询合格的医疗专业人士
• 对喀麦隆背景保持文化敏感性

重要提示：您不是医生。不要提供明确的诊断。"""

prompt = ChatPromptTemplate.from_messages([ ("system", system_message), ("user", "{user_question}") ])

parser = StrOutputParser() chain = prompt | llm | parser

return chain

def get_chat_response(message: str, language: str = "en"): chain = create_chat_chain(language) response = chain.invoke({"user_question": message}) return response

LCEL 说明：管道运算符 `prompt | llm | parser` 创建一条链，其中每一步的输出成为下一步的输入。这比手动调用每个组件更清晰，并提供内置的异步支持、错误处理和流式传输功能。

创建 app/chains/analysis_chain.py：

此链使用 PydanticOutputParser 生成结构化 JSON 输出，强制 LLM 输出与我们的 MedicalAnalysis 模型匹配的数据。

from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import PydanticOutputParser from app.config import load_google_llm from app.models.schemas import MedicalAnalysis

def create_analysis_chain(language: str = "en"): llm = load_google_llm() parser = PydanticOutputParser(pydantic_object=MedicalAnalysis) format_instructions = parser.get_format_instructions()

if language == "fr": system_message = """Vous êtes un assistant médical IA analysant des dossiers médicaux. Fournissez des informations claires, précises et actionnables. Restez objectif et recommandez toujours une consultation médicale professionnelle."""

user_template = """Analysez ce dossier médical et fournissez une analyse structurée:

Dossier Médical: {medical_text}

Contexte Additionnel: {context}

{format_instructions}

Répondez UNIQUEMENT en JSON valide.""" else: system_message = """You are a medical AI assistant analyzing medical records. Provide clear, accurate, and actionable insights. Stay objective and always recommend professional medical consultation."""

user_template = """Analyze this medical record and provide a structured analysis:

Medical Record: {medical_text}

Additional Context: {context}

{format_instructions}

Respond ONLY with valid JSON."""

prompt = ChatPromptTemplate.from_messages([ ("system", system_message), ("user", user_template) ])

prompt = prompt.partial(format_instructions=format_instructions) chain = prompt | llm | parser

return chain

def analyze_medical_record(text: str, context: str = "", language: str = "en"): chain = create_analysis_chain(language)

try: result = chain.invoke({ "medical_text": text, "context": context if context else "No additional context provided" }) return result except Exception as e: print(f"Analysis error: {e}") return MedicalAnalysis( summary=f"Analysis completed but encountered formatting issues: {str(e)[:200]}", key_findings=["Analysis was performed but results need manual review"], recommendations=["Consult with a healthcare professional for detailed interpretation"], next_steps=["Schedule appointment with your doctor", "Keep this record for your medical history"] )

PydanticOutputParser 的妙处：它自动生成详细的 JSON Schema 指令，告诉 LLM 如何格式化其响应。解析器随后验证输出并将其转换为 Python 对象。try-except 块确保即使 JSON 解析失败，我们也总能返回一些有用的东西。

创建 app/services/gemini_service.py：

该服务处理特定于视觉的任务，如从医疗记录图像中提取文本。

from langchain_core.messages import HumanMessage from app.config import load_google_vision_llm import base64 import json

class GeminiService: def init(self): self.vision_llm = load_google_vision_llm()

def extract_text_from_image(self, image_bytes: bytes): try: image_b64 = base64.b64encode(image_bytes).decode('utf-8')

extraction_prompt = """You are a medical text extractor. Extract ALL text from this medical document/record.

Include:

• Patient information
• Test results
• Doctor's notes
• Prescriptions
• Dates and measurements
• Any handwritten text

Format the output clearly and preserve the structure. If text is unclear, indicate with [unclear].

Extract all text now:"""

message = HumanMessage( content=[ {"type": "text", "text": extraction_prompt}, {"type": "image_url", "image_url": f"data:image/jpeg;base64,{image_b64}"} ] )

response = self.vision_llm.invoke([message]) return response.content

except Exception as e: raise Exception(f"Image text extraction error: {str(e)}")

def analyze_image_directly(self, image_bytes: bytes, language: str = "en"): try: image_b64 = base64.b64encode(image_bytes).decode('utf-8')

if language == "fr": prompt = """Analysez cette image de dossier médical et fournissez une analyse au format JSON avec ces clés:

• summary: Aperçu bref de ce que vous voyez
• key_findings: Liste des résultats importants
• recommendations: Recommandations de santé
• next_steps: Actions suggérées

Répondez UNIQUEMENT en JSON valide.""" else: prompt = """Analyze this medical record image and provide analysis in JSON format with these keys:

• summary: Brief overview of what you see
• key_findings: List of important findings
• recommendations: Health recommendations
• next_steps: Suggested action

Respond ONLY with valid JSON."""

message = HumanMessage( content=[ {"type": "text", "text": prompt}, {"type": "image_url", "image_url": f"data:image/jpeg;base64,{image_b64}"} ] )

response = self.vision_llm.invoke([message]) result = json.loads(response.content) return result

except json.JSONDecodeError: return { "summary": response.content[:500], "key_findings": ["Analysis completed - see summary"], "recommendations": ["Consult with a healthcare professional"], "next_steps": ["Schedule appointment with your doctor"] } except Exception as e: raise Exception(f"Image analysis error: {str(e)}")

gemini_service = GeminiService()

为什么用 base64？LangChain 要求图像采用 base64 格式，这是将二进制数据编码为文本的标准方法。使用 `type: "image_url"` 的 HumanMessage 格式是 LangChain 处理多模态内容（文本 + 图像）的方式。

创建 app/services/tavily_service.py：

该服务使用 Tavily 的 AI 驱动搜索引擎处理医学研究搜索。

from tavily import TavilyClient from app.config import settings

class TavilyService: def init(self): self.client = TavilyClient(api_key=settings.tavily_api_key)

def search_medical_research(self, query: str, max_results: int = 5): try: response = self.client.search( query=f"medical research {query}", search_depth="advanced", max_results=max_results, include_domains=[ "pubmed.ncbi.nlm.nih.gov", "nih.gov", "who.int", "cdc.gov", "mayoclinic.org", "webmd.com", "healthline.com", "medicalnewstoday.com" ] ) return response except Exception as e: raise Exception(f"Research search error: {str(e)}")

def format_results(self, raw_results): formatted = [] for result in raw_results.get("results", []): formatted.append({ "title": result.get("title", "Untitled"), "url": result.get("url", ""), "content": result.get("content", "")[:500], "score": result.get("score", 0.0) }) return formatted

tavily_service = TavilyService()

域名过滤：通过指定 `include_domains`，我们确保结果只来自受信任的医学来源，避免博客、论坛或不可靠的网站。

创建 app/routes/health.py：

from fastapi import APIRouter from app.models.schemas import HealthCheckResponse from datetime import datetime

router = APIRouter(prefix="/api", tags=["Health"])

@router.get("/health", response_model=HealthCheckResponse)
async def health_check():
    return HealthCheckResponse(
        status="healthy",
        timestamp=datetime.now(),
        message="MediCare AI Backend is running!"
    )

创建 app/routes/analysis.py：

from fastapi import APIRouter, UploadFile, File, Form, HTTPException
from app.models.schemas import (
    ChatRequest, ChatResponse,
    AnalysisRequest, AnalysisResponse,
    ImageAnalysisResponse
)
from app.chains.chat_chain import get_chat_response
from app.chains.analysis_chain import analyze_medical_record
from app.services.gemini_service import gemini_service
from datetime import datetime

router = APIRouter(prefix="/api", tags=["Analysis"])

@router.post("/chat", response_model=ChatResponse)
async def chat_with_ai(request: ChatRequest):
    try:
        response_text = get_chat_response(
            message=request.message,
            language=request.language
        )
        return ChatResponse(
            response=response_text,
            language=request.language,
            timestamp=datetime.now()
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Chat error: {str(e)}")

@router.post("/analyze-text", response_model=AnalysisResponse)
async def analyze_medical_text(request: AnalysisRequest):
    try:
        analysis = analyze_medical_record(
            text=request.text,
            context=request.context,
            language=request.language
        )

disclaimer = (
            "This analysis is for informational purposes only. "
            "Always consult qualified healthcare professionals for medical advice."
        )

return AnalysisResponse(
            summary=analysis.summary,
            key_findings=analysis.key_findings,
            recommendations=analysis.recommendations,
            next_steps=analysis.next_steps,
            disclaimer=disclaimer,
            language=request.language,
            timestamp=datetime.now()
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Analysis error: {str(e)}")

@router.post("/analyze-image", response_model=ImageAnalysisResponse)
async def analyze_medical_image(
    file: UploadFile = File(...),
    language: str = Form(default="en"),
    extract_text_only: bool = Form(default=False)
):
    if not file.content_type.startswith("image/"):
        raise HTTPException(status_code=400, detail="File must be an image")

try:
        image_bytes = await file.read()
        extracted_text = gemini_service.extract_text_from_image(image_bytes)

if extract_text_only:
            return ImageAnalysisResponse(
                extracted_text=extracted_text,
                analysis=AnalysisResponse(
                    summary="Text extraction completed",
                    key_findings=[],
                    recommendations=[],
                    next_steps=["Review the extracted text", "Analyze if needed"],
                    disclaimer="Text extraction only - no analysis performed",
                    language=language,
                    timestamp=datetime.now()
                )
            )

analysis = analyze_medical_record(text=extracted_text, language=language)
        disclaimer = (
            "This analysis is for informational purposes only. "
            "Always consult qualified healthcare professionals for medical advice."
        )

return ImageAnalysisResponse(
            extracted_text=extracted_text,
            analysis=AnalysisResponse(
                summary=analysis.summary,
                key_findings=analysis.key_findings,
                recommendations=analysis.recommendations,
                next_steps=analysis.next_steps,
                disclaimer=disclaimer,
                language=language,
                timestamp=datetime.now()
            )
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Image analysis error: {str(e)}")

@router.post("/extract-text")
async def extract_text_from_image(file: UploadFile = File(...)):
    if not file.content_type.startswith("image/"):
        raise HTTPException(status_code=400, detail="File must be an image")

try:
     

两步式图像处理：首先使用 Gemini Vision 提取文本（OCR），然后使用 LangChain 分析链对该文本进行分析。与单步分析相比，这种方式能够提供更可靠的结果。

创建 app/routes/research.py：

from fastapi import APIRouter, HTTPException
from app.models.schemas import ResearchRequest, ResearchResponse, ResearchResult
from app.services.tavily_service import tavily_service
from app.chains.chat_chain import get_chat_response
from datetime import datetime

router = APIRouter(prefix="/api", tags=["Research"])

@router.post("/research", response_model=ResearchResponse)
async def search_medical_research(request: ResearchRequest):
    try:
        raw_results = tavily_service.search_medical_research(
            query=request.query,
            max_results=request.max_results
        )

formatted_results = tavily_service.format_results(raw_results)

results_text = "\n\n".join([
            f"Source: {r['title']}\n{r['content']}"
            for r in formatted_results[:3]
        ])

summary_prompt = f"""Based on these medical research results, provide a brief summary in 2-3 sentences:

{results_text}

Focus on the key takeaways and most important information."""

summary = get_chat_response(summary_prompt, request.language)

research_results = [
            ResearchResult(
                title=r["title"],
                url=r["url"],
                content=r["content"],
                score=r["score"]
            )
            for r in formatted_results
        ]

return ResearchResponse(
            query=request.query,
            results=research_results,
            summary=summary,
            timestamp=datetime.now()
        )
    excep