01 一个尴尬的现实：95%的大模型项目，停在Demo阶段

过去两年，我面试了超过200位开发者，看过不下500个AI项目Demo。一个令人唏嘘的规律是——95%的项目在PPT里无所不能，在生产环境里一碰就碎。

这些Demo有着惊人的相似之处：

• 打开一个漂亮的聊天界面，输入问题，模型流式输出一段逻辑清晰、引经据典的回答
• 产品经理满意地点点头，CTO拍板立项，技术团队信心满满地开始开发

然后，噩梦开始了。

当系统从Jupyter Notebook迁移到生产环境，真实业务数据涌入的那一刻——幻觉层出不穷，检索结果张冠李戴，推理延迟从1秒飙升到15秒，Token费用像水龙头没关紧一样哗哗流走。

更糟糕的是，业务方开始质疑：“这东西连我们内部Wiki都查不明白，怎么敢让它面对客户？”

大模型很聪明，但聪明和有用之间，隔着一整个工程世界的距离。

02 “能聊天”和“能干活”之间，差了什么？

让我们用一个真实的银行客服场景来拆解这个差距。

场景：信用卡盗刷投诉处理

“能聊天”的版本：

用户输入：“我的信用卡被盗刷了，怎么办？”

大模型回答：“非常抱歉听到这个消息。建议您立即致电银行客服热线400-xxx-xxxx进行挂失，并前往就近派出所报案。银行会协助您处理后续事宜。”

——回答专业、礼貌、正确。但仅此而已。

“能干活”的版本：

大模型接收到这条消息后，后台发生了一连串动作：

1. 意图识别：识别为“盗刷投诉”场景，优先级P0
2. 身份核验：自动调取用户绑定的信用卡信息，比对姓名、身份证后四位
3. 风控联动：调用风控系统接口，标记该卡为“疑似盗刷”状态，冻结所有非柜台交易
4. 工单创建：在CRM系统中自动创建投诉工单，关联用户ID和卡号
5. 证据保全：将最近5笔交易记录打包，附上时间戳水印
6. 话术生成：根据该用户的历史偏好（沟通记录显示该用户偏好简洁风格），生成安抚话术
7. 转接策略：如果用户情绪词（如“投诉”“赔偿”“你们要负责”）密度超过阈值，自动转接人工高优先队列

回答输出：“王先生您好，我已紧急冻结您尾号8848的信用卡，防止进一步损失。您的投诉工单#T20260707已生成，预计3分钟内专属客服经理将致电您。同时，我已将您近5笔交易记录打包，建议您在通话中与客服核对。请问还有其它需要帮助的吗？”

——这才叫“能干活”。

两者的差距，是被动应答与主动解决问题的差距，是文本生成与业务闭环的差距，是玩具与工具的差距。

03 让大模型上班，需要哪些“管教手段”？

让一个自由散漫的生成式模型变成企业里的“优秀员工”，需要一套完整的“管教体系”。

管教手段一：RAG——给模型配一个“企业级外脑”

大模型记不住你公司的内部信息，除非你给它配一个随时可查的资料库。

import hashlib
import json
from datetime import datetime

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Milvus
from typing import List, Dict

