✨ LangChain 1.0 最佳实践

"""
动态系统提示词示例
根据用户状态和偏好调整提示词
"""
from langchain.agents import create_agent

def generate_system_prompt(user_context: dict) -> str:
    """根据用户上下文生成个性化提示词"""

    base_prompt = "你是一个专业的 AI 助手。\n\n"

    # 根据对话长度调整详细程度
    conversation_length = user_context.get("message_count", 0)
    if conversation_length < 3:
        base_prompt += "请提供详细的解释和背景信息。\n"
    else:
        base_prompt += "用户已经了解基础知识，请简明扼要地回答。\n"

    # 根据用户角色调整权限
    user_role = user_context.get("role", "guest")
    if user_role == "admin":
        base_prompt += "你可以访问管理功能和敏感数据。\n"
    elif user_role == "user":
        base_prompt += "你只能访问用户级别的功能。\n"

    # 根据用户偏好调整风格
    preferences = user_context.get("preferences", {})
    language = preferences.get("language", "zh-CN")
    formality = preferences.get("formality", "professional")

    if formality == "casual":
        base_prompt += "使用轻松、友好的语气。\n"
    else:
        base_prompt += "保持专业、正式的语气。\n"

    # 添加合规要求
    if user_context.get("requires_compliance"):
        base_prompt += "\n重要：遵守 GDPR 和数据隐私法规，不得泄露用户个人信息。\n"

    return base_prompt

# 使用示例
user_context = {
    "message_count": 5,
    "role": "user",
    "preferences": {"language": "zh-CN", "formality": "casual"},
    "requires_compliance": True
}

agent = create_agent(
    model="gpt-4o",
    tools=[...],
    system_prompt=generate_system_prompt(user_context)
)

"""
动态工具选择
根据用户状态和权限过滤可用工具
"""
from langchain.tools import tool
from typing import Literal

# 定义所有工具
@tool
def public_search(query: str) -> str:
    """公开搜索工具（所有用户可用）"""
    return f"搜索结果：{query}"

@tool
def private_database_query(query: str) -> str:
    """私有数据库查询（仅认证用户）"""
    return f"数据库结果：{query}"

@tool
def admin_delete_user(user_id: str) -> str:
    """删除用户（仅管理员）"""
    return f"已删除用户：{user_id}"

@tool
def premium_analysis(data: str) -> str:
    """高级分析（仅付费用户）"""
    return f"分析结果：{data}"

# 工具选择逻辑
def select_tools_for_user(
    user_role: Literal["guest", "user", "premium", "admin"],
    is_authenticated: bool
) -> list:
    """根据用户角色和认证状态选择工具"""

    # 基础工具（所有人可用）
    available_tools = [public_search]

    # 认证用户工具
    if is_authenticated:
        available_tools.append(private_database_query)

    # 付费用户工具
    if user_role in ["premium", "admin"]:
        available_tools.append(premium_analysis)

    # 管理员工具
    if user_role == "admin":
        available_tools.append(admin_delete_user)

    return available_tools

# 使用示例
tools_for_guest = select_tools_for_user("guest", is_authenticated=False)
print(f"访客可用工具：{[t.name for t in tools_for_guest]}")
# 输出：['public_search']

tools_for_admin = select_tools_for_user("admin", is_authenticated=True)
print(f"管理员可用工具：{[t.name for t in tools_for_admin]}")
# 输出：['public_search', 'private_database_query', 'premium_analysis', 'admin_delete_user']

# 创建 Agent
agent = create_agent(
    model="gpt-4o",
    tools=tools_for_admin,  # 动态选择的工具
    system_prompt="..."
)

"""
上下文注入示例
在对话中动态注入相关信息
"""
def inject_context_messages(
    user_messages: list,
    uploaded_files: list = None,
    user_preferences: dict = None
) -> list:
    """在用户消息中注入上下文信息"""

    enriched_messages = []

    # 添加用户偏好上下文
    if user_preferences:
        context_msg = {
            "role": "system",
            "content": f"用户偏好：{user_preferences}"
        }
        enriched_messages.append(context_msg)

    # 添加文件上下文
    if uploaded_files:
        for file in uploaded_files:
            file_context = {
                "role": "system",
                "content": f"用户上传了文件：{file['name']}\n内容摘要：{file['summary']}"
            }
            enriched_messages.append(file_context)

    # 添加用户消息
    enriched_messages.extend(user_messages)

    return enriched_messages

# 使用示例
user_messages = [
    {"role": "user", "content": "分析这个文件"}
]

