部署监控

测试通过后，将应用部署到生产环境并添加监控。

部署方案对比

方案	优点	缺点	成本
Docker + 云服务器	灵活、可控	需要运维	中
Streamlit Cloud	简单、免费	资源有限	免费
LangGraph Cloud	官方支持	较新	按量付费

本章采用 Docker 部署方案。

Docker 配置

Dockerfile

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制代码
COPY . .

# 暴露端口
EXPOSE 8501

# 启动命令
CMD ["streamlit", "run", "ui/app.py", "--server.port=8501", "--server.address=0.0.0.0"]

docker-compose.yml

# docker-compose.yml
version: '3.8'

services:
  research-agent:
    build: .
    ports:
      - "8501:8501"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - TAVILY_API_KEY=${TAVILY_API_KEY}
      - LANGCHAIN_TRACING_V2=true
      - LANGCHAIN_API_KEY=${LANGCHAIN_API_KEY}
    restart: unless-stopped

构建和运行

# 构建镜像
docker build -t research-agent .

# 运行容器
docker-compose up -d

# 查看日志
docker-compose logs -f

LangSmith 监控

LangSmith 是 LangChain 官方的监控平台，提供：

• 调用链追踪
• 性能分析
• 错误监控
• 成本统计

配置 LangSmith

# config.py
import os

# LangSmith 配置
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "research-agent"
os.environ["LANGCHAIN_API_KEY"] = os.getenv("LANGCHAIN_API_KEY")

查看追踪

访问 smith.langchain.com 查看：

• 每次调用的完整链路
• 各节点执行时间
• Token 使用量
• 错误详情

日志配置

# utils/logging.py
import logging
import sys

def setup_logging():
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
        handlers=[
            logging.StreamHandler(sys.stdout),
            logging.FileHandler('app.log')
        ]
    )

logger = logging.getLogger("research_agent")

健康检查

# health.py
from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/health")
def health_check():
    return {"status": "healthy"}

@app.get("/ready")
def readiness_check():
    # 检查依赖服务
    checks = {
        "openai": check_openai(),
        "tavily": check_tavily()
    }
    all_ok = all(checks.values())
    return {"ready": all_ok, "checks": checks}

def check_openai():
    try:
        from langchain_openai import ChatOpenAI
        llm = ChatOpenAI(model="gpt-4o-mini")
        llm.invoke("test")
        return True
    except:
        return False

def check_tavily():
    try:
        from langchain_community.tools.tavily_search import TavilySearchResults
        tool = TavilySearchResults(max_results=1)
        tool.invoke("test")
        return True
    except:
        return False

告警配置

简单的错误告警（使用 Webhook）：

# utils/alerts.py
import requests
import os

WEBHOOK_URL = os.getenv("ALERT_WEBHOOK_URL")

def send_alert(message: str, level: str = "warning"):
    if not WEBHOOK_URL:
        return

    payload = {
        "text": f"[{level.upper()}] Research Agent: {message}"
    }

    try:
        requests.post(WEBHOOK_URL, json=payload, timeout=5)
    except Exception as e:
        print(f"告警发送失败: {e}")

成本监控

追踪 API 调用成本：

# utils/cost.py
from langchain_community.callbacks import get_openai_callback

def run_with_cost_tracking(func, *args, **kwargs):
    with get_openai_callback() as cb:
        result = func(*args, **kwargs)
        print(f"Token 使用: {cb.total_tokens}")
        print(f"成本: ${cb.total_cost:.4f}")
        return result, cb

部署检查清单

部署前确认：

• [ ] 环境变量已配置
• [ ] Docker 镜像构建成功
• [ ] 健康检查端点可访问
• [ ] LangSmith 追踪正常
• [ ] 日志输出正常
• [ ] 告警通道已测试

验收标准

完成部署后，你应该有：

• [ ] 应用成功部署并可访问
• [ ] LangSmith 监控正常工作
• [ ] 日志记录完整
• [ ] 健康检查端点可用
• [ ] 能查看调用成本

项目完成

恭喜！你已完成多 Agent 研究助手的全部开发。

回顾学到的内容：

1. 多 Agent 协作机制
2. LangGraph 状态管理和编排
3. Streamlit 前端开发
4. 集成测试方法
5. Docker 部署和监控

---

← 返回集成测试 | 返回项目二