class EnterpriseRAGSystem:
    """
    企业级RAG系统
    特点：自带权限过滤、版本追踪、来源溯源
    """
    
    def __init__(self, collection_name: str):
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
        self.vector_store = Milvus(
            collection_name=collection_name,
            embedding_function=self.embeddings,
            connection_args={"host": "milvus-prod.internal", "port": "19530"}
        )
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=512,
            chunk_overlap=50,
            separators=["\n\n", "\n", "。", "！", "？", "；", ". ", "! ", "? ", "; "]
        )
    
    def ingest_document(self, content: str, metadata: Dict) -> str:
        """
        文档入库，附带完整元数据
        返回文档ID，用于后续溯源
        """
        # 1. 切分
        chunks = self.text_splitter.split_text(content)
        
        # 2. 为每个chunk生成唯一ID
        doc_id = f"doc_{hashlib.md5(content.encode()).hexdigest()[:12]}"
        
        # 3. 构建带完整上下文的元数据
        enriched_metadata = {
            **metadata,
            "doc_id": doc_id,
            "chunk_count": len(chunks),
            "ingested_at": datetime.now().isoformat(),
            # 【关键】权限控制字段
            "allowed_departments": metadata.get("allowed_departments", []),
            "allowed_user_roles": metadata.get("allowed_user_roles", []),
        }
        
        # 4. 入库
        for idx, chunk in enumerate(chunks):
            self.vector_store.add_texts(
                texts=[chunk],
                metadatas=[{
                    **enriched_metadata,
                    "chunk_index": idx,
                    "total_chunks": len(chunks)
                }]
            )
        
        return doc_id
    
    def retrieve_with_permission(self, query: str, user_info: Dict, top_k: int = 5) -> List[Dict]:
        """
        带权限的检索
        用户只能看到自己有权访问的文档片段
        """
        # 【关键】用过滤条件实现数据隔离
        filter_expr = (
            f"allowed_departments contains '{user_info.get('department')}' OR "
            f"allowed_user_roles contains '{user_info.get('role')}' OR "
            "is_public == true"
        )
        
        results = self.vector_store.similarity_search_with_score(
            query=query,
            k=top_k,
            expr=filter_expr
        )
        
        # 返回结果并附上置信度
        return [
            {
                "content": doc.page_content,
                "source": doc.metadata.get("source", "unknown"),
                "doc_id": doc.metadata.get("doc_id"),
                "score": score,
                "confidence": self._calc_confidence(score)
            }
            for doc, score in results
        ]
    
    def _calc_confidence(self, score: float) -> str:
        """根据相似度分数计算置信等级"""
        if score > 0.85:
            return "high"
        elif score > 0.65:
            return "medium"
        else:
            return "low"

背后的工程智慧：RAG系统最大的坑不是召回率低，而是权限泄露——A部门的员工检索到了B部门的薪资数据。上面的代码通过filter_expr在向量检索层就完成了权限过滤，而不是检索完再过滤，性能和安全性都得到保障。

管教手段二：Agent编排——给大模型配上“手脚”

大模型只会说话，但企业需要它做事。Agent就是大模型的“手和脚”。

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolExecutor
import json
import requests

class AgentState(TypedDict):
    """Agent的状态管理"""
    user_query: str
    user_id: str
    session_id: str
    current_step: str
    tool_calls: list
    tool_results: list
    final_answer: str
    retry_count: int

# 定义可用的工具
TOOLS = [
    {
        "name": "query_customer_info",
        "description": "根据用户ID查询客户信息，包括姓名、会员等级、注册时间",
        "parameters": {
            "user_id": {"type": "string", "description": "用户唯一标识"}
        }
    },
    {
        "name": "check_order_status",
        "description": "查询订单状态，返回物流信息和预计送达时间",
        "parameters": {
            "order_id": {"type": "string", "description": "订单号"}
        }
    },
    {
        "name": "create_ticket",
        "description": "在客服系统中创建工单，需要工单类型和优先级",
        "parameters": {
            "ticket_type": {"type": "string", "enum": ["投诉", "咨询", "售后", "建议"]},
            "priority": {"type": "string", "enum": ["P0", "P1", "P2", "P3"]},
            "description": {"type": "string", "description": "工单描述"}
        }
    },
    {
        "name": "send_sms",
        "description": "向用户手机发送短信通知",
        "parameters": {
            "phone": {"type": "string", "description": "手机号"},
            "content": {"type": "string", "description": "短信内容"}
        }
    }
]

