大模型开发实战:从Prompt工程到AI Agent落地的完整技术路径

大模型开发为何需要系统工程化方法

大模型开发早已不是调几个参数、写几条提示词就能交付的阶段。当GPT-4级别的模型成为基础设施,真正决定业务效果的不再是模型本身的能力天花板,而是Prompt工程、上下文管理、工具调用编排这些工程化环节的落地质量。本文从实际项目经验出发,拆解从Prompt设计到AI Agent完整落地的技术链路。

Prompt工程:结构化提示词的设计范式

Prompt工程不是”写一段描述”,而是用结构化方法把任务意图、约束条件、输出格式精确传递给模型。生产环境中最常用的是CRISPE框架:

CRISPE拆解:

  • Capacity & Role — 定义模型角色:你是一个有10年经验的Java后端架构师
  • Insight — 提供背景上下文:当前系统基于Spring Boot 3.2,日活50万,接口P99延迟要求200ms以内
  • Statement — 明确任务:分析以下接口的性能瓶颈并给出优化方案
  • Personality — 设定输出风格:用代码示例+解释的形式,避免泛泛而谈
  • Experiment — 约定输出格式:按”问题定位→优化措施→代码示例→效果预估”四段式输出

实际代码层面的Prompt模板管理:

from jinja2 import Template

PROMPT_TEMPLATE = """
你是一个{{role}},具备以下专业背景:
{{background}}

任务要求:
{{task}}

输出格式要求:
{{output_format}}

约束条件:
{{constraints}}
"""

def render_prompt(config: dict) -> str:
    template = Template(PROMPT_TEMPLATE)
    return template.render(**config)

# 使用示例
config = {
    "role": "数据库性能调优专家",
    "background": "熟悉MySQL 8.0 InnoDB引擎,擅长慢查询诊断与索引优化",
    "task": "分析以下SQL的执行计划,指出全表扫描风险并给出索引建议",
    "output_format": "1. 原始SQL\n2. EXPLAIN分析\n3. 问题诊断\n4. 优化SQL\n5. 预期效果",
    "constraints": "优化方案必须兼容MySQL 8.0,不允许修改表结构"
}
prompt = render_prompt(config)

上下文窗口管理:长文本场景的工程方案

128K上下文窗口不等于可以无脑塞内容。实际测试中,当输入token超过32K后,模型对中间位置信息的召回率会显著下降——这就是”Lost in the Middle”效应。工程实践中需要分层管理上下文:

class ContextManager:
    """分层上下文管理器"""

    def __init__(self, max_tokens: int = 32000):
        self.max_tokens = max_tokens
        self.system_prompt = ""
        self.few_shot_examples = []
        self.retrieved_chunks = []
        self.conversation_history = []

    def add_system_prompt(self, prompt: str):
        self.system_prompt = prompt

    def add_retrieved_chunks(self, chunks: list, max_chunks: int = 5):
        # RAG场景:只保留相关性最高的top-k
        self.retrieved_chunks = chunks[:max_chunks]

    def add_few_shot(self, examples: list, max_examples: int = 3):
        self.few_shot_examples = examples[:max_examples]

    def build_context(self) -> str:
        parts = [self.system_prompt]

        for ex in self.few_shot_examples:
            parts.append(f"示例输入:{ex['input']}")
            parts.append(f"示例输出:{ex['output']}")

        for chunk in self.retrieved_chunks:
            parts.append(f"参考资料:{chunk}")

        for turn in self.conversation_history[-6:]:
            parts.append(f"{turn['role']}: {turn['content']}")

        return "\n".join(parts)

    def estimate_tokens(self, text: str) -> int:
        # 粗估:中文约1.5 token/字,英文约1.3 token/word
        return int(len(text) * 1.5)

关键原则:system prompt放最前,few-shot examples紧随其后,检索到的RAG内容放在中间靠前位置,对话历史放在末尾——利用模型对首尾信息召回率最高的特性。

Function Calling与工具调用编排

大模型落地最核心的工程环节是Function Calling。当模型需要访问外部数据(数据库查询、API调用、文件读取)时,通过结构化的工具调用协议实现模型与外部系统的交互:

tools = [
    {
        "type": "function",
        "function": {
            "name": "query_mysql",
            "description": "执行MySQL查询语句并返回结果",
            "parameters": {
                "type": "object",
                "properties": {
                    "sql": {
                        "type": "string",
                        "description": "SELECT语句,仅允许查询不允许修改"
                    },
                    "database": {
                        "type": "string",
                        "description": "目标数据库名"
                    }
                },
                "required": ["sql", "database"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_docs",
            "description": "从内部文档库检索相关技术文档",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "检索关键词"
                    },
                    "top_k": {
                        "type": "integer",
                        "description": "返回文档数量,默认3",
                        "default": 3
                    }
                },
                "required": ["query"]
            }
        }
    }
]

工具调用的执行逻辑:

import openai