uploaded_files = [
    {"name": "sales_data.csv", "summary": "2024年销售数据，包含1000条记录"}
]

user_preferences = {
    "analysis_depth": "detailed",
    "chart_type": "bar"
}

final_messages = inject_context_messages(
    user_messages,
    uploaded_files=uploaded_files,
    user_preferences=user_preferences
)

# final_messages 将包含系统消息 + 文件上下文 + 用户消息
result = agent.invoke({"messages": final_messages})

"""
PII 检测中间件
自动检测和处理个人身份信息
"""
from langchain.agents.middleware import PIIMiddleware
from langchain.agents import create_agent

# PII 处理策略
pii_middleware = PIIMiddleware(
    # 检测类型
    pii_types=[
        "email",           # 电子邮件
        "credit_card",     # 信用卡号
        "ip_address",      # IP 地址
        "phone_number",    # 电话号码
        "ssn",            # 社会保障号
    ],

    # 处理策略
    strategy="redact",  # 选项：redact, mask, hash, block

    # 自定义模式（正则表达式）
    custom_patterns={
        "chinese_id": r"\d{17}[\dXx]",  # 中国身份证号
        "bank_card": r"\d{16,19}",      # 银行卡号
    }
)

agent = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[pii_middleware],
    system_prompt="你是客服助手。注意保护用户隐私信息。"
)

# 测试
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "我的邮箱是 [email protected]，电话是 13812345678"
    }]
})

# PII 会被自动处理：
# "我的邮箱是 [REDACTED_EMAIL]，电话是 [REDACTED_PHONE_NUMBER]"

"""
自定义内容安全 Guardrail
使用模型检查输出内容安全性
"""
from langchain.agents.middleware import CustomMiddleware
from langchain_openai import ChatOpenAI

class ContentSafetyGuardrail(CustomMiddleware):
    """内容安全检查中间件"""

    def __init__(self):
        self.safety_checker = ChatOpenAI(model="gpt-4o-mini")

    def after_agent(self, state, config):
        """Agent 执行后检查输出安全性"""
        last_message = state["messages"][-1]

        # 跳过工具调用消息
        if not hasattr(last_message, "content"):
            return state

        content = last_message.content

        # 使用模型检查安全性
        safety_prompt = f"""
        请评估以下 AI 回答是否安全、合规：

        {content}

        评估标准：
        1. 不包含有害、暴力、歧视性内容
        2. 不泄露敏感信息
        3. 符合道德和法律要求

        请回答："安全" 或 "不安全：[原因]"
        """

        response = self.safety_checker.invoke(safety_prompt)
        result = response.content.strip()

        if not result.startswith("安全"):
            # 不安全，替换为安全消息
            state["messages"][-1].content = "抱歉，我无法提供该信息。请换一个问题。"

        return state

# 使用
agent = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[ContentSafetyGuardrail()],
)

"""
多层 Guardrails 组合
构建深度防御体系
"""
from langchain.agents import create_agent
from langchain.agents.middleware import (
    PIIMiddleware,
    HumanInTheLoopMiddleware,
    CustomMiddleware
)

# 第 1 层：输入验证（确定性）
class InputValidationGuardrail(CustomMiddleware):
    """验证输入格式和长度"""

    def before_agent(self, state, config):
        last_message = state["messages"][-1]
        content = last_message.get("content", "")

        # 检查长度
        if len(content) > 5000:
            raise ValueError("输入过长，请控制在 5000 字符以内")

        # 检查禁用词
        forbidden_words = ["hack", "exploit", "jailbreak"]
        if any(word in content.lower() for word in forbidden_words):
            raise ValueError("输入包含不允许的内容")

        return state

# 第 2 层：PII 保护
pii_guardrail = PIIMiddleware(
    pii_types=["email", "phone_number", "credit_card"],
    strategy="redact"
)

# 第 3 层：敏感操作审核
hitl_guardrail = HumanInTheLoopMiddleware(
    interrupt_on={
        "delete_data": True,
        "send_email": True,
        "make_payment": True,
    }
)

# 第 4 层：输出安全检查（模型化）
output_safety_guardrail = ContentSafetyGuardrail()

# 组合所有 Guardrails
agent = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[
        InputValidationGuardrail(),   # 第 1 层
        pii_guardrail,                # 第 2 层
        hitl_guardrail,               # 第 3 层
        output_safety_guardrail,      # 第 4 层
    ],
    checkpointer=checkpointer,  # HITL 需要
    system_prompt="..."
)

# 这样构建了完整的安全防护体系