def build_agent_graph():
    """
    构建Agent工作流图
    这是一个"ReAct"模式的实现：思考-行动-观察-再思考
    """
    graph = StateGraph(AgentState)
    
    # 【节点1】意图理解与工具选择
    def decide_tool(state: AgentState) -> AgentState:
        prompt = f"""
        你是企业服务助手。根据用户问题，判断需要调用哪些工具。
        
        用户问题：{state['user_query']}
        
        可用工具：{json.dumps(TOOLS, ensure_ascii=False)}
        
        请返回JSON数组格式的工具调用列表，每个元素包含tool_name和参数。
        如果不需要工具，返回空数组。
        """
        # 调用LLM决策（此处简化展示逻辑）
        # 实际项目中会调用大模型API
        tool_calls = [
            {"tool_name": "query_customer_info", "params": {"user_id": state['user_id']}},
            {"tool_name": "check_order_status", "params": {"order_id": "ORD-20260707-001"}}
        ]
        state['tool_calls'] = tool_calls
        state['current_step'] = "executing_tools"
        return state
    
    # 【节点2】执行工具
    def execute_tools(state: AgentState) -> AgentState:
        results = []
        for call in state['tool_calls']:
            try:
                # 实际项目中的工具调用会走真正的业务API
                if call['tool_name'] == 'query_customer_info':
                    result = {"name": "张三", "level": "钻石会员", "register_at": "2023-01-15"}
                elif call['tool_name'] == 'check_order_status':
                    result = {"status": "派送中", "location": "杭州市西湖区", "eta": "今日18:00前"}
                else:
                    result = {"error": "未知工具"}
                
                results.append({
                    "tool": call['tool_name'],
                    "params": call['params'],
                    "result": result,
                    "success": True
                })
            except Exception as e:
                results.append({
                    "tool": call['tool_name'],
                    "error": str(e),
                    "success": False
                })
        
        state['tool_results'] = results
        state['current_step'] = "generating_answer"
        return state
    
    # 【节点3】生成最终答案
    def generate_answer(state: AgentState) -> AgentState:
        # 将工具执行结果和用户问题一起交给LLM生成最终回复
        context = f"""
        用户问题：{state['user_query']}
        
        工具执行结果：
        {json.dumps(state['tool_results'], ensure_ascii=False, indent=2)}
        
        请根据工具执行结果，生成一段专业、简洁、有帮助的回复。
        回复中要体现你使用了哪些信息来帮助用户。
        """
        # 调用LLM生成回复（此处简化）
        final = "张先生您好，您是钻石会员。您尾号8848的信用卡已成功激活。"
        state['final_answer'] = final
        state['current_step'] = "done"
        return state
    
    # 【节点4】兜底-降级处理
    def fallback(state: AgentState) -> AgentState:
        """当工具调用失败时，给出降级回复"""
        state['final_answer'] = "系统暂时无法完成您的请求，已为您转接人工客服，请稍候。"
        state['current_step'] = "fallback_done"
        return state
    
    # 构建图
    graph.add_node("decide_tool", decide_tool)
    graph.add_node("execute_tools", execute_tools)
    graph.add_node("generate_answer", generate_answer)
    graph.add_node("fallback", fallback)
    
    graph.set_entry_point("decide_tool")
    graph.add_edge("decide_tool", "execute_tools")
    
    # 【条件边】根据工具执行结果决定下一步
    def after_execution(state: AgentState) -> str:
        # 如果所有工具都执行成功
        if all(r.get('success', False) for r in state['tool_results']):
            return "generate_answer"
        # 如果有失败的，进入降级
        else:
            return "fallback"
    
    graph.add_conditional_edges("execute_tools", after_execution)
    graph.add_edge("generate_answer", END)
    graph.add_edge("fallback", END)
    
    return graph.compile()

Agent编排的核心价值：不是让大模型“想说什么说什么”，而是让大模型“按流程办事”。就像给一个天才员工配上SOP（标准作业程序），确保他每次都能稳定输出。

管教手段三：Prompt Engineering——给大模型写“员工手册”

