面试必杀：详细描述从用户提问，到向量召回、Prompt 注入、LLM 推理、再到工具调用的全链路数据流转

在当今人工智能领域，大型语言模型（LLM）的应用已从简单的问答系统发展到能够理解复杂意图、调用外部工具解决实际问题的智能代理。这一演进背后，是一整套精妙的数据流转机制在支撑。本次讲座将深入剖析从用户提出问题，到系统进行向量召回，构建Prompt，LLM进行推理，直至最终调用外部工具的全链路数据流转过程，旨在为各位编程专家提供一个清晰、严谨的技术视角。

第一章：用户意图的捕获与初步理解

一切智能交互的起点，都源于用户的问题。用户通过各种界面，如Web应用、移动App、聊天机器人或API接口，输入他们的需求。这个阶段的核心任务是准确地捕获用户的原始意图。

1.1 用户输入的接收与预处理

当用户在前端界面输入文本并点击发送时，该文本会通过网络请求（通常是HTTP/HTTPS协议）发送到后端服务。后端服务接收到请求后，会进行一系列初步处理：

• 数据清洗：移除多余的空格、特殊字符，统一大小写（部分场景），处理HTML实体等。
• 语言检测：识别用户输入所使用的语言，以便后续选择合适的语言模型或语言相关的处理模块。
• 内容审核：初步过滤敏感词、违禁内容，确保系统不会处理不当信息。
• 限长检查：确保用户输入长度在系统可接受的范围内，防止恶意攻击或资源耗尽。

例如，一个Web应用的用户输入可能通过以下方式到达后端：

# 假设这是一个Flask或FastAPI的后端接口
from flask import Flask, request, jsonify
import re

app = Flask(__name__)

@app.route('/ask', methods=['POST'])
def handle_user_query():
    user_input_raw = request.json.get('query')

    if not user_input_raw:
        return jsonify({"error": "Query cannot be empty"}), 400

    # 1.1.1 数据清洗示例：移除多余空格，并进行初步的内容过滤
    user_query_cleaned = re.sub(r's+', ' ', user_input_raw).strip()

    # 1.1.2 简单敏感词过滤
    sensitive_words = ["敏感词1", "敏感词2"]
    for word in sensitive_words:
        if word in user_query_cleaned:
            return jsonify({"error": "Content contains sensitive words"}), 403

    # 1.1.3 长度检查
    if len(user_query_cleaned) > 512: # 限制输入最大长度
        return jsonify({"error": "Query too long"}), 413

    print(f"Received and cleaned query: '{user_query_cleaned}'")

    # 后续流程将从这里开始，将 user_query_cleaned 传递给向量召回模块
    # ...
    return jsonify({"message": "Query received, processing..."})

if __name__ == '__main__':
    app.run(debug=True)

用户输入经过初步处理后，形成一个相对干净、规范化的文本，这为后续的语义理解奠定了基础。

第二章：语义化表示与向量召回

用户的问题，无论多么简洁或复杂，都蕴含着特定的语义信息。为了让机器理解这些语义并找到相关信息，我们需要将文本转换为机器可计算的数值表示，即向量（Embeddings），并利用这些向量进行高效的信息召回。

2.1 文本嵌入 (Text Embedding)

文本嵌入是将离散的文本信息映射到连续的、低维的向量空间中的过程。在这个向量空间里，语义相似的文本会拥有相近的向量表示，而语义不相关的文本则距离较远。

2.1.1 嵌入模型的选择

有多种模型可以生成高质量的文本嵌入：

• BERT系列模型 (如bert-base-nli-mean-tokens, all-MiniLM-L6-v2)：这些模型通过预训练任务（如掩码语言模型、下一句预测）学习文本表示，通过平均Token Embeddings或使用[CLS] Token的输出获得句向量。
• OpenAI Embeddings API (如text-embedding-ada-002, text-embedding-3-small, text-embedding-3-large)：这些是专为生成高质量嵌入而优化的专有模型，通常在各种基准测试中表现出色。
• Sentence-BERT (SBERT) 系列模型：SBERT通过孪生网络结构进行训练，旨在生成更具语义区分度的句向量，特别适合语义相似度搜索。
• 其他专有模型：如Cohere、Google等提供的Embedding API。

2.1.2 生成文本嵌入的代码示例

我们将使用Hugging Face transformers库中的Sentence Transformers模型来生成嵌入，因为它在本地部署和离线场景下非常方便。同时，也会展示如何调用OpenAI API。

from sentence_transformers import SentenceTransformer
import openai
import os

# 配置OpenAI API Key
# openai.api_key = os.getenv("OPENAI_API_KEY") # 生产环境建议从环境变量获取

def get_sentence_transformer_embedding(text: str, model_name: str = 'all-MiniLM-L6-v2'):
    """
    使用Sentence Transformers模型生成文本嵌入。
    """
    try:
        model = SentenceTransformer(model_name)
        embedding = model.encode(text)
        return embedding.tolist() # 返回列表形式的向量
    except Exception as e:
        print(f"Error generating embedding with Sentence Transformer: {e}")
        return None

