吃透AI Agent核心：RAG+Function Calling实战全解

FC_PROMPT = """
你是严格的工具调用机器人，只遵守以下规则：
1. 只输出纯JSON，禁止任何文字、解释、标点、备注、代码块。
2. 不需要调用工具：{"need_call": false}
3. 需要调用工具：{"need_call": true, "tool_name": "工具名", "parameters": {参数}}
4. 只能从提供的工具列表选择，禁止编造工具和参数。
5. 必须补全所有必填参数，禁止参数缺失。

工具列表：
{tools}

用户问题：{query}
"""

（2）万能容错FC执行模板（永不崩溃、全场景适配）

解决模型输出脏数据、JSON解析失败、参数报错、工具执行异常等问题，自带多重容错机制，是Agent线上稳定运行的核心保障：

def safe_fc_call(llm, query, tools_info, tools_map):
    """
    工业级万能FC调用函数
    全容错、防解析失败、防输出乱格式、防工具报错
    """
    # 注入强约束提示词
    prompt = FC_PROMPT.format(tools=json.dumps(tools_info, ensure_ascii=False), query=query)
    resp = llm.invoke(prompt)
    content = resp.content.strip()
    
    # 清洗模型不规范输出（去除代码块、多余注释）
    content = content.replace("```json", "").replace("```", "").strip()
    
    # 第一层容错：JSON解析失败兜底
    try:
        fc_result = json.loads(content)
    except Exception:
        return {"need_call": False, "error": "模型输出格式异常，解析失败"}
    
    # 无需调用工具，直接返回
    if not fc_result.get("need_call"):
        return {"need_call": False}
    
    # 第二层容错：工具调用、参数匹配异常兜底
    try:
        tool_name = fc_result["tool_name"]
        params = fc_result["parameters"]
        tool_res = tools_map[tool_name](**params)
        return {
            "need_call": True,
            "tool_name": tool_name,
            "tool_result": tool_res
        }
    except Exception as e:
        return {"need_call": False, "error": f"工具执行异常：{str(e)}"}

（3）标准化工具描述模板（统一规范，降低模型理解偏差）

统一工具定义格式，让LLM精准理解每个工具的用途、参数要求，从根源减少错调、漏调、参数缺失问题：

# 工具信息描述（给LLM决策使用）
tools_info = [
    {
        "name": "rag_search",
        "desc": "检索公司规则、退款政策、客服流程等私有知识库静态知识",
        "params": [{"name": "query", "desc": "用户的核心问题", "required": True}]
    },
    {
        "name": "get_order_status",
        "desc": "查询用户订单实时物流状态，获取动态业务数据",
        "params": [{"name": "order_id", "desc": "用户提供的订单编号", "required": True}]
    }
]

# 工具映射（代码实际执行使用）
tools_map = {
    "rag_search": rag_search,
    "get_order_status": get_order_status
}

四、AI Agent开发的核心重难点复盘

很多人以为RAG是Agent最难的部分，实则不然。结合实战踩坑经验，核心难度优先级排序：

Function Calling稳定性 > 提示词工程 > RAG检索精度优化 > 对话上下文管理

1. FC是Agent的生命线（最大难点）

常见致命问题：模型不调用工具、错误调用工具、参数传递错误、输出格式混乱（非JSON）、解析报错崩溃。绝大多数线上Agent故障，都源于FC不稳定，而非RAG精度问题。

2. 提示词是稳定性天花板

随意编写的提示词，会导致模型输出不可控、随意编造内容、不遵守工具调用规则。工业级Agent必须用强约束标准化提示词，强制模型固定格式输出。

3. RAG是体验优化项

RAG优化只会影响回答精准度，不会导致系统崩溃；但FC出错，会直接让整个Agent瘫痪。

五、实战踩坑总结（行业真实痛点）

FC调用失效：模型不触发RAG检索、不调用业务接口，直接编造答案
格式解析报错：模型输出多余文字、代码块标记，导致JSON解析失败
知识库污染：跳过文档清洗，噪音数据导致检索无效
参数传递错误：FC参数描述不清晰，模型传参缺失、错误
模型幻觉失控：无强约束提示词，无知识库内容时随意编造答案

六、工业级可落地代码：RAG+万能FC+完整Agent

以下代码整合了文档清洗、标准RAG流水线、永不翻车的FC模板、全异常容错机制，可直接运行，适配学习和小型项目落地。

依赖安装

pip install langchain langchain-openai chromadb python-dotenv re

完整可运行源码

流程解析：

提前离线备好知识库（清洗→切片→向量化→入库）【只跑一次】

用户说话

FC 先判断（大脑指挥）

调用对应工具（RAG / 订单工具）

拿到素材

LLM 结合提示词整理答案输出

import json
import re
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain.schema import Document

# ====================== 全局配置 ======================
API_KEY = "你的OpenAI Key"
BASE_URL = "https://api.chatanywhere.tech/v1"  # 国内代理地址

# ====================== 1. 初始化LLM大脑（低温度保证稳定） ======================
llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0,  # 0=杜绝随机编造，保证决策稳定
    api_key=API_KEY,
    base_url=BASE_URL
)

