大模型应用开发实战：从Prompt工程到多Agent协作系统

原创

用户12608867

发布于 2026-08-21 13:40:03

2110

大模型应用开发实战：从Prompt工程到多Agent协作系统

本文基于“JK时间-AI大模型应用开发实战营”的实战项目经验，系统梳理了从Prompt设计、RAG检索增强生成到多Agent协作的全链路开发方法，并附有可落地的代码实现与性能优化策略。

一、引言：为什么需要系统化的大模型应用开发？

大模型（LLM）的推理能力已足够强大，但将其转化为稳定、可控、可扩展的生产级应用，依然面临诸多挑战：

Prompt敏感性：微小的措辞变化可能导致输出质量剧烈波动；
知识时效性与幻觉：模型固有知识截止日期限制，且容易“编造”事实；
复杂任务分解：单一模型难以完成多步骤、多工具调用的业务逻辑；
成本与延迟：每次请求都携带大量上下文，Token消耗和响应时间难以控制。

本实战营的核心理念是 “工程化驯服大模型” ，通过分层架构、标准化工具链和可观测性设计，让LLM真正服务于业务。下文将逐步拆解一个企业级智能客服助手的完整实现，涵盖Prompt工程、RAG、Function Calling及多Agent协作。

二、基础层：Prompt工程与结构化输出

2.1 动态Prompt模板与版本管理

生产环境中，Prompt不是静态字符串，而是可配置、可回滚的模板资产。我们使用Jinja2 + YAML配置管理：

# prompt_config.yaml
templates:
  customer_service:
    system: |
      你是一名{{ role }}，擅长{{ expertise }}。
      当前时间：{{ current_time }}
      公司政策：{{ policy_summary }}
      回答要求：
      1. 始终使用{{ language }}回复
      2. 若无法确定，请明确告知用户并建议人工通道
      3. 输出必须为合法JSON，包含"answer"和"confidence"字段
    user: |
      用户问题：{{ user_query }}
      对话历史：{{ history }}

# loader.py
import yaml
from jinja2 import Template
from datetime import datetime

class PromptManager:
    def __init__(self, config_path):
        with open(config_path) as f:
            self.config = yaml.safe_load(f)
    
    def render(self, template_key, **kwargs):
        template_str = self.config['templates'][template_key]
        # 注入默认变量
        kwargs.setdefault('current_time', datetime.now().strftime('%Y-%m-%d %H:%M'))
        kwargs.setdefault('language', '中文')
        return Template(template_str).render(**kwargs)

2.2 结构化输出：约束解码与Pydantic校验

为避免模型输出自由文本导致的解析错误，我们采用JSON Mode + Pydantic Schema双重约束：

from pydantic import BaseModel, Field
from typing import Optional
import json
from openai import OpenAI

class ServiceResponse(BaseModel):
    answer: str = Field(description="给用户的最终回复")
    confidence: float = Field(ge=0.0, le=1.0, description="模型置信度")
    need_human: bool = Field(default=False, description="是否需要转人工")
    action_items: Optional[list[str]] = Field(default=None, description="待办事项列表")

client = OpenAI(api_key="your-key")

def get_structured_response(user_query: str, history: list) -> ServiceResponse:
    pm = PromptManager("prompt_config.yaml")
    system_prompt = pm.render("customer_service", role="资深客服专家", expertise="售后与退换货", policy_summary="7天无理由，运费自理")
    user_prompt = pm.render("user", user_query=user_query, history="\n".join(history))
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ],
        response_format={"type": "json_object"},
        temperature=0.2
    )
    raw = json.loads(response.choices[0].message.content)
    # 自动校验并转换
    return ServiceResponse(**raw)

关键点：我们强制模型输出JSON，并用Pydantic做运行时校验，当confidence < 0.6或need_human=True时触发路由转接。

三、RAG：企业知识库的精准检索增强

3.1 多路召回 + 重排序（Rerank）架构

单纯使用向量相似度检索容易受到“语义漂移”影响，我们采用 BM25（关键词） + 向量（稠密） 双路召回，再通过交叉编码器重排序，提升Top-K准确率。

向量索引（FAISS）与BM25索引构建：

from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain_community.retrievers import BM25Retriever
from langchain.schema import Document

# 假设docs为List[Document]，包含page_content和metadata
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = FAISS.from_documents(docs, embeddings)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 20})

bm25_retriever = BM25Retriever.from_documents(docs)
bm25_retriever.k = 20

def hybrid_recall(query: str):
    vec_docs = vector_retriever.invoke(query)
    bm_docs = bm25_retriever.invoke(query)
    # 合并去重（基于文档ID）
    all_docs = {doc.metadata.get("id"): doc for doc in vec_docs + bm_docs}
    return list(all_docs.values())