def run_agent(messages: list, tools: list, max_rounds: int = 5) -> str:
    """多轮工具调用执行器"""
    for _ in range(max_rounds):
        response = openai.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )

        msg = response.choices[0].message

        if msg.tool_calls:
            messages.append(msg)
            for tool_call in msg.tool_calls:
                # 执行工具函数
                result = execute_tool(
                    tool_call.function.name,
                    json.loads(tool_call.function.arguments)
                )
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps(result, ensure_ascii=False)
                })
        else:
            return msg.content

    return "达到最大调用轮数限制"


def execute_tool(name: str, args: dict) -> dict:
    """工具执行分发器"""
    tool_map = {
        "query_mysql": query_mysql,
        "search_docs": search_docs,
    }
    handler = tool_map.get(name)
    if not handler:
        return {"error": f"未知工具: {name}"}
    return handler(**args)

AI Agent架构:从单工具调用到多步推理

当业务场景需要多步推理和条件分支时,简单的Function Calling循环就不够了。需要引入Agent框架,实现规划-执行-反思的闭环:

class Agent:
    """轻量级ReAct Agent"""

    def __init__(self, llm, tools: list, max_iterations: int = 10):
        self.llm = llm
        self.tools = {t["function"]["name"]: t["function"] for t in tools}
        self.max_iterations = max_iterations
        self.trace = []

    def think(self, messages: list) -> dict:
        """思考阶段:决定下一步行动"""
        response = self.llm.chat(messages, tools=list(self.tools.values()))
        return response

    def act(self, tool_call: dict) -> dict:
        """执行阶段:调用具体工具"""
        name = tool_call["name"]
        args = tool_call["arguments"]
        return execute_tool(name, args)

    def observe(self, tool_call: dict, result: dict):
        """观察阶段:记录执行结果"""
        self.trace.append({
            "action": tool_call["name"],
            "args": tool_call["arguments"],
            "result": result
        })

    def run(self, task: str) -> str:
        messages = [{"role": "user", "content": task}]
        for i in range(self.max_iterations):
            response = self.think(messages)
            if not response.get("tool_calls"):
                return response["content"]
            for tc in response["tool_calls"]:
                result = self.act(tc)
                self.observe(tc, result)
                messages.append({
                    "role": "tool",
                    "content": json.dumps(result, ensure_ascii=False)
                })
        return "Agent达到最大迭代次数"

模型部署:从API调用到本地推理优化

大模型开发最终要落地到部署环节。当业务对延迟和成本有要求时,需要从API调用切换到本地推理:

vLLM部署方案核心配置:

# vLLM启动命令
python -m vllm.entrypoints.openai.api_server \
    --model /models/Qwen2.5-72B-Instruct \
    --tensor-parallel-size 4 \
    --gpu-memory-utilization 0.9 \
    --max-model-len 16384 \
    --quantization awq \
    --port 8000

# 对应的Python客户端
from openai import OpenAI

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

response = client.chat.completions.create(
    model="/models/Qwen2.5-72B-Instruct",
    messages=[{"role": "user", "content": "分析以下微服务日志中的异常模式"}],
    temperature=0.1,
    max_tokens=2048
)

量化部署的性能对比(A100 80G x 4):

| 模型规格 | 量化方式 | 显存占用 | 首token延迟 | 吞吐量 |
|———|———|———|———–|——-|
| 72B | FP16 | 320GB | 180ms | 12 tok/s |
| 72B | AWQ 4bit | 80GB | 95ms | 45 tok/s |
| 72B | GPTQ 4bit | 82GB | 102ms | 42 tok/s |

AWQ量化在保持模型效果损失低于2%的前提下,将显存需求降低75%,吞吐量提升3.7倍。对于大多数业务场景,这个trade-off完全可以接受。

常见问题诊断与排查

Q: 模型输出格式不稳定怎么办?

在system prompt中用JSON Schema约束输出,并在few-shot中给出2-3个标准格式示例。如果仍不稳定,考虑用JSON Mode(response_format={“type”: “json_object”})强制JSON输出,或者后处理阶段用正则提取结构化内容。

Q: Function Calling出现幻觉调用怎么办?

模型有时会调用不存在的工具名或传递错误参数。解决方案:在工具执行器中做参数校验,对不存在的工具名返回明确错误信息,让模型自行纠正。同时限制tool_choice为必选工具列表,避免模型随意编造。

Q: 多轮对话中上下文丢失如何处理?

实现滑动窗口+摘要机制:当对话轮数超过阈值时,对早期对话做摘要压缩,只保留最近N轮完整上下文。摘要本身也可以用同一模型生成,形成递归压缩。

原创文章,作者:小编,如若转载,请注明出处:https://www.yunthe.com/da-mo-xing-kai-fa-shi-zhan-cong-prompt-gong-cheng-dao/

(0)
小编小编
上一篇 14小时前
下一篇 14小时前

相关推荐