AI 应用与智能体开发：从 LLM 调用到生产级 Agent 系统的专业实践

原创

资源大佬 jzit-top

发布于 2026-09-18 15:46:57

1520

摘要

AI 应用与智能体（Agent）开发正在从“调用大模型 API”走向“构建具备感知、规划、行动、记忆与协作能力的自治系统”。AI 应用的核心是用 LLM 解决具体业务问题，而智能体则进一步引入工具调用、多步推理、状态管理与环境交互。专业落地的关键，不在于追逐最大模型，而在于理解概率性组件带来的不确定性，并在能力、延迟、成本、可控性与安全之间做出系统性权衡。本文从技术栈分层、AI 应用架构、Agent 核心循环、工具协议、记忆、多智能体、框架选型、评估、安全与生产化等维度，给出一套可落地的工程方法，并附带可直接运行的 Python 代码。

关键词：AI 应用；智能体；Agent；Function Calling；ReAct；RAG；LangGraph；评估；安全护栏

1. 从 AI 应用到智能体：能力演进

AI 应用与智能体的关系可以抽象为：

AI 应用 = LLM + Prompt + Context + API
智能体 = LLM + Planning + Memory + Tools + Action Loop

AI 应用：以 LLM 为核心，完成问答、摘要、翻译、分类、生成等任务；
智能体：在 AI 应用基础上增加自主决策、多步执行、工具调用与环境反馈。

二者不是替代关系，而是复杂度递进。大多数业务从 AI 应用起步，只有任务需要多步动态决策时才引入 Agent。

专业原则：

简单任务用 Prompt + API，不要过度 Agent 化；
所有工具调用必须白名单、参数校验、权限控制；
所有关键操作必须可中断、可审计、可回滚；
评估与可观测性是生产化的前提；
安全护栏不是可选项，而是基础设施。

2. 技术栈分层

应用层：客服、写作、搜索、Copilot、RPA、数据分析
编排层：Prompt、Chain、Graph、Tool Calling、Memory
能力层：RAG、微调、函数调用、多模态
推理层：OpenAI API / vLLM / TGI / llama.cpp / Ollama
数据层：向量库、文档解析、Embedding、缓存
工程层：评估、监控、限流、成本、安全、灰度

常用 Python 库：

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

3. AI 应用基础：LLM 调用与结构化输出

3.1 基础调用

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"),
)

resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "你是一个严谨的中文技术助手。"},
        {"role": "user", "content": "用一句话解释什么是智能体。"},
    ],
    temperature=0.3,
    max_tokens=256,
)
print(resp.choices[0].message.content)

3.2 结构化输出

import json
from pydantic import BaseModel, Field, ValidationError


class Task(BaseModel):
    title: str = Field(description="任务标题")
    priority: str = Field(description="优先级：high/medium/low")
    tags: list[str] = Field(default_factory=list)
    estimate_hours: float = Field(description="预估工时")


def extract_task(text: str) -> Task:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": "你只输出 JSON。"},
            {
                "role": "user",
                "content": (
                    f"从以下文本抽取任务信息：\n{text}\n"
                    f"schema={Task.model_json_schema()}"
                ),
            },
        ],
        temperature=0,
    )
    raw = resp.choices[0].message.content
    try:
        return Task.model_validate_json(raw)
    except ValidationError as e:
        # 让模型自我修正一次
        fix = client.chat.completions.create(
            model="gpt-4o-mini",
            response_format={"type": "json_object"},
            messages=[
                {"role": "system", "content": "修正 JSON 使其符合 schema。"},
                {
                    "role": "user",
                    "content": (
                        f"schema={Task.model_json_schema()}\n"
                        f"json={raw}\nerror={e}"
                    ),
                },
            ],
            temperature=0,
        )
        return Task.model_validate_json(fix.choices[0].message.content)


task = extract_task("下周一前完成用户登录模块，优先级高，预计 16 小时。")
print(task)

结构化输出是 AI 应用与业务系统之间的桥梁。生产环境必须校验、修复、记录失败。

4. RAG：让 AI 应用拥有事实依据

RAG 是当前最实用的 LLM 落地范式：用检索提供事实，用生成组织语言。

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.mat: np.ndarray | None = None

    def add(self, docs: list[str]):
        self.docs.extend(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) -> list[str]:
        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)[:top_k]
        return [self.docs[i] for i in idx]