"""
动态模型选择
根据任务复杂度选择合适的模型
"""
from langchain_openai import ChatOpenAI
from langchain.agents.middleware import CustomMiddleware

class SmartModelRouter(CustomMiddleware):
    """智能模型路由中间件"""

    def __init__(self):
        self.cheap_model = ChatOpenAI(model="gpt-3.5-turbo")
        self.expensive_model = ChatOpenAI(model="gpt-4o")

    def before_model(self, state, config):
        """根据任务选择模型"""
        last_message = state["messages"][-1]
        content = last_message.get("content", "")

        # 简单任务标识
        simple_keywords = ["你好", "谢谢", "再见", "什么时候", "多少钱"]
        is_simple = any(kw in content for kw in simple_keywords)

        # 复杂任务标识
        complex_keywords = ["分析", "解释", "设计", "规划", "对比"]
        is_complex = any(kw in content for kw in complex_keywords)

        # 选择模型
        if is_simple:
            config["model"] = self.cheap_model
            print("🟢 使用 GPT-3.5-turbo（简单任务）")
        elif is_complex:
            config["model"] = self.expensive_model
            print("🔴 使用 GPT-4o（复杂任务）")
        else:
            # 默认使用便宜模型
            config["model"] = self.cheap_model
            print("🟡 使用 GPT-3.5-turbo（默认）")

        return state

# 使用
agent = create_agent(
    model="gpt-4o",  # 默认模型
    tools=[...],
    middleware=[SmartModelRouter()],
)

"""
Token 优化示例
减少不必要的 Token 消耗
"""
from langchain.agents.middleware import SummarizationMiddleware

# 1. 使用摘要中间件
summarization = SummarizationMiddleware(
    max_messages=10,  # 保留最近 10 条消息
    summarize_older=True  # 摘要旧消息
)

# 2. 精简系统提示词
# ❌ 冗长的提示词
bad_prompt = """
你是一个非常专业且经验丰富的客服助手。
你需要帮助用户解决各种各样的问题，包括但不限于...
请始终保持礼貌、耐心、友好的态度...
（省略 500 字）
"""

# ✅ 精简的提示词
good_prompt = """
你是专业客服助手。
职责：解答问题、查询订单、处理售后。
原则：礼貌、准确、及时。
"""

# 3. 限制对话历史长度
def trim_conversation_history(messages: list, max_messages: int = 20) -> list:
    """保留最近的 N 条消息"""
    if len(messages) <= max_messages:
        return messages

    # 保留系统消息 + 最近的用户/AI 消息
    system_messages = [m for m in messages if m["role"] == "system"]
    recent_messages = messages[-max_messages:]

    return system_messages + recent_messages

# 4. 使用更小的 Embedding 模型
from langchain_openai import OpenAIEmbeddings

# text-embedding-3-small: 512 维，成本更低
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

"""
成本优化完整示例
"""
import functools
from datetime import datetime, timedelta

# 1. 结果缓存
cache = {}