def get_openai_embedding(text: str, model_name: str = 'text-embedding-3-small'):
    """
    使用OpenAI API生成文本嵌入。
    """
    try:
        # 确保OPENAI_API_KEY已设置
        if not openai.api_key:
            raise ValueError("OPENAI_API_KEY environment variable not set.")

        response = openai.embeddings.create(
            input=[text],
            model=model_name
        )
        return response.data[0].embedding
    except Exception as e:
        print(f"Error generating embedding with OpenAI: {e}")
        return None

# 示例
user_query = "什么是向量数据库？"
document_chunk_1 = "向量数据库是一种专门用于存储、管理和高效检索高维向量数据的数据库。"
document_chunk_2 = "关系型数据库是基于关系模型构建的数据库。"

# 使用Sentence Transformer
query_embedding_st = get_sentence_transformer_embedding(user_query)
doc1_embedding_st = get_sentence_transformer_embedding(document_chunk_1)
doc2_embedding_st = get_sentence_transformer_embedding(document_chunk_2)

if query_embedding_st and doc1_embedding_st and doc2_embedding_st:
    print(f"Sentence Transformer Query Embedding dimension: {len(query_embedding_st)}")
    # print(f"Query Embedding (ST): {query_embedding_st[:5]}...") # 打印前5个元素
    print(f"Doc1 Embedding (ST): {doc1_embedding_st[:5]}...")
    print(f"Doc2 Embedding (ST): {doc2_embedding_st[:5]}...")

# 使用OpenAI Embeddings (需要配置API Key)
# query_embedding_openai = get_openai_embedding(user_query)
# if query_embedding_openai:
#     print(f"nOpenAI Query Embedding dimension: {len(query_embedding_openai)}")
#     print(f"Query Embedding (OpenAI): {query_embedding_openai[:5]}...")

2.2 向量数据库 (Vector Databases)

生成文本嵌入后，我们需要一个地方来存储这些向量，并能够快速地根据查询向量找到最相似的文档向量。这就是向量数据库的作用。

2.2.1 作用与架构

• 存储：保存高维向量及其关联的元数据（如原始文本、文档ID、来源等）。
• 检索：根据查询向量，在海量向量数据中进行高效的相似度搜索（Approximate Nearest Neighbor, ANN），返回Top-K个最相似的结果。
• 索引结构：向量数据库通常采用特定的索引算法来加速ANN搜索，例如：
  • HNSW (Hierarchical Navigable Small World)：构建多层图结构，在搜索时从粗粒度到细粒度逐步逼近目标。
  • IVF_FLAT (Inverted File Index with Flat Quantization)：将向量空间划分为多个聚类，搜索时只在与查询向量最近的几个聚类中进行精确搜索。
  • DiskANN：优化磁盘I/O，适用于超大规模数据集。

2.2.2 常见向量数据库

• 专用向量数据库：Pinecone, Weaviate, Milvus, Qdrant, Vespa。
• 嵌入式/库：FAISS (Facebook AI Similarity Search), Annoy, Chroma。
• 支持向量的传统数据库：PostgreSQL (pgvector扩展), Redis (Redis Stack)。

2.2.3 向量存储与搜索的代码示例 (使用Chroma)

我们将使用Chroma，一个轻量级的开源向量数据库，支持本地运行和客户端-服务器模式。

import chromadb
from chromadb.utils import embedding_functions
import uuid

# 2.2.3.1 初始化Chroma客户端
# persistent_client = chromadb.PersistentClient(path="/path/to/my/chroma_db") # 持久化存储
client = chromadb.Client() # 内存存储，讲座演示方便

# 2.2.3.2 选择一个嵌入函数。Chroma可以集成多种embedding模型。
# 这里我们使用Chroma内置的Sentence Transformers。
# 注意：这会下载模型，如果前面已经用SentenceTransformer库下载过，这里会复用。
sentence_transformer_ef = embedding_functions.SentenceTransformerEmbeddingFunction(model_name="all-MiniLM-L6-v2")

# 2.2.3.3 创建或获取一个Collection
# Collection是Chroma中存储文档和向量的逻辑单元
collection_name = "my_knowledge_base"
try:
    collection = client.get_collection(name=collection_name, embedding_function=sentence_transformer_ef)
except:
    collection = client.create_collection(name=collection_name, embedding_function=sentence_transformer_ef)
    print(f"Created new collection: {collection_name}")