store = VectorStore()
store.add([
    "智能体通过工具调用与环境交互，完成多步任务。",
    "ReAct 让模型在推理与行动之间交替，提升可解释性。",
    "Function Calling 是让模型输出结构化工具调用的标准方式。",
])


def rag_answer(question: str) -> str:
    context = "\n".join(store.search(question))
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "只基于给定资料回答，不知道就说不知道。"},
            {"role": "user", "content": f"资料：\n{context}\n\n问题：{question}"},
        ],
        temperature=0.2,
    )
    return resp.choices[0].message.content


print(rag_answer("ReAct 是什么？"))

生产级 RAG 还需要文档解析、切分、元数据、混合检索、重排序、查询改写、权限过滤与引用溯源。

5. 工具协议：让 LLM 连接外部世界

工具是 Agent 的核心能力。生产环境优先使用 Function Calling，而非让模型输出自由文本再解析。

import ast
import json
import logging
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 search(query: str) -> str:
    return f"[搜索结果] 关于「{query}」的资料：..."


registry = ToolRegistry()
registry.register(Tool(
    name="search",
    description="搜索互联网或知识库",
    func=search,
    schema={
        "type": "object",
        "properties": {"query": {"type": "string"}},
        "required": ["query"],
    },
))
registry.register(Tool(
    name="calculator",
    description="执行安全数学表达式计算",
    func=safe_calculator,
    schema={
        "type": "object",
        "properties": {"expression": {"type": "string"}},
        "required": ["expression"],
    },
))

工具设计原则：

名称与描述清晰，避免模型误调用；
参数必须校验，禁止直接执行 Shell、SQL、文件删除；
工具异常转为观察结果，而非中断循环；
高风险工具需人工确认。

6. ReAct Agent：观察—思考—行动循环

import json
import logging
from openai import OpenAI

client = OpenAI(api_key="EMPTY", base_url="http://localhost:8000/v1")
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,
    system_prompt=(
        "你是一个严谨的助手。需要计算时调用 calculator，"
        "需要事实时调用 search。不确定时明确说明，不要编造。"
    ),
)

print(agent.run("计算 12*(3+4)，并搜索 ReAct 的核心思想。"))

工程要点：温度设为 0、记录每步日志、工具异常转为观察结果、最大步数防无限循环。

7. 规划与反思：Plan-and-Execute 与 Reflexion

ReAct 适合短任务。复杂任务需要显式规划与反思。

class Planner:
    def __init__(self, model: str = "gpt-4o-mini"):
        self.model = model

    def plan(self, task: str) -> list[str]:
        resp = client.chat.completions.create(
            model=self.model,
            temperature=0,
            messages=[
                {
                    "role": "system",
                    "content": (
                        "你是任务规划器。将任务拆成 3-7 个可执行步骤，"
                        "每行一个步骤，不要编号，不要解释。"
                    ),
                },
                {"role": "user", "content": task},
            ],
        )
        text = resp.choices[0].message.content or ""
        return [line.strip() for line in text.splitlines() if line.strip()]


class Critic:
    def __init__(self, model: str = "gpt-4o-mini"):
        self.model = model

    def review(self, task: str, result: str) -> str:
        resp = client.chat.completions.create(
            model=self.model,
            temperature=0,
            messages=[
                {
                    "role": "system",
                    "content": (
                        "你是评审员。检查结果是否完成任务，"
                        "指出遗漏、错误与改进建议。"
                    ),
                },
                {"role": "user", "content": f"任务：{task}\n结果：{result}"},
            ],
        )
        return resp.choices[0].message.content or ""


def reflexion_loop(agent: Agent, critic: Critic, task: str, max_rounds: int = 3):
    result = agent.run(task)
    for _ in range(max_rounds):
        review = critic.review(task, result)
        if "通过" in review or "完成" in review:
            return result
        result = agent.run(f"根据评审意见修正：\n{review}\n原结果：{result}")
    return result

规划与反思的代价是更多 token 与延迟，适合高价值、可验证任务。

8. 记忆：短期、长期与状态

Agent 记忆分三层：