# ====================== 2. 工程级文档清洗工具 ======================
def clean_text(text):
    """万能文本清洗：去除页码、网址、乱码、空行、冗余符号"""
    text = re.sub(r'第\d+页', '', text)
    text = re.sub(r'www\..+?com', '', text)
    text = re.sub(r'\n+', '\n', text)
    text = re.sub(r' +', ' ', text)
    text = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9\s\.,;:!?，。；：！？]', '', text)
    return text.strip()

# ====================== 3. 标准化RAG知识库构建 ======================
# 原始业务文档（带噪音数据，模拟真实场景）
raw_docs = [
    "公司退款规则：收货7天内可无理由退款，超过7天需要质量问题才能退。第1页",
    "客服工作流程：先问候，再查订单，最后解答问题。www.abc.com官方客服规范"
]

# 1.文档清洗
cleaned_docs = [clean_text(doc) for doc in raw_docs]

# 2.文本切片
text_splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=50)
documents = [Document(page_content=d) for d in cleaned_docs]
splits = text_splitter.split_documents(documents)

# 3.向量化+入库
embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")
vector_db = Chroma.from_documents(documents=splits, embedding=embeddings)
rag_retriever = vector_db.as_retriever(search_kwargs={"k": 1})

# RAG检索工具（纳入FC工具池）
def rag_search(query: str) -> str:
    """检索公司规则、退款政策、客服流程等私有知识库"""
    docs = rag_retriever.invoke(query)
    return docs[0].page_content if docs else "无相关信息"

# ====================== 4. 自定义业务工具 ======================
def get_order_status(order_id: str) -> str:
    """查询订单物流状态，获取实时业务数据"""
    return f"订单 {order_id} 状态：已发货，预计明天送达"

# ====================== 5. 工业级万能FC强约束提示词 ======================
FC_PROMPT = """
你是严格的工具调用机器人，只遵守以下规则：
1. 只输出纯JSON，禁止任何文字、解释、标点、备注、代码块。
2. 不需要调用工具：{"need_call": false}
3. 需要调用工具：{"need_call": true, "tool_name": "工具名", "parameters": {参数}}
4. 只能从提供的工具列表选择，禁止编造工具和参数。
5. 必须补全所有必填参数，禁止参数缺失。

工具列表：
{tools}

用户问题：{query}
"""

# 工具描述池（给LLM决策参考）
tools_info = [
    {
        "name": "rag_search",
        "desc": "检索公司规则、退款政策、客服流程等私有知识库",
        "params": [{"name": "query", "desc": "用户的核心问题", "required": True}]
    },
    {
        "name": "get_order_status",
        "desc": "查询用户订单实时物流状态",
        "params": [{"name": "order_id", "desc": "用户提供的订单编号", "required": True}]
    }
]

# 工具执行映射（代码调用）
tools_map = {
    "rag_search": rag_search,
    "get_order_status": get_order_status
}

# ====================== 6. 全容错FC核心函数 ======================
def safe_fc_call(query):
    """永不崩溃的工具调用，兼容模型所有不规范输出"""
    prompt = FC_PROMPT.format(tools=json.dumps(tools_info, ensure_ascii=False), query=query)
    resp = llm.invoke(prompt)
    content = resp.content.strip()
    
    # 清洗模型多余输出（代码块、注释等）
    content = content.replace("```json", "").replace("```", "").strip()
    
    # JSON解析容错
    try:
        fc_result = json.loads(content)
    except:
        return {"need_call": False, "error": "工具调用解析失败"}
    
    # 无需调用工具
    if not fc_result.get("need_call"):
        return {"need_call": False}
    
    # 执行工具容错
    try:
        tool_name = fc_result["tool_name"]
        params = fc_result["parameters"]
        tool_res = tools_map[tool_name](**params)
        return {
            "need_call": True,
            "tool_name": tool_name,
            "tool_result": tool_res
        }
    except Exception as e:
        return {"need_call": False, "error": f"工具执行异常：{str(e)}"}

# ====================== 7. 完整Agent对话逻辑 ======================
def agent_chat(query):
    print(f"\n===== 用户问题：{query} =====")
    # 工具决策与执行
    fc_result = safe_fc_call(query)
    
    # 无工具需求，直接回答
    if not fc_result["need_call"]:
        answer = llm.invoke(query).content
        print(f"🤖 AI直接回答：{answer}")
        return
    
    # 有工具需求，整合数据生成最终答案
    tool_name = fc_result["tool_name"]
    tool_data = fc_result["tool_result"]
    print(f"✅ 成功调用工具：{tool_name}")
    print(f"📦 工具返回数据：{tool_data}")
    
    final_prompt = f"严格根据已知资料回答用户问题，禁止编造内容。用户问题：{query}\n参考资料：{tool_data}"
    final_answer = llm.invoke(final_prompt).content
    print(f"🎉 AI最终回答：{final_answer}")

# ====================== 8. 功能测试 ======================
if __name__ == "__main__":
    agent_chat("超过7天能退款吗？")    # 触发RAG检索
    agent_chat("我的订单12345状态如何？") # 触发业务FC
    agent_chat("你好呀")               # 直接对话回答