# 2.2.3.4 准备文档数据
documents_to_add = [
    {"id": str(uuid.uuid4()), "text": "向量数据库是一种专门用于存储、管理和高效检索高维向量数据的数据库。", "metadata": {"source": "wiki", "category": "database"}},
    {"id": str(uuid.uuid4()), "text": "HNSW是一种用于近似最近邻搜索的图结构索引算法，广泛应用于向量数据库。", "metadata": {"source": "research_paper", "category": "algorithm"}},
    {"id": str(uuid.uuid4()), "text": "大型语言模型（LLM）是基于Transformer架构的深度学习模型，能够理解和生成人类语言。", "metadata": {"source": "wiki", "category": "AI"}},
    {"id": str(uuid.uuid4()), "text": "Prompt工程是设计有效指令以引导LLM生成所需输出的技术。", "metadata": {"source": "blog", "category": "AI"}},
    {"id": str(uuid.uuid4()), "text": "关系型数据库如MySQL和PostgreSQL，使用表格存储数据，并通过SQL进行查询。", "metadata": {"source": "textbook", "category": "database"}},
]

# 2.2.3.5 将文档添加到Collection
# Chroma会自动为文本生成嵌入并存储
try:
    current_ids = collection.get(ids=[d["id"] for d in documents_to_add])['ids']
    new_docs = [d for d in documents_to_add if d["id"] not in current_ids]
    if new_docs:
        collection.add(
            documents=[d["text"] for d in new_docs],
            metadatas=[d["metadata"] for d in new_docs],
            ids=[d["id"] for d in new_docs]
        )
        print(f"Added {len(new_docs)} new documents to collection.")
    else:
        print("All documents already exist in collection.")
except Exception as e:
    print(f"Error adding documents: {e}")

# 2.2.3.6 执行向量召回 (相似度搜索)
user_query = "我想了解关于向量数据库的知识"
# Chroma的query方法会自动为查询文本生成嵌入
search_results = collection.query(
    query_texts=[user_query],
    n_results=2, # 返回最相似的2个结果
    # where={"category": "database"} # 可以根据元数据进行过滤
)

print("n--- 向量召回结果 ---")
for i, (doc, meta, dist) in enumerate(zip(
    search_results['documents'][0],
    search_results['metadatas'][0],
    search_results['distances'][0]
)):
    print(f"Result {i+1}:")
    print(f"  Document: {doc}")
    print(f"  Metadata: {meta}")
    print(f"  Distance: {dist:.4f}") # 距离越小越相似

2.3 召回策略 (Recall Strategies)

除了基本的Top-K相似度搜索，实际应用中还会结合多种策略来优化召回效果：

• 相似度度量：最常用的是余弦相似度，它衡量两个向量方向的相似性；此外还有欧氏距离、曼哈顿距离等。向量数据库通常会根据索引类型自动选择或优化距离计算。
• Top-K检索：直接返回与查询向量最相似的K个文档。
• 多路召回 (Hybrid Search)：结合关键词搜索（如BM25）和向量搜索。关键词搜索擅长精确匹配，而向量搜索擅长语义匹配，两者结合可以显著提升召回的全面性和准确性。
• 过滤与范围搜索：在向量搜索的同时，可以根据文档的元数据进行过滤，例如“只搜索2023年发布的文档”或“只搜索属于特定品类的商品”。

向量召回模块的目标是高效地从大规模知识库中抽取出与用户查询语义上最相关的少量（通常是几条到几十条）文档片段，作为LLM推理的上下文。

第三章：上下文构建与Prompt工程

从向量数据库召回的结果，虽然语义相关，但可能存在冗余、信息密度不均等问题。这一阶段的任务是将这些原始召回结果，经过精细处理和组织，注入到一个结构化的Prompt中，以引导LLM生成高质量的响应。

3.1 召回结果的筛选与排序

• 重排序 (Re-ranking)：召回阶段的相似度可能不够精确。可以使用一个更小的、但更复杂的交叉编码器（Cross-encoder）模型，对Top-K召回结果进行二次排序。交叉编码器能够同时考虑查询和召回文档的语义，给出更准确的相关性得分。
• 去重 (Deduplication)：召回的文档片段可能来自同一篇文档的不同部分，存在内容重复。需要识别并移除高度相似的片段。
• 长度限制与摘要：LLM的上下文窗口是有限的。如果召回结果过长，需要进行截断或摘要，以确保关键信息能够被包含在内，同时避免超出LLM的输入限制。
• 多样性提升：有时我们不希望返回的K个结果都高度相似，而是希望它们能覆盖用户意图的不同方面。可以采用最大边际相关性（Maximal Marginal Relevance, MMR）等算法来平衡相关性和多样性。

3.2 Prompt模板与结构

一个高质量的Prompt通常包含以下几个核心部分：

• 系统指令 (System Prompt)：定义LLM的角色、语气、行为约束以及期望的输出格式。这是最关键的部分，它为LLM设定了整个对话的基调和规则。
• 用户查询 (User Query)：用户原始的问题或请求。
• 召回上下文 (Retrieved Context)：从向量数据库中召回并经过处理的相关文档片段。这是RAG（Retrieval Augmented Generation）模式的核心。
• Few-shot examples (可选)：提供几个输入-输出示例，帮助LLM更好地理解任务并模仿期望的输出格式和风格。
• 工具定义 (Tool Definitions)：如果LLM需要调用外部工具，工具的详细描述（包括名称、用途、参数）会在此处注入。