class SystemPromptFactory:
    """
    Prompt工厂：针对不同业务场景生成不同的System Prompt
    相当于给大模型准备了不同岗位的"入职培训手册"
    """
    
    @staticmethod
    def get_customer_service_prompt() -> str:
        return """
        【角色定位】
        你是XX银行智能客服助手，拥有3年以上客户服务经验，专业、耐心、细致。
        
        【行为准则】
        1. 第一句话必须完成身份确认（称呼用户姓氏+先生/女士）
        2. 每段回复必须包含一个明确的可执行动作（如"请点击""已为您""即将发送"）
        3. 涉及资金/账户操作时，必须提醒用户确认身份
        4. 当无法确认答案时，主动引导至人工，严禁编造信息
        
        【风格要求】
        语言简洁、信息密度高，避免"亲""哦""呢"等过度亲昵用词
        每段回复控制在3-5句话，关键信息加【】强调
        
        【禁止事项】
        - 禁止承诺"一定""保证"等绝对化表述
        - 禁止询问用户身份证号、密码等敏感信息
        - 禁止对未发生的交易做任何预测
        
        【输出格式】
        回复必须以JSON格式输出，包含以下字段：
        {
            "reply": "正式回复内容",
            "action_required": "用户需要执行的操作，无则填null",
            "confidence": 0.0-1.0之间的小数，表示对答案的把握程度
        }
        """
    
    @staticmethod
    def get_internal_qa_prompt() -> str:
        return """
        【角色定位】
        你是企业内部知识助手，帮助员工快速查找公司制度、流程、技术文档。
        
        【核心原则】
        1. 必须基于提供的【背景知识】回答，不得使用外部知识
        2. 如果【背景知识】中找不到答案，直接回复"未在知识库中找到相关信息"
        3. 引用时必须标注来源文档名称和章节
        
        【可信度分级】
        - 高置信度：背景知识中直接有明确答案 → 可直接回答
        - 中置信度：背景知识中有相关信息但需推断 → 回答时加"根据现有资料推测"
        - 低置信度：背景知识中无相关信息 → 明确告知未知
        
        【输出格式】
        回复中须附上引用来源，格式：[来源：文档名-章节]
        """

Prompt工程的本质：不是“写一段话让模型理解”，而是给大模型制定一套可执行的规则系统。好的Prompt像一份详尽的员工手册，告诉模型什么该做、什么不该做、怎么做、做到什么标准。

04 现实世界的避坑实录

坑一：“上下文窗口无限大”是毒鸡汤

某次上线，团队天真地认为“反正模型有128K上下文，把整个知识库都塞进去不就行了”。结果：

• 首Token延迟从800ms飙升到8.7秒
• 用户等不及，纷纷关掉页面
• 月底收到云账单，费用翻了17倍

教训：RAG不是上下文塞满，而是精准检索。Top-k值不是越大越好，3-5个片段通常是性价比最优解。

坑二：流式响应被“截胡”

前端调接口等了10秒没反应，排查发现Nginx默认开启了proxy_buffering，把所有数据攒够才返回。用户看到的是“空白页→瞬间满屏文字”，体验极差。

解决方案：

location /api/ai/ {
    proxy_pass http://ai-backend/;
    proxy_buffering off;           # 关键配置
    proxy_cache off;               # 关闭缓存
    proxy_read_timeout 300s;       # 大模型思考需要时间
    chunked_transfer_encoding on;  # 启用分块传输
}

坑三：数据权限“形同虚设”

前端传来{"project_id": "proj_001"}，后端直接透传给Python引擎。某个懂技术的员工用Postman把project_id改成proj_002，看到了不该看的竞品分析报告。

红线原则：任何来自前端的权限参数，都必须在后端二次校验。永远不要把权限校验交给前端或AI层。

05 从“能聊天”到“能干活”，只差一个你

我们正在做的，就是把这套“管教体系”产品化、平台化——让大模型不只是会对话，而是能真正接入企业的业务流、数据流、权限体系，成为能独立完成任务的“数字员工”。

如果你具备：

• 扎实的工程底子（Java + Python双修，熟悉Spring Boot和FastAPI）
• 全栈视野（前端React/Vue不陌生，能理解从界面到数据库的完整链路）
• AI应用经验（RAG、Agent、Prompt Engineering至少有一个方向有实战）
• 产品思维（懂业务、懂用户，不为了炫技而堆砌技术）

那么，欢迎加入我们。

一起让大模型从“能聊天”变成“能干活”。

---

简历投递：recruit@ai-engineering.com
标题请注明：【AI全栈】+ 姓名
我们会在24小时内回复