智能应用开发：从 LLM 集成到生产级智能系统的专业实践

原创

用户12687280

发布于 2026-09-18 18:10:59

1680

摘要

智能应用（Intelligent Application）不是“给传统应用加一个聊天框”，而是以大模型为推理内核，融合检索、工具、记忆、多模态与业务规则，构建能够理解意图、自主决策、调用服务并持续进化的软件系统。它既不同于纯 LLM 调用，也不同于传统规则引擎，而是在概率性生成与确定性业务之间建立可控的工程边界。专业落地的关键，不在于追逐最大模型，而在于理解上下文工程、能力编排、状态管理、评估闭环与安全护栏之间的系统性权衡。本文从架构分层、核心能力、开发模式、代码实战、评估、安全与生产化等维度，给出一套可落地的智能应用工程方法。

关键词：智能应用；LLM；RAG；Function Calling；Agent；上下文工程；评估；安全护栏

1. 智能应用的定位

智能应用的核心公式：

Intelligent App = LLM + Context + Retrieval + Tools + Memory + Guardrails

与传统应用的区别：

维度	传统应用	智能应用
输入	表单、参数	自然语言、多模态
逻辑	确定性规则	概率性推理 + 规则兜底
输出	结构化数据	生成内容 + 结构化动作
交互	固定流程	动态意图理解
失败模式	明确异常	幻觉、偏差、不确定性
测试	断言明确	评估集 + 人工校准

智能应用不是替代传统应用，而是在其上加一层“理解与决策”的能力。真正的专业系统，是概率组件与确定性组件的混合架构。

专业原则：

能用规则解决的，不要用模型；
能用检索解决的，不要用微调；
能用简单 Prompt 解决的，不要上 Agent；
所有生成内容必须可校验、可追溯、可回滚；
评估与可观测性是生产化的前提。

2. 架构分层

交互层：对话、搜索、Copilot、嵌入式助手
理解层：意图识别、实体抽取、查询改写
编排层：Prompt、Chain、Graph、Tool Calling、Memory
能力层：RAG、函数调用、多模态、业务 API
推理层：OpenAI / vLLM / TGI / Ollama / 本地模型
数据层：向量库、文档解析、Embedding、缓存、特征
治理层：评估、监控、限流、成本、安全、灰度

常用 Python 技术栈：

API 与编排：openai、langchain、langgraph、llama-index、dspy；
本地推理：transformers、vllm、ollama；
向量检索：faiss、chromadb、qdrant-client、pgvector；
服务化：fastapi、uvicorn、celery、redis；
评估：ragas、deepeval、promptfoo；
可观测性：opentelemetry、langfuse、langsmith。

3. 核心能力一：上下文工程与 Prompt

智能应用的第一能力不是“写提示词”，而是上下文工程：把正确的信息、以正确的格式、在正确的时机注入模型。

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY", "EMPTY"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
)

SYSTEM_PROMPT = """你是企业智能助手，负责回答产品与订单问题。

规则：
1. 只基于给定上下文回答，不知道就说“我需要转人工”。
2. 不承诺价格、库存、时效以外的内容。
3. 输出使用简体中文，不超过 150 字。
4. 涉及退款、投诉、法律问题，必须转人工。
"""


def build_messages(
    question: str,
    context: str,
    history: list[dict] | None = None,
) -> list[dict]:
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    if history:
        messages.extend(history[-6:])  # 保留最近 6 轮
    messages.append({
        "role": "user",
        "content": f"参考资料：\n{context}\n\n用户问题：{question}",
    })
    return messages


def ask(question: str, context: str = "", history: list[dict] | None = None) -> str:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=build_messages(question, context, history),
        temperature=0.2,
        max_tokens=512,
    )
    return resp.choices[0].message.content

上下文工程要点：

系统提示定义角色、边界、格式与安全规则；
历史消息截断与摘要，避免上下文膨胀；
检索结果去重、排序、压缩；
结构化输出用 schema 约束；
关键规则放在系统提示，而非用户消息。

4. 核心能力二：RAG 检索增强

RAG 是智能应用拥有事实依据的核心手段。

import numpy as np
from openai import OpenAI

client = OpenAI(api_key="EMPTY", base_url="http://localhost:8000/v1")