3.3 Prompt注入 (Prompt Injection)

“Prompt注入”在此语境下指的是将动态获取的上下文信息（如召回文档、用户历史对话、工具定义等）以结构化的方式“注入”到预定义的Prompt模板中，形成最终发送给LLM的完整输入。

import textwrap

class PromptBuilder:
    def __init__(self, system_role: str = "你是一个专业的技术助手，请根据提供的上下文信息严谨地回答用户的问题。"):
        self.system_role = system_role
        self.context_template = "n以下是参考信息，请务必基于这些信息进行回答：n{context}n"
        self.tool_template = "n你可以使用以下工具来完成任务：n{tools_definition}n"
        self.user_query_template = "n用户问题：{query}n"
        self.final_instruction = "n请开始你的回答或行动计划。"

    def build_prompt(self, user_query: str, retrieved_docs: list[str], tool_definitions: str = None) -> str:
        """
        构建发送给LLM的完整Prompt。
        """
        prompt_parts = [f"<|system|>n{self.system_role}n"]

        # 注入召回的上下文
        if retrieved_docs:
            context_str = "n".join([f"文档片段 {i+1}:n{doc}" for i, doc in enumerate(retrieved_docs)])
            prompt_parts.append(self.context_template.format(context=context_str))

        # 注入工具定义
        if tool_definitions:
            prompt_parts.append(self.tool_template.format(tools_definition=tool_definitions))

        # 注入用户问题
        prompt_parts.append(self.user_query_template.format(query=user_query))
        prompt_parts.append(self.final_instruction)
        prompt_parts.append("<|end|>n") # 结束符，具体取决于模型

        # 合并所有部分
        full_prompt = "".join(prompt_parts)
        return full_prompt

# 模拟召回结果
retrieved_documents = [
    "向量数据库是一种专门用于存储、管理和高效检索高维向量数据的数据库。",
    "HNSW是向量数据库中常用的索引算法，它通过构建多层图来加速最近邻搜索。",
    "Chroma DB是一个轻量级的开源向量数据库，支持Python客户端，易于集成。",
]

# 模拟工具定义 (将在第五章详细说明)
# 这是一个简化的JSON Schema表示
tool_schema_example = """
{
  "tools": [
    {
      "type": "function",
      "function": {
        "name": "get_current_weather",
        "description": "获取指定城市当前的天气情况",
        "parameters": {
          "type": "object",
          "properties": {
            "location": {
              "type": "string",
              "description": "城市名称，例如 '北京', '上海'"
            },
            "unit": {
              "type": "string",
              "enum": ["摄氏度", "华氏度"],
              "description": "温度单位"
            }
          },
          "required": ["location"]
        }
      }
    }
  ]
}
"""

# 用户问题
user_question = "请解释一下什么是向量数据库，并告诉我HNSW算法在其中扮演什么角色。"

# 构建Prompt
prompt_builder = PromptBuilder()
final_prompt = prompt_builder.build_prompt(
    user_query=user_question,
    retrieved_docs=retrieved_documents,
    tool_definitions=tool_schema_example # 暂时注入工具定义
)

print("--- 最终注入LLM的Prompt ---")
# 使用textwrap.dedent和fill来美化输出，实际Prompt是一个长字符串
print(textwrap.fill(final_prompt, width=100))

Prompt注入的安全性考虑：

这里“Prompt注入”也有另一层含义，即恶意用户试图通过巧妙构造输入来操纵LLM行为，使其忽略系统指令或执行非预期操作。防范此类攻击是Prompt工程中的一个重要课题，通常通过以下方式：

• 明确的系统指令：在Prompt开头给出强力的、优先级最高的系统指令。
• 输入消毒：对用户输入进行严格的过滤和转义。
• 沙箱机制：限制LLM执行工具的权限。
• 模型微调：训练模型更专注于遵循系统指令。

第四章：大型语言模型推理 (Large Language Model Inference)

构建好完整的Prompt后，下一步就是将其发送给LLM进行推理，生成响应。这个阶段是整个数据流转的核心，LLM在这里展现其强大的语言理解和生成能力。

4.1 推理过程

LLM的推理过程可以概括为以下几个步骤：

