OpenAI Function Calling + RAG 构建 AI Agent

目标

• 掌握 Function Calling（工具调用）机制
• 构建基于 Embeddings 的 RAG 知识库问答
• 实现完整的 AI Agent 循环

完整代码

1. Function Calling —— 让 GPT 调用你的函数

import json
from openai import OpenAI

client = OpenAI()

# ============================================================
# 定义工具（函数）
# ============================================================
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的当前天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如 Beijing, Shanghai"
                    }
                },
                "required": ["city"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "search_database",
            "description": "在内部数据库中搜索订单信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {"type": "string"},
                    "customer_name": {"type": "string"},
                },
            },
        },
    },
]


# 模拟函数实现
def get_weather(city: str) -> dict:
    """模拟天气查询"""
    return {
        "city": city,
        "temperature": 22,
        "condition": "晴朗",
        "humidity": "45%",
    }


def search_database(order_id: str = None, customer_name: str = None) -> dict:
    """模拟数据库搜索"""
    return {
        "orders": [
            {"id": "ORD-001", "product": "MacBook Pro", "status": "已发货"},
        ]
    }


# 函数映射表
available_functions = {
    "get_weather": get_weather,
    "search_database": search_database,
}


def run_agent(user_query: str):
    """AI Agent 主循环"""
    messages = [{"role": "user", "content": user_query}]

    # 第一次调用：GPT 决定是否要调用工具
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        tools=tools,
        tool_choice="auto",
    )

    response_message = response.choices[0].message
    tool_calls = response_message.tool_calls

    # 如果 GPT 需要调用工具
    if tool_calls:
        messages.append(response_message)

        for tool_call in tool_calls:
            function_name = tool_call.function.name
            function_args = json.loads(tool_call.function.arguments)

            print(f"🔧 调用函数: {function_name}({function_args})")

            # 执行函数
            function_to_call = available_functions[function_name]
            function_response = function_to_call(**function_args)

            # 将函数结果添加到消息中
            messages.append({
                "tool_call_id": tool_call.id,
                "role": "tool",
                "name": function_name,
                "content": json.dumps(function_response, ensure_ascii=False),
            })

        # 第二次调用：GPT 根据函数结果生成最终回复
        final_response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
        )

        return final_response.choices[0].message.content

    return response_message.content


# 测试
print(run_agent("北京今天天气怎么样？"))
print(run_agent("帮我查一下订单 ORD-001 的状态"))
print(run_agent("你好，你是谁？"))  # 不需要工具调用

2. RAG（检索增强生成）

import numpy as np

# ============================================================
# 构建知识库
# ============================================================
documents = [
    "Vibe 平台是一个智能元器件管理系统，支持元器件搜索、文档生成和代码例程。",
    "用户可以使用自然语言搜索技术栈，如 React、Spring Boot、PyTorch等。",
    "平台支持自动生成 Markdown 格式的技术文档和代码示例。",
    "价格单位为分（cents），1元 = 100分。",
    "所有回复必须使用中文，元器件处理需要按工作流顺序执行。",
]

# 获取 Embeddings
def get_embeddings(texts: list[str]) -> list[list[float]]:
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=texts,
    )
    return [d.embedding for d in response.data]


doc_embeddings = get_embeddings(documents)
print(f"知识库大小: {len(documents)} 条文档, 向量维度: {len(doc_embeddings[0])}")

# ============================================================
# 语义搜索
# ============================================================
def semantic_search(query: str, top_k: int = 3) -> list[str]:
    """根据语义相似度检索最相关的文档"""
    query_embedding = get_embeddings([query])[0]

    # 计算余弦相似度
    def cosine_similarity(a, b):
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

    similarities = [
        cosine_similarity(query_embedding, doc_emb)
        for doc_emb in doc_embeddings
    ]

    # 取 top-k
    top_indices = np.argsort(similarities)[-top_k:][::-1]
    return [documents[i] for i in top_indices]


# ============================================================
# RAG 问答
# ============================================================
def rag_qa(query: str) -> str:
    """基于知识库的问答"""
    # 1. 检索相关文档
    relevant_docs = semantic_search(query, top_k=3)
    context = "\n\n".join(relevant_docs)

    print(f"📚 检索到 {len(relevant_docs)} 条相关文档")

    # 2. 用检索到的上下文增强 Prompt
    system_prompt = f"""你是一个知识库助手。请基于以下上下文回答问题。
如果上下文不足以回答，请说明。

上下文：
{context}"""

    # 3. 调用 GPT 生成答案
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": query},
        ],
        temperature=0.3,  # 降低随机性
    )

    return response.choices[0].message.content


# 测试
print("\n" + "="*50)
print(rag_qa("价格单位是什么？"))
print("\n" + rag_qa("如何搜索技术栈？"))

3. 完整的 Agent + RAG 架构

用户提问
    │
    ▼
┌─────────────────────┐
│   Router (GPT-4o)    │ → 判断：需要工具？需要检索？
└──────┬──────────────┘
       │
  ┌────┼────┐
  ▼    ▼    ▼
Tool  RAG  Direct
  │    │    │
  └────┼────┘
       ▼
   Final Answer

关键要点

概念	说明
tools 参数	向 GPT 注册可调用函数
tool_choice: "auto"	GPT 自行决定是否调用工具
role: "tool"	返回函数执行结果给 GPT
text-embedding-3-small	性价比最高的嵌入模型
余弦相似度	衡量向量间语义相似度的标准方法
RAG vs 微调	RAG 更新知识只需改文档，微调需重新训练