内容
# 医疗收入循环自动化
<p align="center">
<img src="https://img.shields.io/badge/Python-3.12+-blue.svg" alt="Python 3.12+">
<img src="https://img.shields.io/badge/Google%20ADK-2.4-4285F4.svg" alt="Google ADK 2.4">
<img src="https://img.shields.io/badge/Gemini-2.5%20Flash-8E75B2.svg" alt="Gemini 2.5 Flash">
<img src="https://img.shields.io/badge/FastAPI-0.139+-009688.svg" alt="FastAPI 0.139+">
<img src="https://img.shields.io/badge/A2A-Agent2Agent%20Protocol-5C6BC0.svg" alt="Agent2Agent 协议">
<img src="https://img.shields.io/badge/MCP-Model%20Context%20Protocol-F97316.svg" alt="模型上下文协议">
<img src="https://img.shields.io/badge/Gradio-6.20-FF7C00.svg" alt="Gradio 6.20">
<img src="https://img.shields.io/badge/Google%20Cloud-Run-4285F4.svg" alt="Google Cloud Run">
<img src="https://img.shields.io/badge/Docker-Container-2496ED.svg" alt="Docker">
</p>
## 业务问题
医疗团队经常手动检查保险、选择计费代码、发送理赔并处理被拒绝的理赔。这种工作可能很慢。缺少数据或错误的代码可能会延迟付款或导致收入损失。
这项工作也可能分散在许多系统中。工作人员必须重复步骤，并在理赔失败后查找错误。
## 如何解决问题
该系统将主要的计费步骤放在一个工作流中。它读取理赔数据、检查保险号码、建议ICD-10代码、发送测试理赔并检查结果。如果代码缺失，它会重试编码步骤。如果理赔被拒绝，它会撰写上诉信。它还保存了一个简短的理赔摘要以供以后使用。
Google ADK控制工作流和代理。Gemini帮助进行受理、编码和上诉写作。固定规则检查重要结果。Agent2Agent（A2A）协议允许其他代理使用工作流。模型上下文协议（MCP）使计费工具可用于MCP客户端。FastAPI支持普通的Web应用程序，Gradio为用户提供了一个简单的Web界面。Google Cloud Run托管服务。
这是一个技术原型。它不是医疗产品，不应用于实际的临床或计费决策。
## 理赔如何在系统中流转
系统按以下步骤处理理赔：
1. 受理代理读取患者ID、保险号码和就诊原因。
2. 固定规则检查保险号码。
3. 编码代理从批准的ICD-10目录中建议代码。
4. 验证删除不支持的代码。
5. 异步理赔步骤模拟缓慢的外部计费API。
6. 审计步骤列出问题并给出问题分数。
7. 缺失的代码返回到编码，最多两次。
8. 上诉代理为被拒绝的理赔撰写信件。
9. 内存步骤存储简短的理赔摘要。
Gemini用于受理、编码和上诉写作。如果Gemini不可用，则使用本地规则和模板。
## 架构
<p align="center">
<img src="assets/architecture.png" alt="医疗收入循环代理架构" width="100%">
</p>
## 主要功能
### 多代理工作流
- 受理代理解析请求。
- 编码代理建议诊断代码。
- 上诉代理在需要时撰写简短的上诉。
- Python工作流节点处理保险、验证、理赔提交、审计、重试和内存。
- Pydantic模型在每一步之间传递类型化数据。
### 模型上下文协议（MCP）工具
`mcp_server.py` 使用官方MCP Python SDK中的`FastMCP`。MCP客户端可以发现并调用这些工具：
- `verify_insurance`
- `search_code_catalog`
- `submit_claim`
- `calculate_issue_score`
服务器支持本地MCP客户端的stdio和Cloud Run的Streamable HTTP。
### Agent2Agent（A2A）协议
`a2a_server.py` 使用ADK的`to_a2a()`公开完整的工作流，通过Agent2Agent协议。ADK创建代理卡和A2A路由。这允许另一个代理发现服务并向其发送工作。
`POST /a2a/rcm`路由是一个单独的兼容性API。它接受普通的JSON，对于不使用A2A协议的应用程序很有用。
### 长时间运行的操作
理赔提交使用`asyncio.sleep()`表示网络延迟，而不阻塞服务器。它在成功时返回操作ID和理赔ID。
### 会话和内存
- ADK会话通过`DatabaseSessionService`存储。
- 调用者可以发送相同的会话ID以继续会话。
- 工作流历史记录每个重要决定。
- 最后500个字符保存为紧凑的上下文。
- 理赔内存存储在SQLite中，可以在`GET /memory`中查看。
SQLite适用于本地测试。在Google Cloud上使用Cloud SQL或Agent Runtime进行持久的生产存储。
### 检查和日志
- 审计步骤检查保险、代码、审查需求和拒绝状态。
- `issue_score`是审计问题的数量。
- 缺失的代码启动编码重试。
- 日志以JSON写入，以便Cloud Logging可以轻松读取。
- 测试涵盖提交、拒绝、重试、备用解析、上下文压缩和内存。
### 代码结构
- `models.py` 包含Pydantic数据模型。
- `core.py` 包含计费规则。
- `agent.py` 包含Google ADK工作流。
- `runtime.py` 运行工作流并管理会话。
- `main.py` 包含API路由。
- `gradio_app.py` 包含用户界面。
- `mcp_server.py` 使用`FastMCP`，因此没有手动协议代码。
## 项目结构
```text
healthcare-revenue-cycle-agent/
├── .dockerignore
├── .env.example
├── .python-version
├── assets/
│ └── architecture.png
├── rcm_agent/
│ ├── __init__.py
│ ├── agent.py
│ ├── core.py
│ ├── logging_config.py
│ ├── memory.py
│ └── models.py
├── tests/
│ ├── test_core.py
│ └── test_memory.py
├── a2a_server.py
├── gradio_app.py
├── main.py
├── mcp_server.py
├── runtime.py
├── Dockerfile
├── README.md
├── deploy.sh
├── pyproject.toml
├── ai_rcm_agent_v2.ipynb
└── uv.lock
```
## 运行项目步骤
您需要`uv`、Google Cloud CLI和一个启用了计费的Google Cloud项目。该项目使用Python 3.12。
### 1. 打开项目文件夹
```bash
cd rcm-adk-google-cloud
```
### 2. 安装项目包
```bash
uv sync
```
`uv`创建`.venv`并安装锁定包版本。
### 3. 设置Google Cloud项目
复制示例设置文件：
```bash
cp .env.example .env
```
打开`.env`。将`your-project-id`替换为您的Google Cloud项目ID。
### 4. 登录Google Cloud
```bash
gcloud auth application-default login
gcloud config set project YOUR_PROJECT_ID
```
将`YOUR_PROJECT_ID`替换为相同的项目ID。
### 5. 运行测试
```bash
uv run pytest
```
### 6. 启动应用程序
```bash
uv run uvicorn main:app --reload --env-file .env
```
### 7. 打开应用程序
- ADK界面：`http://localhost:8000`
- Gradio界面：`http://localhost:8000/ui`
- API文档：`http://localhost:8000/docs`
## 发送测试API请求
保持应用程序运行。打开第二个终端并运行：
```bash
curl -X POST http://localhost:8000/a2a/rcm \
-H "Content-Type: application/json" \
-d '{
"patient_id": "P001",
"insurance_number": "AET123456",
"visit_reason": "Type 2 diabetes and high blood pressure"
}'
```
在下一个请求中使用返回的`session_id`继续会话。
## 运行Agent2Agent（A2A）服务
```bash
uv run uvicorn a2a_server:app --host 0.0.0.0 --port 8001 --env-file .env
```
Google ADK生成代理卡和A2A端点。
## 运行模型上下文协议（MCP）服务
stdio模式：
```bash
uv run python mcp_server.py
```
Streamable HTTP模式：
```bash
MCP_TRANSPORT=http PORT=8002 uv run python mcp_server.py
```
Streamable HTTP MCP端点是`http://localhost:8002/mcp`。
## Google Cloud部署
脚本部署三个私有Cloud Run服务：
- 主要ADK API和Gradio UI
- A2A协议服务
- MCP协议服务
```bash
gcloud auth login
export GOOGLE_CLOUD_PROJECT="your-project-id"
export GOOGLE_CLOUD_LOCATION="us-central1"
chmod +x deploy.sh
./deploy.sh
```
在运行脚本之前替换`your-project-id`。脚本启用所需的Google Cloud API，创建服务帐户并部署三个服务。
使用Cloud Run代理在本地打开私有服务：
```bash
gcloud run services proxy rcm-adk-service \
--project "$GOOGLE_CLOUD_PROJECT" \
--region "$GOOGLE_CLOUD_LOCATION"
```
## 待办事项
1. 在安全的测试环境中连接保险资格API和理赔结算中心。
2. 使用批准的ICD-10和CPT数据源。要求医疗编码员在发送理赔之前批准代码。
3. 将会话和内存移动到Cloud SQL。添加IAM、Secret Manager和加密审计日志。