1. Tokenization：将输入的Prompt文本分解成模型能够理解的最小单位——Token。Token可以是单词、词根、字符或子词单元。不同的模型有不同的Tokenizer。
2. 前向传播：Token序列被转换为嵌入向量，然后经过Transformer模型的多个层（Encoder-Decoder或Decoder-only），每个层包含自注意力机制和前馈网络。模型在此过程中捕捉Token之间的复杂关系和上下文信息。
3. 自回归生成：LLM是一种自回归模型，它一次生成一个Token。在生成第一个Token后，模型会将这个新生成的Token作为输入的一部分，与之前的Prompt一起，再次进行前向传播，预测下一个Token。这个过程重复进行，直到生成表示结束的特殊Token（如<|endoftext|>）或达到最大生成长度。
4. 采样策略：在每个生成步骤，模型会输出一个所有可能Token的概率分布。为了增加生成文本的多样性和创造性，通常会使用采样策略而不是简单地选择概率最高的Token：
   • Temperature (温度)：控制生成文本的随机性。温度越高，输出越随机、富有创造性；温度越低，输出越确定、保守。
   • Top-P (核采样)：选择概率累积和达到P的所有Token进行采样。
   • Top-K：只从概率最高的K个Token中进行采样。

4.2 LLM的选择与集成

• 模型API调用：最常见的集成方式是调用云服务提供商的LLM API，例如OpenAI的GPT系列（GPT-3.5, GPT-4）、Anthropic的Claude系列、Google的Gemini系列。这些API通常提供便捷的SDK和强大的模型能力。
• 本地部署模型：对于对数据隐私、成本或定制化有更高要求的场景，可以选择在本地或私有云部署开源LLM，如Llama 2、Mistral、Mixtral、Qwen等。这通常需要GPU资源和更复杂的部署管理。
• 异步与流式传输：为了提升用户体验，LLM的响应通常支持异步调用和流式传输（Streaming）。异步调用允许后端在等待LLM响应时处理其他任务；流式传输则允许前端逐字显示LLM的输出，减少用户等待时间。

4.2.1 调用OpenAI API进行推理的代码示例

这里我们将使用OpenAI的chat.completions.create方法，模拟一个基于Prompt进行推理的过程。

import openai
import os
import json

# 配置OpenAI API Key
# openai.api_key = os.getenv("OPENAI_API_KEY") # 确保已设置环境变量

def call_llm_inference(prompt: str, model: str = "gpt-3.5-turbo", temperature: float = 0.7, max_tokens: int = 500):
    """
    调用OpenAI API进行LLM推理。
    """
    try:
        if not openai.api_key:
            raise ValueError("OPENAI_API_KEY environment variable not set.")

        # OpenAI Chat Completion API期望messages格式
        messages = [
            {"role": "user", "content": prompt}
        ]

        response = openai.chat.completions.create(
            model=model,
            messages=messages,
            temperature=temperature,
            max_tokens=max_tokens,
            # stream=True # 可以设置为True进行流式输出
        )

        # 非流式输出
        llm_response_content = response.choices[0].message.content
        print("n--- LLM 原始推理结果 ---")
        print(llm_response_content)
        return llm_response_content

    except openai.APIError as e:
        print(f"OpenAI API error: {e}")
        return None
    except Exception as e:
        print(f"Error during LLM inference: {e}")
        return None

# 假设我们已经从上一阶段获得了 `final_prompt`
# llm_raw_output = call_llm_inference(final_prompt)
# print(f"nLLM Output: {llm_raw_output}")

# 示例：如果LLM被指示调用工具，它的输出会是结构化的JSON
# 假设LLM识别到需要调用工具，并生成了以下内容：
tool_call_output_example = """
Thought: 用户询问天气，我需要调用 `get_current_weather` 工具。
Action:
{
  "tool_calls": [
    {
      "id": "call_12345",
      "function": {
        "name": "get_current_weather",
        "arguments": {
          "location": "北京",
          "unit": "摄氏度"
        }
      },
      "type": "function"
    }
  ]
}
"""
print("n--- LLM 生成的工具调用示例 ---")
print(tool_call_output_example)

# 进一步解析LLM的输出，以确定是直接回答还是需要工具调用
def parse_llm_output(output: str):
    if "tool_calls" in output and "function" in output:
        # 这是一个简化的判断，实际需要更鲁棒的JSON解析
        try:
            # 假设工具调用部分是JSON格式
            # 真实场景中，OpenAI的Function Calling API会直接返回结构化的tool_calls
            # 这里我们模拟从字符串中提取
            action_start = output.find("Action:")
            if action_start != -1:
                json_str = output[action_start + len("Action:"):].strip()
                # 尝试解析JSON
                parsed_json = json.loads(json_str)
                if "tool_calls" in parsed_json and isinstance(parsed_json["tool_calls"], list):
                    return "tool_call", parsed_json["tool_calls"]
            return "text_response", output
        except json.JSONDecodeError:
            return "text_response", output # 解析失败，认为是文本响应
    return "text_response", output

# output_type, parsed_content = parse_llm_output(tool_call_output_example)
# print(f"nParsed LLM Output Type: {output_type}")
# if output_type == "tool_call":
#     print(f"Parsed Tool Calls: {parsed_content}")
# else:
#     print(f"Parsed Text Response: {parsed_content}")