def embed(texts: list[str]) -> np.ndarray:
    resp = client.embeddings.create(model="text-embedding-3-small", input=texts)
    return np.array([d.embedding for d in resp.data], dtype=np.float32)


class VectorStore:
    def __init__(self):
        self.docs: list[str] = []
        self.metadata: list[dict] = []
        self.mat: np.ndarray | None = None

    def add(self, docs: list[str], metadata: list[dict] | None = None):
        self.docs.extend(docs)
        self.metadata.extend(metadata or [{} for _ in docs])
        vecs = embed(docs)
        self.mat = vecs if self.mat is None else np.vstack([self.mat, vecs])

    def search(
        self,
        query: str,
        top_k: int = 3,
        filter_fn=None,
    ) -> list[tuple[str, dict, float]]:
        q = embed([query])[0]
        sims = self.mat @ q / (
            np.linalg.norm(self.mat, axis=1) * np.linalg.norm(q) + 1e-8
        )
        idx = np.argsort(-sims)
        results = []
        for i in idx:
            if filter_fn and not filter_fn(self.metadata[i]):
                continue
            results.append((self.docs[i], self.metadata[i], float(sims[i])))
            if len(results) >= top_k:
                break
        return results


store = VectorStore()
store.add(
    [
        "退货政策：签收后 7 天内可无理由退货，商品需不影响二次销售。",
        "发货时效：现货商品 48 小时内发出，预售以详情页为准。",
        "发票：支持电子普票，下单时勾选，发货后 3 个工作日开出。",
    ],
    metadata=[
        {"source": "policy", "tenant": "default"},
        {"source": "logistics", "tenant": "default"},
        {"source": "invoice", "tenant": "default"},
    ],
)


def rag_answer(question: str, tenant: str = "default") -> str:
    results = store.search(
        question,
        top_k=3,
        filter_fn=lambda m: m.get("tenant") == tenant,
    )
    context = "\n".join(doc for doc, _, _ in results)
    return ask(question, context)


print(rag_answer("我买的衣服不合适能退吗？"))

生产级 RAG 还需要：

文档解析：PDF、HTML、Word、表格、OCR；
切分策略：递归、语义、按标题层级；
元数据：来源、时间、权限、版本；
混合检索：向量 + BM25 + 关键词；
重排序：Cross-Encoder；
查询改写：HyDE、多查询、子问题分解；
权限过滤：不同用户检索不同文档；
引用溯源：回答附带来源。

5. 核心能力三：Function Calling 与工具

工具让智能应用从“会说”变成“会做”。

import ast
import json
import operator
from dataclasses import dataclass
from typing import Callable, Any


@dataclass
class Tool:
    name: str
    description: str
    func: Callable[..., Any]
    schema: dict


class ToolRegistry:
    def __init__(self):
        self._tools: dict[str, Tool] = {}

    def register(self, tool: Tool):
        self._tools[tool.name] = tool

    def get(self, name: str) -> Tool:
        if name not in self._tools:
            raise ValueError(f"未知工具: {name}")
        return self._tools[name]

    def to_openai_tools(self) -> list[dict]:
        return [
            {
                "type": "function",
                "function": {
                    "name": t.name,
                    "description": t.description,
                    "parameters": t.schema,
                },
            }
            for t in self._tools.values()
        ]


def safe_calculator(expression: str) -> str:
    ops = {
        ast.Add: operator.add,
        ast.Sub: operator.sub,
        ast.Mult: operator.mul,
        ast.Div: operator.truediv,
        ast.Pow: operator.pow,
        ast.USub: operator.neg,
    }

    def _eval(node):
        if isinstance(node, ast.Expression):
            return _eval(node.body)
        if isinstance(node, ast.Constant):
            return node.value
        if isinstance(node, ast.BinOp):
            return ops[type(node.op)](_eval(node.left), _eval(node.right))
        if isinstance(node, ast.UnaryOp):
            return ops[type(node.op)](_eval(node.operand))
        raise ValueError("不支持的表达式")

    return str(_eval(ast.parse(expression, mode="eval")))


def query_order(order_id: str) -> str:
    # 模拟订单查询，真实场景接入业务 API
    fake_db = {
        "A1001": "已发货，预计明天送达",
        "A1002": "待支付",
        "A1003": "已签收",
    }
    return fake_db.get(order_id, "未找到该订单")


