从提示词工程到Agent Skill：AI技术栈的演进路径与实战解析

1次阅读

共计 4619 个字符，预计需要花费 12 分钟才能阅读完成。

在 AI 应用开发中，提示词工程（Prompt Engineering）是构建智能体的基础，但它面临着诸多局限性。首先，上下文窗口的限制使得模型无法处理过长的输入或复杂的多轮对话。例如，GPT- 4 的上下文窗口虽然较大，但在处理超长文档或多轮对话时，依然会出现信息丢失的问题。

其次，多步推理的困难是另一个主要挑战。传统的提示词工程往往依赖于单次请求 - 响应模式，无法有效处理需要多步推理的任务。例如，一个复杂的数学问题可能需要分解为多个子问题，而传统的提示词工程难以实现这种分步处理。

此外，提示词工程的可扩展性和维护性也是问题。随着应用复杂度的增加，提示词的数量和复杂度会急剧上升，导致管理和调试变得困难。

基础 Prompt Engineering 是最初的阶段，主要通过设计高质量的提示词来引导模型生成期望的输出。以下是一个使用 OpenAI API 的示例：

import openai

def generate_response(prompt: str) -> str:
    """Generate a response using OpenAI's GPT-4 model.

    Args:
        prompt: The input prompt to guide the model's response.

    Returns:
        The generated response from the model.
    """
    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

Chain-of-Thought（CoT）通过引导模型分步思考，解决了多步推理的问题。以下是实现 CoT 的示例：

def chain_of_thought(question: str) -> str:
    """
    Implement Chain-of-Thought reasoning for a given question.

    Args:
        question: The question to be answered.

    Returns:
        The step-by-step reasoning and final answer.
    """prompt = f"""Question: {question}
    Let's think step by step:
    1. """
    response = generate_response(prompt)
    return response

Tool Augmentation 通过集成外部工具（如搜索引擎、数据库）来扩展模型的能力。以下是使用 LangChain 和 LlamaIndex 的对比示例：

LangChain 示例：

from langchain.agents import load_tools
from langchain.agents import initialize_agent
from langchain.llms import OpenAI

llm = OpenAI(temperature=0)
tools = load_tools(["serpapi", "llm-math"], llm=llm)
agent = initialize_agent(tools, llm, agent="zero-shot-react-description", verbose=True)
agent.run("What is the current price of Bitcoin in USD?")

LlamaIndex 示例：

from llama_index import GPTVectorStoreIndex, SimpleDirectoryReader

documents = SimpleDirectoryReader('data').load_data()
index = GPTVectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()
response = query_engine.query("What is the capital of France?")
print(response)

Agent Skill 抽象通过将功能模块化为可复用的技能，提升了系统的可维护性和扩展性。以下是一个技能类的设计示例：

from typing import Dict, Any

class Skill:
    """Base class for all Agent Skills."""

    def __init__(self, name: str, description: str):
        self.name = name
        self.description = description

    def execute(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
        """
        Execute the skill with the given input data.

        Args:
            input_data: A dictionary containing input parameters.

        Returns:
            A dictionary containing the output of the skill execution.
        """raise NotImplementedError("Subclasses must implement this method.")

class MathSkill(Skill):
    """A skill for performing mathematical calculations."""

    def __init__(self):
        super().__init__("math", "Performs basic mathematical calculations.")

    def execute(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
        expression = input_data.get("expression")
        try:
            result = eval(expression)
            return {"result": result}
        except Exception as e:
            return {"error": str(e)}

在对话系统中，确保每次请求的幂等性至关重要。可以通过唯一会话 ID 和状态机来实现：

from typing import Dict, Any

class ConversationState:
    """Manages the state of a conversation to ensure idempotency."""

    def __init__(self):
        self.states: Dict[str, Any] = {}

    def get_state(self, session_id: str) -> Any:
        """
        Retrieve the current state of a conversation.

        Args:
            session_id: The unique identifier for the conversation.

        Returns:
            The current state of the conversation.
        """
        return self.states.get(session_id, {})

    def update_state(self, session_id: str, new_state: Any) -> None:
        """
        Update the state of a conversation.

        Args:
            session_id: The unique identifier for the conversation.
            new_state: The new state to be stored.
        """
        self.states[session_id] = new_state

冷启动问题可以通过预加载常用技能和缓存结果来优化：

from typing import Dict, Any

class SkillCache:
    """Caches the results of skill executions to optimize cold starts."""

    def __init__(self):
        self.cache: Dict[str, Any] = {}

    def get(self, key: str) -> Any:
        """
        Retrieve a cached result.

        Args:
            key: The cache key.

        Returns:
            The cached result if it exists, otherwise None.
        """
        return self.cache.get(key)

    def set(self, key: str, value: Any) -> None:
        """
        Store a result in the cache.

        Args:
            key: The cache key.
            value: The result to be cached.
        """
        self.cache[key] = value

监控 API 调用成本可以通过记录 token 使用量和请求次数来实现：

from typing import Dict, Any

class CostMonitor:
    """Monitors the cost of API calls by tracking token usage and request counts."""

    def __init__(self):
        self.total_tokens: int = 0
        self.total_requests: int = 0

    def record_usage(self, tokens: int) -> None:
        """
        Record the token usage for a single API call.

        Args:
            tokens: The number of tokens used in the API call.
        """
        self.total_tokens += tokens
        self.total_requests += 1

    def get_metrics(self) -> Dict[str, int]:
        """
        Retrieve the current cost metrics.

        Returns:
            A dictionary containing the total tokens and requests.
        """return {"total_tokens": self.total_tokens,"total_requests": self.total_requests}