短期记忆：对话消息列表；
长期记忆：向量库、知识图谱、数据库；
任务状态：计划、步骤、中间结果、预算。

import numpy as np


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]

生产环境需注意权限隔离、数据脱敏、记忆过期与冲突处理。

9. 多智能体协作

多 Agent 不是越多越好。常见角色：Planner、Researcher、Executor、Critic、Coordinator。

class MultiAgentSystem:
    def __init__(self, planner: Planner, agent: Agent, critic: Critic):
        self.planner = planner
        self.agent = agent
        self.critic = critic

    def run(self, task: str) -> str:
        plan = self.planner.plan(task)
        context = ""
        for i, step in enumerate(plan, 1):
            result = self.agent.run(
                f"当前步骤 {i}: {step}\n已有上下文：{context}"
            )
            context += f"\n步骤{i}结果：{result}"

        review = self.critic.review(task, context)
        if "通过" in review or "完成" in review:
            return context
        return self.agent.run(f"根据评审修正：{review}\n原结果：{context}")

多 Agent 挑战：通信开销、上下文膨胀、角色冲突、成本线性增长、调试困难。建议优先单 Agent + 工具 + 规划。

10. 框架选型

框架	定位	适用场景
LangChain	LLM 应用编排	快速原型、RAG、工具调用
LangGraph	状态图编排	有状态、可中断、多 Agent
AutoGen	多 Agent 对话	研究、协作任务
CrewAI	角色化多 Agent	业务流程模拟
Semantic Kernel	企业级编排	.NET/Python 混合
OpenAI Assistants	托管 Agent	快速接入、托管线程
Dify / Coze	低代码平台	业务人员搭建

选型原则：状态复杂度决定框架。简单工具调用不需要重框架；有状态、可中断、多分支、多 Agent 时，LangGraph 这类状态图更合适。

11. 评估与可观测性

没有评估，就没有 Agent 工程。评估指标：任务成功率、平均步数、工具调用准确率、延迟 P50/P95、token 成本、安全违规率、人工接管率。

import time


def evaluate_agent(agent: Agent, cases: list[dict]) -> dict:
    results = []
    for case in cases:
        start = time.time()
        try:
            output = agent.run(case["input"])
            success = case["check"](output)
        except Exception as e:
            output = str(e)
            success = False
        results.append({
            "success": success,
            "latency": time.time() - start,
            "steps": agent.step_count,
            "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,
        "avg_steps": sum(r["steps"] for r in results) / total,
        "details": results,
    }

可观测性：每步记录 trace_id、step、tool、args、latency、tokens；接入 OpenTelemetry、LangSmith、LangFuse 或自建看板。

12. 生产化：FastAPI 服务与安全护栏

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

app = FastAPI(title="Agent Service")
agent = Agent(registry=registry)

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


class AgentRequest(BaseModel):
    input: str = Field(..., min_length=1, max_length=2000)
    user_role: str = Field("user")


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


@app.post("/agent/run")
def run(req: AgentRequest):
    try:
        output = agent.run(req.input)
        if not guard_output(output):
            raise HTTPException(status_code=400, detail="输出包含违规内容")
        return {"output": output}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


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

生产清单：

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

13. 常见反模式

把 Agent 当万能，简单任务也上多 Agent；
没有最大步数与预算控制；
让模型直接执行 Shell、SQL、文件删除；
工具描述模糊，导致误调用；
不做评估就上线；
没有日志与回放能力；
上下文无限增长；
用 Agent 解决知识更新问题，而不是 RAG；
多 Agent 角色重叠，成本失控；
忽视提示注入与权限隔离。

14. 结论

AI 应用与智能体开发是 LLM 工程的两个层次：前者以 Prompt、RAG、结构化输出和函数调用解决具体业务问题；后者在 AI 应用基础上引入规划、记忆、工具、反思与多 Agent 协作，形成自治系统。专业实践要求我们从 Function Calling 与 ReAct 循环起步，逐步引入 Planner、Critic、长期记忆和多 Agent；工程上必须建立评估、可观测性、成本治理与安全护栏。真正生产级的 Agent，不是最像人的 Agent，而是最可靠、最可解释、最可治理的 Agent。

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

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

LLM

AI Agent安全网关