registry = ToolRegistry()
registry.register(Tool(
    name="calculator",
    description="执行安全数学表达式计算",
    func=safe_calculator,
    schema={
        "type": "object",
        "properties": {"expression": {"type": "string"}},
        "required": ["expression"],
    },
))
registry.register(Tool(
    name="query_order",
    description="根据订单号查询订单状态",
    func=query_order,
    schema={
        "type": "object",
        "properties": {"order_id": {"type": "string"}},
        "required": ["order_id"],
    },
))

工具设计原则：

名称与描述清晰，避免误调用；
参数必须校验；
高风险工具需人工确认；
工具异常转为观察结果；
记录每次调用的日志与耗时。

6. 核心能力四：Agent 循环与记忆

6.1 ReAct Agent

import json
import logging

logger = logging.getLogger("agent")


class Agent:
    def __init__(
        self,
        registry: ToolRegistry,
        model: str = "gpt-4o-mini",
        system_prompt: str = "你是一个严谨的智能助手。",
        max_steps: int = 8,
    ):
        self.registry = registry
        self.model = model
        self.system_prompt = system_prompt
        self.max_steps = max_steps
        self.messages: list[dict] = [{"role": "system", "content": system_prompt}]
        self.step_count = 0

    def run(self, user_input: str) -> str:
        self.messages.append({"role": "user", "content": user_input})

        for step in range(1, self.max_steps + 1):
            self.step_count = step
            resp = client.chat.completions.create(
                model=self.model,
                messages=self.messages,
                tools=self.registry.to_openai_tools(),
                tool_choice="auto",
                temperature=0,
            )
            msg = resp.choices[0].message

            assistant_msg = {"role": "assistant", "content": msg.content or ""}
            if msg.tool_calls:
                assistant_msg["tool_calls"] = [
                    {
                        "id": call.id,
                        "type": "function",
                        "function": {
                            "name": call.function.name,
                            "arguments": call.function.arguments,
                        },
                    }
                    for call in msg.tool_calls
                ]
            self.messages.append(assistant_msg)

            if not msg.tool_calls:
                return msg.content or ""

            for call in msg.tool_calls:
                name = call.function.name
                try:
                    args = json.loads(call.function.arguments or "{}")
                except json.JSONDecodeError:
                    args = {}

                logger.info("step=%s tool=%s args=%s", step, name, args)

                try:
                    tool = self.registry.get(name)
                    result = tool.func(**args)
                except Exception as e:
                    result = f"工具执行失败: {e}"

                self.messages.append({
                    "role": "tool",
                    "tool_call_id": call.id,
                    "content": str(result),
                })

        return "达到最大步数，任务未完成。"


agent = Agent(registry=registry)
print(agent.run("订单 A1001 到哪了？再帮我算一下 12*(3+4)。"))

6.2 长期记忆

class LongTermMemory:
    def __init__(self):
        self.texts: list[str] = []
        self.vectors: np.ndarray | None = None

    def _embed(self, texts: list[str]) -> np.ndarray:
        resp = client.embeddings.create(
            model="text-embedding-3-small", input=texts
        )
        return np.array([d.embedding for d in resp.data], dtype=np.float32)

    def add(self, text: str):
        self.texts.append(text)
        vec = self._embed([text])
        self.vectors = vec if self.vectors is None else np.vstack([self.vectors, vec])

    def search(self, query: str, top_k: int = 3) -> list[str]:
        if self.vectors is None:
            return []
        q = self._embed([query])[0]
        sims = self.vectors @ q / (
            np.linalg.norm(self.vectors, axis=1) * np.linalg.norm(q) + 1e-8
        )
        idx = np.argsort(-sims)[:top_k]
        return [self.texts[i] for i in idx]

记忆分层：短期对话、长期知识、任务状态。生产环境需注意权限隔离、脱敏、过期与冲突处理。

7. 完整智能应用示例：企业智能助手