def cache_result(ttl_seconds: int = 3600):
    """缓存装饰器"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            cache_key = f"{func.__name__}:{args}:{kwargs}"

            # 检查缓存
            if cache_key in cache:
                result, timestamp = cache[cache_key]
                if datetime.now() - timestamp < timedelta(seconds=ttl_seconds):
                    print(f"💰 使用缓存结果：{func.__name__}")
                    return result

            # 执行函数
            result = func(*args, **kwargs)

            # 保存缓存
            cache[cache_key] = (result, datetime.now())
            return result

        return wrapper
    return decorator

# 2. FAQ 缓存
@cache_result(ttl_seconds=3600)  # 缓存 1 小时
def search_faq_cached(query: str) -> str:
    """带缓存的 FAQ 搜索"""
    return search_faq(query)

# 3. 批量 Embedding
def batch_embed_documents(texts: list[str], batch_size: int = 100):
    """批量生成 Embeddings，降低成本"""
    embeddings_model = OpenAIEmbeddings(model="text-embedding-3-small")

    all_embeddings = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        batch_embeddings = embeddings_model.embed_documents(batch)
        all_embeddings.extend(batch_embeddings)

    return all_embeddings

# 4. 限流保护
import time
from collections import deque

class RateLimiter:
    """简单的限流器"""

    def __init__(self, max_requests: int, time_window: int):
        self.max_requests = max_requests
        self.time_window = time_window
        self.requests = deque()

    def allow_request(self) -> bool:
        """检查是否允许请求"""
        now = time.time()

        # 清理过期请求
        while self.requests and self.requests[0] < now - self.time_window:
            self.requests.popleft()

        # 检查限制
        if len(self.requests) >= self.max_requests:
            return False

        # 记录请求
        self.requests.append(now)
        return True

# 使用限流器
rate_limiter = RateLimiter(max_requests=100, time_window=60)  # 每分钟 100 次

def process_request(user_message: str):
    if not rate_limiter.allow_request():
        return "请求过于频繁，请稍后再试"

    # 处理请求
    return agent.invoke({"messages": [{"role": "user", "content": user_message}]})

# 5. 成本监控
def track_cost(num_input_tokens: int, num_output_tokens: int, model: str):
    """追踪成本"""
    # GPT-4o 定价（示例）
    pricing = {
        "gpt-4o": {"input": 0.00001, "output": 0.00003},  # 每 token 价格
        "gpt-3.5-turbo": {"input": 0.000001, "output": 0.000002},
    }

    price = pricing.get(model, pricing["gpt-4o"])
    cost = num_input_tokens * price["input"] + num_output_tokens * price["output"]

    print(f"💸 本次调用成本：${cost:.6f}")
    return cost

"""
LangSmith 追踪配置
"""
import os

# 配置 LangSmith
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "customer-service-bot"  # 项目名称

# 所有 Agent 调用都会自动追踪
result = agent.invoke({"messages": [...]})

# 查看追踪：https://smith.langchain.com/

"""
结构化日志记录
"""
import logging
import json
from datetime import datetime

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("customer-service")

class LoggingMiddleware(CustomMiddleware):
    """日志记录中间件"""

    def before_agent(self, state, config):
        """记录请求开始"""
        user_message = state["messages"][-1].get("content", "")

        log_data = {
            "event": "agent_request",
            "timestamp": datetime.now().isoformat(),
            "user_id": config.get("context", {}).get("user_id"),
            "message_preview": user_message[:100],
        }

        logger.info(json.dumps(log_data, ensure_ascii=False))
        return state

    def after_agent(self, state, config):
        """记录响应完成"""
        ai_message = state["messages"][-1]

        log_data = {
            "event": "agent_response",
            "timestamp": datetime.now().isoformat(),
            "user_id": config.get("context", {}).get("user_id"),
            "has_tool_calls": hasattr(ai_message, "tool_calls") and bool(ai_message.tool_calls),
            "response_length": len(ai_message.content) if hasattr(ai_message, "content") else 0,
        }

        logger.info(json.dumps(log_data, ensure_ascii=False))
        return state

# 使用
agent = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[LoggingMiddleware()],
)

"""
工具单元测试
"""
import pytest
from unittest.mock import Mock

def test_get_my_orders():
    """测试订单查询工具"""
    # 模拟 Runtime
    mock_runtime = Mock()
    mock_runtime.context = {"user_id": "user_123"}

    # 调用工具
    result = get_my_orders.invoke({"runtime": mock_runtime})

    # 断言
    assert "订单" in result
    assert "user_123" in result or len(result) > 0

def test_search_faq_with_valid_query():
    """测试 FAQ 搜索（有效查询）"""
    result = search_faq.invoke({"query": "配送时间"})

    assert "配送" in result or "工作日" in result
    assert len(result) > 0

def test_search_faq_with_invalid_query():
    """测试 FAQ 搜索（无效查询）"""
    result = search_faq.invoke({"query": "askdjflkajsdf"})

    assert "没有找到" in result or "抱歉" in result

def test_cancel_order_without_permission():
    """测试取消订单（无权限）"""
    mock_runtime = Mock()
    mock_runtime.context = {"user_id": "wrong_user"}

    result = cancel_order.invoke({
        "order_id": "ORDER-2024001",
        "reason": "测试",
        "runtime": mock_runtime
    })

    assert "未找到" in result or "无法取消" in result

"""
Agent 集成测试（模拟 LLM）
"""
import pytest
from unittest.mock import patch

@patch('langchain_openai.ChatOpenAI.invoke')
def test_agent_faq_flow(mock_llm):
    """测试 FAQ 查询流程"""
    # 模拟 LLM 返回工具调用
    mock_llm.return_value = Mock(
        content="",
        tool_calls=[{
            "name": "search_faq",
            "args": {"query": "配送时间"},
            "id": "call_123"
        }]
    )

    # 创建 Agent
    agent = create_agent(
        model="gpt-4o",
        tools=[search_faq],
    )

    # 调用
    result = agent.invoke({
        "messages": [{"role": "user", "content": "多久能收到货？"}]
    })

    # 验证工具被调用
    assert any("配送" in str(msg) for msg in result["messages"])

def test_agent_with_checkpointer():
    """测试多轮对话（真实 Agent）"""
    from langgraph.checkpoint.memory import InMemorySaver

    checkpointer = InMemorySaver()
    agent = create_agent(
        model="gpt-3.5-turbo",  # 使用便宜模型测试
        tools=[get_my_orders],
        checkpointer=checkpointer,
    )

    config = {"configurable": {"thread_id": "test_001"}}

    # 第一轮对话
    result1 = agent.invoke(
        {"messages": [{"role": "user", "content": "你好"}]},
        config=config
    )

    # 第二轮对话（应该记住上下文）
    result2 = agent.invoke(
        {"messages": [{"role": "user", "content": "我刚才说了什么？"}]},
        config=config
    )

    # 验证对话历史
    assert len(result2["messages"]) > 2  # 包含历史消息

"""
生产环境 Agent 配置
包含所有最佳实践
"""
import os
from langchain.agents import create_agent
from langchain.agents.middleware import (
    PIIMiddleware,
    HumanInTheLoopMiddleware,
    SummarizationMiddleware,
)
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.postgres import PostgresStore

# ==================== 环境配置 ====================

# 数据库连接
DB_URI = os.getenv("DATABASE_URL")
assert DB_URI, "DATABASE_URL 环境变量未设置"

# LangSmith 追踪
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = os.getenv("LANGSMITH_API_KEY")
os.environ["LANGCHAIN_PROJECT"] = "production-customer-service"

# ==================== 持久化配置 ====================

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()

    with PostgresStore.from_conn_string(DB_URI) as store:
        store.setup()

        # ==================== 中间件配置 ====================

        # PII 保护
        pii_middleware = PIIMiddleware(
            pii_types=["email", "phone_number", "credit_card", "ip_address"],
            strategy="redact",
            custom_patterns={
                "chinese_id": r"\d{17}[\dXx]",
                "bank_card": r"\d{16,19}",
            }
        )

        # 对话摘要
        summarization = SummarizationMiddleware(
            max_messages=20,
            summarize_older=True
        )

        # 敏感操作审核
        hitl = HumanInTheLoopMiddleware(
            interrupt_on={
                "cancel_order": True,
                "refund_order": True,
                "delete_user_data": True,
            }
        )

        # 自定义中间件
        logging_middleware = LoggingMiddleware()
        content_safety = ContentSafetyGuardrail()

        # ==================== 创建 Agent ====================

        production_agent = create_agent(
            model="gpt-4o",
            tools=all_production_tools,  # 所有工具
            middleware=[
                logging_middleware,      # 日志
                pii_middleware,          # PII 保护
                summarization,           # 摘要
                hitl,                    # 人工介入
                content_safety,          # 内容安全
            ],
            checkpointer=checkpointer,   # PostgreSQL 持久化
            store=store,                 # PostgreSQL Store
            context_schema=CustomerContext,
            system_prompt=generate_production_prompt(),  # 动态提示词
        )

        # ==================== 监控和告警 ====================

        def invoke_with_monitoring(user_message: str, user_id: str, thread_id: str):
            """带监控的调用"""
            import time

            start_time = time.time()

            try:
                config = {
                    "configurable": {"thread_id": thread_id},
                    "context": {"user_id": user_id}
                }

                result = production_agent.invoke(
                    {"messages": [{"role": "user", "content": user_message}]},
                    config=config
                )

                # 记录成功
                latency = time.time() - start_time
                logger.info(json.dumps({
                    "event": "agent_success",
                    "user_id": user_id,
                    "latency": latency,
                }))

                return result

            except Exception as e:
                # 记录错误
                logger.error(json.dumps({
                    "event": "agent_error",
                    "user_id": user_id,
                    "error": str(e),
                }))

                # 发送告警
                # send_alert(f"Agent error: {e}")

                # 返回友好错误消息
                return {
                    "messages": [{
                        "role": "assistant",
                        "content": "抱歉，系统遇到了一些问题。请稍后再试或联系客服。"
                    }]
                }

        # ==================== 健康检查 ====================

        def health_check():
            """健康检查端点"""
            checks = {
                "database": check_database_connection(DB_URI),
                "model": check_model_availability(),
                "vector_store": check_vector_store(),
            }

            all_healthy = all(checks.values())

            return {
                "status": "healthy" if all_healthy else "unhealthy",
                "checks": checks
            }