在LLM推理阶段，模型不仅生成自然语言文本，当它被赋予工具调用的能力时，它还会根据用户意图和上下文，生成结构化的工具调用指令。这构成了从LLM到工具调用的桥梁。

第五章：工具调用与执行

当LLM识别到用户意图无法仅通过其内部知识库满足，而需要外部信息或操作时，它会生成一个工具调用指令。这个指令会被系统捕获并执行，然后将工具执行的结果反馈给LLM，形成一个思考-行动-观察-再思考的循环。

5.1 工具调用的原理

• 模型识别意图：LLM在推理过程中，通过分析用户问题、系统指令和提供的工具定义，判断是否需要调用某个工具。
• 生成结构化指令：如果需要调用工具，LLM会生成一个结构化的输出（通常是JSON格式），明确指定要调用的工具名称及其参数。这是LLM具备“函数调用”（Function Calling）能力的关键。
• JSON Schema for tool definition：为了让LLM理解工具的用途和参数，我们需要提供工具的JSON Schema定义。这个Schema详细描述了工具的名称、描述以及它接受的参数类型、描述和是否必需。

5.2 工具定义与注册

工具是预先定义好的函数或API接口，它们封装了具体的业务逻辑或外部服务调用。

import inspect

# 5.2.1 定义工具函数
def get_current_weather(location: str, unit: str = "摄氏度") -> dict:
    """
    获取指定城市当前的天气情况。
    :param location: 城市名称，例如 '北京', '上海'
    :param unit: 温度单位，可选 '摄氏度' 或 '华氏度'，默认为 '摄氏度'
    :return: 包含天气信息的字典
    """
    print(f"DEBUG: Calling get_current_weather for {location} in {unit}")
    # 实际场景中会调用外部天气API
    if location == "北京":
        return {"location": "北京", "temperature": "25", "unit": unit, "description": "晴朗"}
    elif location == "上海":
        return {"location": "上海", "temperature": "28", "unit": unit, "description": "多云"}
    else:
        return {"location": location, "temperature": "未知", "unit": unit, "description": "无法获取"}

def search_product_catalog(query: str, category: str = None, max_results: int = 5) -> list[dict]:
    """
    在产品目录中搜索商品。
    :param query: 搜索关键词
    :param category: 可选的产品类别
    :param max_results: 返回的最大结果数量
    :return: 匹配商品的列表
    """
    print(f"DEBUG: Calling search_product_catalog for '{query}' in category '{category}' with {max_results} results.")
    # 实际场景中会查询产品数据库
    if "手机" in query:
        return [
            {"name": "智能手机X", "price": 5999, "category": "电子产品"},
            {"name": "超薄手机Y", "price": 4500, "category": "电子产品"}
        ][:max_results]
    elif "笔记本" in query:
        return [
            {"name": "高性能笔记本A", "price": 8999, "category": "电子产品"},
            {"name": "轻薄笔记本B", "price": 6500, "category": "电子产品"}
        ][:max_results]
    return []

# 5.2.2 将工具函数转换为LLM可理解的JSON Schema格式
# OpenAI等API通常有自己的辅助函数来完成此操作，但理解其结构很重要
def get_tool_definition(func):
    """
    从Python函数生成OpenAI Function Calling兼容的JSON Schema定义。
    这是一个简化版本，实际可能需要更复杂的反射和类型映射。
    """
    signature = inspect.signature(func)
    params = signature.parameters
    properties = {}
    required_params = []

    for name, param in params.items():
        param_type = "string" # 默认类型
        description = ""
        # 尝试从docstring或类型注解获取描述和类型
        if param.annotation is not inspect.Parameter.empty:
            if param.annotation == str:
                param_type = "string"
            elif param.annotation == int:
                param_type = "integer"
            elif param.annotation == float:
                param_type = "number"
            elif param.annotation == bool:
                param_type = "boolean"
            # 更多类型映射...

        # 从函数的docstring中解析参数描述 (非常简化)
        if func.__doc__:
            for line in func.__doc__.split('n'):
                line = line.strip()
                if line.startswith(f":param {name}:"):
                    description = line.replace(f":param {name}:", "").strip()
                    break

        prop_info = {"type": param_type, "description": description}
        if param.default is not inspect.Parameter.empty:
            prop_info["default"] = param.default
        else:
            required_params.append(name)
        properties[name] = prop_info

    # 特殊处理 enum，例如天气单位
    if func.__name__ == "get_current_weather":
        if "unit" in properties:
            properties["unit"]["enum"] = ["摄氏度", "华氏度"]

    tool_schema = {
        "type": "function",
        "function": {
            "name": func.__name__,
            "description": func.__doc__.strip().split('n')[0] if func.__doc__ else "",
            "parameters": {
                "type": "object",
                "properties": properties,
                "required": required_params
            }
        }
    }
    return tool_schema

# 注册工具
available_tools = {
    "get_current_weather": get_current_weather,
    "search_product_catalog": search_product_catalog
}