重排序（使用Cohere或BGE-reranker）：

from sentence_transformers import CrossEncoder

# 下载BGE-reranker模型
reranker = CrossEncoder('BAAI/bge-reranker-v2-m3', max_length=512)

def rerank(query: str, candidates: list[Document], top_k: int = 5):
    pairs = [[query, doc.page_content] for doc in candidates]
    scores = reranker.predict(pairs)  # 返回相似度分数
    sorted_idx = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)
    return [candidates[i] for i in sorted_idx[:top_k]]

# 完整RAG流程
def retrieve_context(query: str):
    candidates = hybrid_recall(query)
    top_docs = rerank(query, candidates, top_k=5)
    return "\n\n".join([f"[{doc.metadata.get('source')}]\n{doc.page_content}" for doc in top_docs])

3.2 上下文压缩与Token优化

检索到的文档可能很长，直接拼接会浪费Token且引入噪声。我们使用LLMLingua进行压缩：

from llmlingua import PromptCompressor

compressor = PromptCompressor(model_name="microsoft/llmlingua-2-bert-base")
def compress_context(context: str, target_ratio=0.5):
    compressed = compressor.compress_prompt(
        context,
        target_ratio=target_ratio,
        condition_in_question="回答以下问题，需要基于提供的参考信息",
    )
    return compressed['compressed_prompt']

这样在最终送入LLM前，上下文体积缩小约50%，而关键信息保留率超过90%。

四、Function Calling：让模型主动调用外部工具

智能客服需要查订单、退换货、物流跟踪等能力，我们通过OpenAI的Function Calling或MCP协议实现工具调用。

定义工具Schema：

tools = [
    {
        "type": "function",
        "function": {
            "name": "query_order_status",
            "description": "根据订单号查询当前物流状态和预计送达时间",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {"type": "string", "description": "订单编号，格式为ORD-xxxxxx"},
                },
                "required": ["order_id"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "create_return_request",
            "description": "为指定订单创建退货申请，需要用户确认商品是否已拆封",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {"type": "string"},
                    "reason": {"type": "string", "enum": ["质量问题", "尺寸不符", "其他"]},
                    "is_unopened": {"type": "boolean"}
                },
                "required": ["order_id", "reason"]
            }
        }
    }
]

工具执行器与模型循环：

def execute_tool(tool_call):
    name = tool_call.function.name
    args = json.loads(tool_call.function.arguments)
    if name == "query_order_status":
        # 模拟调用订单API
        return f"订单{args['order_id']}当前在【广州中转站】，预计2026-08-23送达"
    elif name == "create_return_request":
        # 模拟创建退货
        return f"退货申请已提交，单号RET-{args['order_id']}，请等待审核"
    return "未知工具"

def chat_with_tools(user_message, history):
    messages = history + [{"role": "user", "content": user_message}]
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=tools,
        tool_choice="auto"
    )
    msg = response.choices[0].message
    if msg.tool_calls:
        # 执行所有工具调用
        for tc in msg.tool_calls:
            result = execute_tool(tc)
            messages.append(msg)  # 助手原始响应
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": result
            })
        # 再次请求模型，生成最终回答
        final = client.chat.completions.create(
            model="gpt-4o",
            messages=messages
        )
        return final.choices[0].message.content
    else:
        return msg.content

注意：生产中需实现工具超时重试、幂等性、权限校验等企业级特性。

五、多Agent协作：从单一大模型到专家小组

复杂任务（如“处理客户投诉并分析根因，同时生成整改工单”）适合拆解为多个Agent协同完成。我们基于LangGraph构建有状态的多Agent工作流。

5.1 定义Agent节点与状态图

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # 累积消息
    customer_id: str
    order_id: str
    analysis_result: dict
    final_response: str

# 三个Agent：客服专员、质检员、工单生成器
def customer_service_agent(state: AgentState):
    # 调用大模型+工具处理基础诉求
    reply = chat_with_tools(state["messages"][-1]["content"], state["messages"])
    return {"messages": [{"role": "assistant", "content": reply}]}

def quality_analyst_agent(state: AgentState):
    # 分析对话历史，提取问题等级和根因
    history = "\n".join([m["content"] for m in state["messages"] if m["role"] in ("user", "assistant")])
    prompt = f"分析以下客服对话，输出JSON：{{'severity':'高/中/低', 'root_cause':'简述', 'suggestion':'改进建议'}}\n对话：{history}"
    resp = client.chat.completions.create(model="gpt-4o", messages=[{"role": "user", "content": prompt}], response_format={"type": "json_object"})
    return {"analysis_result": json.loads(resp.choices[0].message.content)}