class IntelligentAssistant:
    def __init__(self, registry: ToolRegistry, vector_store: VectorStore):
        self.agent = Agent(registry=registry)
        self.vector_store = vector_store
        self.memory = LongTermMemory()

    def chat(self, question: str, tenant: str = "default") -> str:
        # 1. 检索知识
        results = self.vector_store.search(
            question,
            top_k=3,
            filter_fn=lambda m: m.get("tenant") == tenant,
        )
        context = "\n".join(doc for doc, _, _ in results)

        # 2. 检索长期记忆
        memories = self.memory.search(question, top_k=2)
        memory_text = "\n".join(memories)

        # 3. 构建增强输入
        enriched = (
            f"用户问题：{question}\n\n"
            f"知识库：\n{context}\n\n"
            f"历史记忆：\n{memory_text}"
        )

        # 4. Agent 执行
        answer = self.agent.run(enriched)

        # 5. 写入记忆
        self.memory.add(f"Q: {question}\nA: {answer[:200]}")

        return answer


assistant = IntelligentAssistant(registry=registry, vector_store=store)
print(assistant.chat("退货政策是什么？顺便查一下订单 A1002。"))

8. 评估闭环

没有评估，就没有智能应用的迭代。

import json
import time


def evaluate(assistant: IntelligentAssistant, cases: list[dict]) -> dict:
    results = []
    for case in cases:
        start = time.time()
        try:
            output = assistant.chat(case["input"])
            success = case["check"](output)
        except Exception as e:
            output = str(e)
            success = False
        results.append({
            "success": success,
            "latency": time.time() - start,
            "output": output,
        })

    total = len(results)
    return {
        "success_rate": sum(r["success"] for r in results) / total,
        "avg_latency": sum(r["latency"] for r in results) / total,
        "details": results,
    }


def llm_judge(question: str, answer: str, reference: str) -> dict:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": "你是严格评审，只输出 JSON。"},
            {
                "role": "user",
                "content": (
                    f"问题：{question}\n回答：{answer}\n参考答案：{reference}\n"
                    '输出 {"relevance":0-10,"factuality":0-10,'
                    '"completeness":0-10,"safety":0-10,"comments":""}'
                ),
            },
        ],
        temperature=0,
    )
    return json.loads(resp.choices[0].message.content)

评估维度：

相关性、事实性、完整性、格式合规；
工具调用准确率与冗余率；
延迟 P50/P95；
token 成本；
安全违规率；
人工接管率。

9. 生产化：服务、监控与安全

9.1 FastAPI 服务

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

app = FastAPI(title="Intelligent Assistant")
assistant = IntelligentAssistant(registry=registry, vector_store=store)


class ChatRequest(BaseModel):
    question: str = Field(..., min_length=1, max_length=2000)
    tenant: str = Field("default")


@app.post("/chat")
def chat(req: ChatRequest):
    try:
        answer = assistant.chat(req.question, req.tenant)
        return {"answer": answer}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/health/live")
def live():
    return {"status": "ok"}


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

9.2 安全护栏

BANNED_WORDS = ["违法", "暴力", "仇恨", "隐私泄露"]
DANGEROUS_TOOLS = {"shell", "delete_file", "execute_sql"}


def guard_output(text: str) -> bool:
    return not any(word in text for word in BANNED_WORDS)


def guard_tool_call(tool_name: str, user_role: str):
    if tool_name in DANGEROUS_TOOLS and user_role != "admin":
        raise PermissionError(f"无权调用工具: {tool_name}")
    return True

生产清单：

工具白名单与权限模型；
输入输出审核；
最大步数、最大 token、最大成本；
超时、重试、熔断、降级；
人工确认高风险操作；
全链路日志与审计；
评估集与回归测试；
灰度发布与回滚；
数据脱敏与隔离；
遵守 OWASP LLM Top 10。

10. 常见反模式

给传统应用硬塞聊天框，不改变交互设计；
用 LLM 替代确定性规则；
不做 RAG，靠模型记忆回答事实；
不做评估就上线；
无最大步数与成本控制；
让模型直接执行危险操作；
上下文无限增长；
无日志、无回放、无版本管理；
生成内容不审核就发布；
忽视权限隔离与提示注入。

11. 结论

智能应用开发是 LLM 工程的核心实践。它要求我们把大模型嵌入真实业务系统，通过上下文工程控制行为，通过 RAG 提供事实，通过 Function Calling 连接服务，通过 Agent 循环实现多步任务，通过记忆保持连续性，通过评估闭环持续迭代，通过安全护栏控制风险。真正专业的智能应用，不是最像人的应用，而是最可靠、最可解释、最可治理的系统。它不是在概率与确定性之间二选一，而是在二者之间建立可控的工程边界。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

智能创作

智能硬件