tool_definitions_for_llm = [get_tool_definition(func) for func in available_tools.values()]

print("n--- LLM可理解的工具定义 (JSON Schema) ---")
print(json.dumps(tool_definitions_for_llm, indent=2, ensure_ascii=False))

# 将这些定义注入到Prompt中，或直接作为OpenAI ChatCompletion API的`tools`参数
# prompt_builder.build_prompt(..., tool_definitions=json.dumps({"tools": tool_definitions_for_llm}))

5.3 代理 (Agent) 与规划 (Planning)

当LLM需要执行多个工具调用，或者需要迭代地思考和行动时，我们就进入了“代理”（Agent）的范畴。代理通常采用ReAct（Reasoning and Acting）等框架，模拟人类的思维过程：

• 思考 (Thought)：LLM分析当前状态、用户请求和历史信息，决定下一步应该做什么。
• 行动 (Action)：LLM生成一个工具调用指令。
• 观察 (Observation)：系统执行工具，并将结果（观察）反馈给LLM。
• 再思考/再行动：LLM根据新的观察再次思考，决定是继续调用工具，还是生成最终答案。

这个循环会持续进行，直到LLM认为任务已完成并生成最终的自然语言响应。

5.4 工具执行与结果回传

这是一个核心的协调过程：

1. LLM生成工具调用：LLM的输出被解析，识别出它想要调用的工具和参数。
2. 代理执行工具：一个代理执行器（Agent Executor）根据LLM的指令，查找对应的工具函数，并用提供的参数调用该函数。
3. 结果捕获：工具函数的返回值被捕获。
4. 结果回传LLM：工具执行的结果（Observation）被格式化，作为新的上下文信息，再次注入到Prompt中，发送给LLM。

# 5.4.1 模拟代理执行器
class AgentExecutor:
    def __init__(self, tools_map: dict):
        self.tools = tools_map

    def execute_tool_call(self, tool_call_data: dict) -> dict:
        """
        根据LLM生成的工具调用数据执行工具。
        tool_call_data 应该包含 'name' 和 'arguments'。
        """
        tool_name = tool_call_data.get("function", {}).get("name")
        tool_args = tool_call_data.get("function", {}).get("arguments", {})

        if tool_name not in self.tools:
            return {"error": f"Tool '{tool_name}' not found."}

        func = self.tools[tool_name]
        try:
            # 确保参数类型匹配，这里进行简单的类型转换
            processed_args = {}
            sig = inspect.signature(func)
            for param_name, param_info in sig.parameters.items():
                if param_name in tool_args:
                    arg_value = tool_args[param_name]
                    if param_info.annotation is int:
                        processed_args[param_name] = int(arg_value)
                    elif param_info.annotation is float:
                        processed_args[param_name] = float(arg_value)
                    else:
                        processed_args[param_name] = arg_value
                elif param_info.default is not inspect.Parameter.empty:
                    processed_args[param_name] = param_info.default # 使用默认值
                elif param_info.kind == inspect.Parameter.VAR_POSITIONAL or 
                     param_info.kind == inspect.Parameter.VAR_KEYWORD:
                    pass # 忽略 *args, **kwargs
                else:
                    raise ValueError(f"Missing required argument '{param_name}' for tool '{tool_name}'.")

            result = func(**processed_args)
            return {"tool_output": result}
        except Exception as e:
            return {"error": f"Error executing tool '{tool_name}': {str(e)}"}

# 初始化代理执行器
agent_executor = AgentExecutor(available_tools)

# 模拟LLM生成的工具调用指令
llm_tool_call_instruction = {
    "id": "call_12345",
    "function": {
        "name": "get_current_weather",
        "arguments": {
            "location": "北京",
            "unit": "摄氏度"
        }
    },
    "type": "function"
}

# 5.4.2 执行工具
tool_execution_result = agent_executor.execute_tool_call(llm_tool_call_instruction)
print("n--- 工具执行结果 ---")
print(json.dumps(tool_execution_result, indent=2, ensure_ascii=False))

# 5.4.3 将工具执行结果回传给LLM (作为新的消息)
# 在OpenAI的ChatCompletion API中，这会作为role="tool"的消息
# messages.append({"role": "tool", "tool_call_id": "call_12345", "content": json.dumps(tool_execution_result['tool_output'])})
# 随后再次调用LLM进行推理

这一循环机制使得LLM能够突破其自身知识库的限制，与真实世界进行交互，从而解决更广泛、更复杂的任务。

第六章：全链路数据流转与优化

我们已经详细考察了从用户提问到工具调用的各个阶段。现在，我们将它们串联起来，并讨论整个系统的优化与安全性考量。

6.1 整个流程的串联

整个数据流转可以概括为以下步骤，通常由一个协调器（Orchestrator）或代理框架来管理：