def ticket_generator_agent(state: AgentState):
    # 生成工单（模拟）
    ticket = {
        "title": f"投诉工单 - 订单{state.get('order_id')}",
        "detail": state["analysis_result"],
        "assignee": "售后主管"
    }
    return {"final_response": f"工单已生成：{json.dumps(ticket, ensure_ascii=False)}"}

# 构建状态图
workflow = StateGraph(AgentState)
workflow.add_node("service", customer_service_agent)
workflow.add_node("analyze", quality_analyst_agent)
workflow.add_node("ticket", ticket_generator_agent)

workflow.set_entry_point("service")
workflow.add_edge("service", "analyze")
workflow.add_edge("analyze", "ticket")
workflow.add_edge("ticket", END)

app = workflow.compile()

5.2 执行与流式输出

initial_state = {
    "messages": [{"role": "user", "content": "我的订单ORD-12345收到货有破损，要求赔偿"}],
    "customer_id": "C10086",
    "order_id": "ORD-12345"
}
for event in app.stream(initial_state):
    for node, output in event.items():
        print(f"节点 {node} 输出：{output}")

LangGraph支持条件边、检查点持久化，适合构建长期运行的客服流程。

六、性能优化与可观测性

6.1 缓存策略（语义缓存）

使用GPTCache对相似Query命中缓存，大幅降低延迟和费用：

from gptcache import Cache
from gptcache.manager import vectorbase, get_data_manager
from gptcache.similarity_evaluation.distance import SearchDistanceEvaluation

cache = Cache()
cache.init(
    data_manager=get_data_manager(data_path="./cache_data"),
    similarity_evaluation=SearchDistanceEvaluation()
)
# 在调用LLM前检查
cached_answer = cache.get(user_query)
if cached_answer is not None:
    return cached_answer
# 否则调用LLM，然后存储
answer = llm_call(...)
cache.put(user_query, answer)

6.2 全链路追踪（LangSmith / OpenTelemetry）

集成LangSmith记录每次检索、重排序、LLM调用的耗时与Token用量，便于调优阈值和Prompt。

from langsmith import traceable

@traceable(run_type="retriever")
def hybrid_recall(query): ...

@traceable(run_type="llm")
def chat_with_tools(...): ...

6.3 异步与批处理

对于高并发场景，将RAG检索和重排序改为异步（asyncio），使用Semaphore控制并发数，避免API限流。

七、部署与CI/CD实践

我们使用Docker + FastAPI封装服务，并通过GitHub Actions自动构建镜像推送到腾讯云容器镜像仓库，再通过Kubernetes滚动更新。

核心API端点：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class QueryRequest(BaseModel):
    user_id: str
    query: str
    history: list[str] = []

@app.post("/v1/chat")
async def chat_endpoint(req: QueryRequest):
    # 调用上述完整流程
    context = retrieve_context(req.query)
    final_answer = process_with_tools(req.query, req.history, context)
    return {"answer": final_answer, "trace_id": request.state.trace_id}

健康检查、限流（使用slowapi）、日志结构化等均为标准配置，此处不赘述。

八、总结与展望

本文从Prompt工程、RAG、Function Calling到多Agent协作，完整呈现了一个企业级大模型应用的开发脉络。核心经验：

Prompt即代码：模板化、版本化、测试化；
检索质量决定生成质量：多路召回+重排序是必须；
工具调用需防御性设计：异常处理、超时、回退；
多Agent适合复杂流程，但增加了状态管理复杂度，需权衡；
可观测性不是附加品，是生产环境的基础设施。

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

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

aiops

大模型应用开发实战：从Prompt工程到多Agent协作系统

大模型应用开发实战：从Prompt工程到多Agent协作系统

大模型应用开发实战：从Prompt工程到多Agent协作系统

一、引言：为什么需要系统化的大模型应用开发？

二、基础层：Prompt工程与结构化输出

2.1 动态Prompt模板与版本管理

2.2 结构化输出：约束解码与Pydantic校验

三、RAG：企业知识库的精准检索增强

3.1 多路召回 + 重排序（Rerank）架构

3.2 上下文压缩与Token优化

四、Function Calling：让模型主动调用外部工具

五、多Agent协作：从单一大模型到专家小组

5.1 定义Agent节点与状态图

5.2 执行与流式输出

六、性能优化与可观测性

6.1 缓存策略（语义缓存）

6.2 全链路追踪（LangSmith / OpenTelemetry）

6.3 异步与批处理

七、部署与CI/CD实践

八、总结与展望

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