1. 用户输入：用户通过UI提交请求。
2. 预处理：后端服务接收、清洗、标准化用户输入。
3. 语义化：将用户查询生成嵌入向量。
4. 向量召回：在向量数据库中进行相似度搜索，获取相关文档片段。
5. 上下文构建：对召回结果进行筛选、重排序，并结合系统指令、工具定义等，构建完整的Prompt。
6. LLM推理：将Prompt发送给LLM。
7. 结果解析：解析LLM的输出，判断是直接回答、生成工具调用，还是请求更多信息。
   • 如果是直接回答：将LLM的自然语言响应返回给用户。
   • 如果是工具调用：
     • 提取工具名称和参数。
     • 调用代理执行器执行相应工具。
     • 将工具执行结果作为新的上下文，再次构建Prompt并发送给LLM。
     • 重复步骤6和7，直到LLM生成最终答案。
8. 结果呈现：将最终的LLM响应（可能是直接回答，也可能是工具调用后生成的回答）呈现给用户。

这个过程形成了一个动态的、基于反馈的循环，使得系统能够进行多轮对话和复杂任务处理。

数据流转核心表：

阶段	输入数据	输出数据	关键处理模块/技术
用户提问	原始文本	清洗后的文本	Web/App前端, 后端API Gateway, 文本清洗、语言检测
向量召回	清洗后的文本 (用户查询)	查询嵌入向量, 召回的文档片段及元数据	嵌入模型 (Sentence-BERT, OpenAI Embeddings), 向量数据库 (Chroma, Pinecone)
Prompt注入	查询嵌入向量, 召回文档, 工具定义, 系统指令	完整、结构化的LLM Prompt字符串	Prompt工程, 重排序模型, RAG (Retrieval Augmented Generation)
LLM推理	完整Prompt字符串	LLM原始输出 (自然语言回答 或 工具调用指令)	LLM模型 (GPT-4, Claude, Llama 2), Tokenization, 自回归生成, 采样策略
工具调用	LLM原始输出 (工具调用指令)	工具执行结果 (结构化数据)	Agent Executor, 工具函数定义, JSON Schema解析
结果回传LLM	工具执行结果	带有工具结果的新Prompt (或Message)	Prompt工程, ReAct框架
最终结果呈现	LLM最终回答	格式化后的答案	前端UI, 后端API, 文本渲染

6.2 性能与成本优化

• 缓存机制：对常见的用户查询、Embedding结果、甚至LLM的中间推理结果进行缓存，减少重复计算和API调用。
• 模型选择与量化：根据任务需求选择合适大小和能力的LLM。对于本地部署模型，可以使用量化技术（如int8, int4）降低内存和计算需求。
• 并行处理：将不依赖顺序的任务并行化，例如同时进行多个文档的Embedding生成。
• 批处理 (Batching)：将多个用户的查询打包成一个批次发送给Embedding模型或LLM API，提高吞吐量。
• 异步I/O：在等待外部API（如LLM API、向量数据库）响应时，允许系统处理其他请求，提高并发能力。

6.3 错误处理与鲁棒性

• 超时与重试机制：对外部API调用设置超时，并实现合理的重试策略（指数退避）。
• LLM输出解析失败：当LLM输出的JSON格式不正确或无法解析时，需要有备用方案，例如回退到纯文本回答，或尝试修复Prompt再次推理。
• 工具执行失败：工具调用可能因为网络问题、参数错误或业务逻辑失败。需要捕获这些错误，并将其以友好的方式反馈给LLM，让LLM决定如何处理（例如，向用户道歉并询问更多信息，或尝试调用另一个工具）。
• 日志与监控：记录详细的日志，监控系统各模块的性能和错误率，以便及时发现和解决问题。

6.4 安全性考虑

• Prompt Injection防御：如前所述，通过加强系统指令、输入消毒、沙箱机制等来防范恶意Prompt。
• 工具执行的权限控制：确保LLM只能调用被授权的工具，并且工具只能访问被授权的数据和资源。例如，工具不能随意删除数据库记录，只能执行查询操作。
• 数据隐私：在整个数据流转过程中，确保用户敏感数据得到妥善处理和保护，遵循GDPR、CCPA等法规。例如，在发送给LLM之前，对敏感信息进行脱敏处理。
• 速率限制：对用户请求和LLM API调用进行速率限制，防止滥用和DoS攻击。

这个复杂而精妙的数据流转机制，是现代AI系统能够实现智能交互和自动化任务的核心所在。通过对每个环节的精细设计、优化和严格的安全控制，我们能够构建出强大、可靠且富有创新性的智能应用。

---

从用户意图的微妙捕捉，到语义向量的精准召回，再到智能Prompt的巧妙编织，以及LLM的深邃推理，直至外部工具的精准执行，每一步都承载着数据与智能的交织。这不仅是一条数据流转的链路，更是现代AI系统实现与真实世界交互、解决复杂问题的能力之源。理解并掌握这一全链路，是构建下一代智能应用的关键。