1、Function Calling产生的背景

1.1、原因

大语言模型本身局限性：
比如GPT4的知识库截至2024年4月，无法获取最新信息数据。
例如无法回答“今天的天气怎么样”这种简单问题，极大限制了其实际应用价值。
尽管大模型只能给出文字建议，无法直接解决问题。

1.2、本质

赋予大语言模型调用外部API的能力，也成为AIAgent的核心技能。
让大语言模型调用外部函数的能力，即模型可以不再仅仅根据自身的知识进行回答，而是可以额外挂载一个函数库。

1.3、调用步骤

1、大模型应用/AIAgent将用户提问和函数说明一起告知大模型，
2、大模型判断并回复具体调用函数和参数，
3、在按照实际需求调用外部函数并获取函数运行结果后，将结果告知大模型，
4、大模型再基于函数运行结果进行回答。

Funciton_Calling完整调用流程

2、Function Calling机制和全链路解析

2.1、Function Calling的实现过程

发起第一次模型调用应用程序向大模型发送用户问题和可用工具清单。
接收模型的工具调用指令（工具名称与入参）若模型判断需要调用外部工具，返回JSON格式的指令，指定函数名称与入参。若模型判断无需调用工具，会返回自然语言格式的回复。
在应用端运行工具应用程序执行指定工具，获取输出结果。
发起第二次模型调用将工具输出结果添加到消息数组（messages），再次调用模型。
接收来自模型的最终响应模型综合工具输出与用户问题，生成自然语言回复。

工作流程图（摘自阿里百炼）

阿里百炼function calling

具体代码实现

import pandas as pd
from openai import OpenAI
from dotenv import  load_dotenv

from models import get_normal_client, ALI_TONGYI_MAX_MODEL, ALI_TONGYI_PLUS_MODEL, ALI_TONGYI_DEEPSEEK_V3

client = get_normal_client()
import json
from io import StringIO

# 加载样例数据
df_complex = pd.DataFrame({
    'Name': ['Alice', 'Bob', 'Charlie'],
    'Age': [25, 30, 35],
    'Salary': [50000.0, 100000.5, 150000.75],
    'IsMarried': [True, False, True]
})
print(df_complex)

# 将样例数据中的DataFrame转换为JSON格式
df_complex_json = df_complex.to_json(orient='split')

# 编写函数功能
def calculate_total_age_from_split_json(input_json):
    """
    从给定的JSON格式字符串（按'split'方向排列）中解析出DataFrame，计算所有人的年龄总和，并以JSON格式返回结果。
    参数:
    input_json (str): 包含个体数据的JSON格式字符串。
    返回:
    str: 所有人的年龄总和，以JSON格式返回。
    """
    print("函数calculate_total_age_from_split_json被调用")
    # 将JSON字符串转换为DataFrame
    df = pd.read_json(StringIO(input_json), orient='split')
    # 计算所有人的年龄总和
    total_age = df['Age'].sum()
    # 将结果转换为字符串形式，然后使用json.dumps()转换为JSON格式
    return json.dumps({"total_age": str(total_age)})

#  测试函数功能  使用函数计算年龄总和，并以JSON格式输出
result = calculate_total_age_from_split_json(df_complex_json)
print("The JSON output is:", result)

#定义一个函数仓库
function_repository = {
    "calculate_total_age_from_split_json": calculate_total_age_from_split_json,
}

# 构建messages
messages = [
    {"role": "system","content": "你是一位优秀的数据分析师, 现在有这样一个数据集input_json：%s，数据集以JSON形式呈现" % df_complex_json},
    {"role": "user", "content": "请在数据集input_json上执行计算所有人年龄总和"}
]
response = client.chat.completions.create(
    model=ALI_TONGYI_MAX_MODEL,
    messages=messages,
    #把有哪些工具可以用，告诉我们的大模型
    tools = [{
            "type": "function",#固定写法
            "function": {
                "name": "calculate_total_age_from_split_json",
                "description": "计算年龄总和的函数，会从给定的JSON格式字符串中解析出DataFrame，计算所有人的年龄总和，并以JSON格式返回结果。",
                "parameters": {
                    "type": "object",#固定写法
                    "properties": {
                        "input_json": {
                            "type": "string",
                            "description": "包含待计算年龄总和的数据集",
                        },
                    },
                "required": ["input_json"],
                }
            }
        }],
    tool_choice="auto"
)
print('发送给大模型的消息: ',messages)
print('大模型的应答全文:  ',response)
print('大模型的应答:  ',response.choices[0].message)


function_name = response.choices[0].message.tool_calls[0].function.name
print('大模型认为应该执行的函数: ', function_name)

function_args = json.loads(response.choices[0].message.tool_calls[0].function.arguments)
print('大模型认为应该执行的函数所需要的参数:',function_args)

tool_calls_id = response.choices[0].message.tool_calls[0].id
print('应该执行的函数的ID: ',tool_calls_id)


local_fuction_call = function_repository[function_name]
print('本地函数调用: ',local_fuction_call)
function_response = local_fuction_call(**function_args)
print('本地函数调用结果:',function_response)

#报文的拼接，希望大模型能够以自然语言的形式返回结果给用户
messages.append(response.choices[0].message)
print('发送给大模型的消息拼接，加上 - 1、应答: ',messages)

#工具的调用结果
messages.append({
    "role": "tool", #固定写法
    "name": function_name,
    "tool_call_id": tool_calls_id,
    "content": function_response
    }
)

print('发送给大模型的消息拼接，加上 - 2、函数执行结果: ',messages)

# 再次向ChatCompletion模型提问
final_response = client.chat.completions.create(
    model=ALI_TONGYI_MAX_MODEL,
    messages=messages,
)

print('answer: ',final_response.choices[0].message.content)

2.2、动态SQL生成与数据库查询

要求：用自然语言描述查询数据库
1、初始化数据库

import pymysql
import json

from models import get_normal_client, ALI_TONGYI_MAX_MODEL, ALI_TONGYI_PLUS_MODEL

# 创建一个LLM客户端
client = get_normal_client()

#表的描述
database_schema_string = """
CREATE TABLE IF NOT EXISTS Classes (
    class_id INT PRIMARY KEY COMMENT '班级的ID编号',
    class_name VARCHAR(100) NOT NULL COMMENT '班级的名称'
) ENGINE=InnoDB COMMENT = '班级表';

CREATE TABLE IF NOT EXISTS Students (
    student_id INT PRIMARY KEY COMMENT '学生的唯一性ID编号',
    name VARCHAR(100) NOT NULL COMMENT '学生姓名',
    class_id INT COMMENT '学生所在班级的ID编号，和班级表中的班级ID编号对应'
) ENGINE=InnoDB COMMENT = '学生表';

CREATE TABLE IF NOT EXISTS Scores (
    score_id INT PRIMARY KEY COMMENT '学生成绩表的唯一性ID编号',
    student_id INT COMMENT '学生个人的ID编号，和学生的唯一性ID编号对应',
    subject VARCHAR(100) NOT NULL COMMENT '考试科目，中文名称标识',
    score FLOAT NOT NULL COMMENT '考试科目的分数'
) ENGINE=InnoDB COMMENT = '学生科目成绩表';
"""
# 程序运行前请运行 db_init.py，并确保数据库和表以及表中数据已存在
# 端口、用户名、密码、数据库IP地址请根据自己的实际情况进行修改
connection = pymysql.connect(
    host='127.0.0.1',
    port=3306,
    user='root',
    password='1234',
    database='ALLLM',
    charset='utf8mb4'  # 添加推荐的字符集参数
)

cursor = connection.cursor()

def get_sql_completion(messages, model=ALI_TONGYI_PLUS_MODEL):
    response = client.chat.completions.create(
        model=model,
        messages=messages,
        temperature=0,
        tools=[{
            "type": "function",
            "function": {
                "name": "ask_database",
                "description": "查询数据库的函数。输出是数据库中表的记录",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {
                            "type": "string",
                            "description": f"""
                            一个SQL查询，这个SQL查询提取信息以回答用户的问题。
                            SQL应该使用这个数据库架构来编写:
                            {database_schema_string}
                            SQL查询查询应以纯文本形式返回，而不是JSON格式。
                            查询应仅包含MySQL支持的语法.
                            """,
                        }
                    },
                    "required": ["query"],
                }
            }
        }],
    )
    return response.choices[0].message

#专门负责和数据库进行交互的工具
def ask_database(query):
    cursor.execute(query)
    records = cursor.fetchall()
    return records


prompt = "查询一班的学生数学成绩是多少？"
messages = [
    {"role": "system", "content": "基于表回答用户问题"},
    {"role": "user", "content": prompt}
]

#response的内容应该是什么？
response = get_sql_completion(messages)
messages.append(response)

if response.tool_calls is not None:
    tool_call = response.tool_calls[0]
    if tool_call.function.name == "ask_database":
        arguments = tool_call.function.arguments
        args = json.loads(arguments)
        print("====SQL====")
        print(args["query"])
        result = ask_database(args["query"])
        print("====MySQL数据库查询结果====")
        print(result)

        messages.append({
            "tool_call_id": tool_call.id,
            "role": "tool",
            "name": "ask_database",
            "content": str(result)
        })
        response = get_sql_completion(messages)
        print("====最终回复====")
        print(response.content)

2、利用大模型根据用户要求查询数据库

import pymysql
import json

from models import get_normal_client, ALI_TONGYI_MAX_MODEL, ALI_TONGYI_PLUS_MODEL

# 创建一个LLM客户端
client = get_normal_client()

#表的描述
database_schema_string = """
CREATE TABLE IF NOT EXISTS Classes (
    class_id INT PRIMARY KEY COMMENT '班级的ID编号',
    class_name VARCHAR(100) NOT NULL COMMENT '班级的名称'
) ENGINE=InnoDB COMMENT = '班级表';

CREATE TABLE IF NOT EXISTS Students (
    student_id INT PRIMARY KEY COMMENT '学生的唯一性ID编号',
    name VARCHAR(100) NOT NULL COMMENT '学生姓名',
    class_id INT COMMENT '学生所在班级的ID编号，和班级表中的班级ID编号对应'
) ENGINE=InnoDB COMMENT = '学生表';

CREATE TABLE IF NOT EXISTS Scores (
    score_id INT PRIMARY KEY COMMENT '学生成绩表的唯一性ID编号',
    student_id INT COMMENT '学生个人的ID编号，和学生的唯一性ID编号对应',
    subject VARCHAR(100) NOT NULL COMMENT '考试科目，中文名称标识',
    score FLOAT NOT NULL COMMENT '考试科目的分数'
) ENGINE=InnoDB COMMENT = '学生科目成绩表';
"""
# 程序运行前请运行 db_init.py，并确保数据库和表以及表中数据已存在
# 端口、用户名、密码、数据库IP地址请根据自己的实际情况进行修改
connection = pymysql.connect(
    host='127.0.0.1',
    port=3306,
    user='root',
    password='1234',
    database='ALLLM',
    charset='utf8mb4'  # 添加推荐的字符集参数
)

cursor = connection.cursor()

def get_sql_completion(messages, model=ALI_TONGYI_PLUS_MODEL):
    response = client.chat.completions.create(
        model=model,
        messages=messages,
        temperature=0,
        tools=[{
            "type": "function",
            "function": {
                "name": "ask_database",
                "description": "查询数据库的函数。输出是数据库中表的记录",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {
                            "type": "string",
                            "description": f"""
                            一个SQL查询，这个SQL查询提取信息以回答用户的问题。
                            SQL应该使用这个数据库架构来编写:
                            {database_schema_string}
                            SQL查询查询应以纯文本形式返回，而不是JSON格式。
                            查询应仅包含MySQL支持的语法.
                            """,
                        }
                    },
                    "required": ["query"],
                }
            }
        }],
    )
    return response.choices[0].message

#专门负责和数据库进行交互的工具
def ask_database(query):
    cursor.execute(query)
    records = cursor.fetchall()
    return records


prompt = "查询一班的学生数学成绩是多少？"
messages = [
    {"role": "system", "content": "基于表回答用户问题"},
    {"role": "user", "content": prompt}
]

#response的内容应该是什么？
response = get_sql_completion(messages)
messages.append(response)

if response.tool_calls is not None:
    tool_call = response.tool_calls[0]
    if tool_call.function.name == "ask_database":
        arguments = tool_call.function.arguments
        args = json.loads(arguments)
        print("====SQL====")
        print(args["query"])
        result = ask_database(args["query"])
        print("====MySQL数据库查询结果====")
        print(result)

        messages.append({
            "tool_call_id": tool_call.id,
            "role": "tool",
            "name": "ask_database",
            "content": str(result)
        })
        response = get_sql_completion(messages)
        print("====最终回复====")
        print(response.content)

3、AI Agent：核心认知框架

3.1、AI Agent实时决策：ReAct

一种用于增强大型语言模型的方法，它通过结合推理（Reasoning）和行动（Acting）来增强推理和决策的效果。

ReAct

ReAct框架的设计哲学是:在动态和不确定的环境中，有效的决策需要持续的学习和适应，以及快速将推理转化为行动的能力，即形成有效的观察一思考一行动一再观察的循环。

ReAct执行过程

v0.3版本代码是使用了ReAct模式的Agent

from dotenv import load_dotenv
from langchain_classic.agents import create_react_agent, AgentExecutor
from langchain_tavily import TavilySearch
from langchain_core.prompts import PromptTemplate
from langchain_core.tools import tool

from models import get_lc_o_ali_model_client, ALI_TONGYI_MAX_MODEL, ALI_TONGYI_TURBO_MODEL
import numexpr
import math

load_dotenv()
# 初始化大模型客户端，不同的模型表现不同
llm = get_lc_o_ali_model_client(model=ALI_TONGYI_MAX_MODEL)
@tool
def calculator(expression: str) -> str:
    """使用Python的numexpr库计算表达式。表达式应该是解决问题的单行数学表达式。.
    例子:
        "37593 * 67"
        "37593**(1/5)"
        "37593^(1/5)"
    """
    local_dict = {"pi": math.pi, "e": math.e}
    return str(
        numexpr.evaluate(
            expression.strip(),
            global_dict={},
            local_dict=local_dict,  # 添加常用数学函数
        )
    )
# 设置工具
search = TavilySearch(max_results=3)
tools = [calculator,search]

# 设置提示模板
"""提示词中三个变量是必须的：
tools：包含每个工具的描述和参数。
tool_names：包含所有工具名称。
agent_scratchpad：以字符串形式包含以前的代理操作和工具输出。
"""
template = '''
Answer the following questions as best you can. You can use the following tools:

{tools}

Use the following format:

Question: the input question you must answer
Thought: you should think about what to do
Action: the name of the tool to use, should be one of [{tool_names}]
Action Input: the input to the tool
Observation: the result of the tool
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: Now I know the final answer
Final Answer: the final answer to the original question

Begin!

Question: {input}
Thought: {agent_scratchpad}'''

"""以下是上面提示词的中文意思，注意！这里Question、Action等只能用英文提示词！
因为LangChain在处理时是根据英文单词来进行的，用中文的话会出错，
如果想使用中文提示词，必须自定义？"""
template = ('''
    '尽你所能回答以下问题。如果能力不够你可以使用以下工具:'
    '{tools}
    使用以下格式：'
    'Question: 你必须回答的输入问题'
    'Thought: 你应该始终思考该做什么'
    'Action: 要采取的行动，应该是 [{tool_names}]之一'
    'Action Input: 行动的输入'
    'Observation: 行动的结果'
    '... (这个想法/行动/行动输入/观察可以重复N次)'
    'Thought: 我现在知道最终答案了'
    'Final Answer: 原始输入问题的最终答案'
    '开始!'
    'Question: {input}'
    'Thought: {agent_scratchpad}'
    '''
)
prompt = PromptTemplate.from_template(template)

# 初始化Agent
#使用了React模式的Agent
agent = create_react_agent(llm, tools, prompt)

# 构建AgentExecutor
agent_executor = AgentExecutor(agent=agent,tools=tools,handle_parsing_errors=True,verbose=True)

# 执行AgentExecutor
agent_executor.invoke({"input": """在中国，目前市场上玫瑰花的一般进货价格是多少？大概是多少钱一支？\n如果我在此基础上加价5%，应该如何定价？"""})

V1.0的create_agent 函数实际上是构建在 LangGraph 引擎之上。

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_tavily import TavilySearch
from langchain_core.tools import tool

from models import get_lc_o_ali_model_client, ALI_TONGYI_MAX_MODEL, ALI_TONGYI_TURBO_MODEL
import numexpr
import math

load_dotenv()
# 初始化大模型客户端，不同的模型表现不同
llm = get_lc_o_ali_model_client(model=ALI_TONGYI_MAX_MODEL)
@tool
def calculator(expression: str) -> str:
    """使用Python的numexpr库计算表达式。表达式应该是解决问题的单行数学表达式。.
    例子:
        "37593 * 67"
        "37593**(1/5)"
        "37593^(1/5)"
    """
    local_dict = {"pi": math.pi, "e": math.e}
    return str(
        numexpr.evaluate(
            expression.strip(),
            global_dict={},
            local_dict=local_dict,  # 添加常用数学函数
        )
    )
# 设置工具
search = TavilySearch(max_results=3)
tools = [calculator,search]

# LangChain 1.0 的 create_agent 函数实际上是构建在 LangGraph 引擎之上的，
# 它本身就实现了 ReAct 模式，
# 虽然不像之前的版本那样在提示词中明确写出"Thought/Action/Observation"的格式，
# 但底层机制仍然遵循这个循环模式。可以通过debug=True参数来查看详细日志输出
agent = create_agent(
    model=llm,
    tools=tools,
    system_prompt="尽你所能回答以下问题。如果能力不够你可以使用以下工具:",
    debug=True
)

# 执行Agent
result = agent.invoke({
    "messages": [
        {"role": "user", "content": "在中国，目前市场上玫瑰花的一般进货价格是多少？大概是多少钱一支？\n如果我在此基础上加价5%，应该如何定价？"}
    ]
})

# 提取并显示最终答案
final_answer = result['messages'][-1].content
print("最后结果为：")
print(final_answer)

3.2、AI Agent策略蓝图：Plan-and-Execute

计划与执行（Plan-and-Execute）框架侧重于先规划一系列的行动，然后执行。这个框架可以使大模型能够先综合考虑任务的多个方面，然后按照计划进行行动。应用在比较复杂的项目管理中或者需要多步决策的场景下会比较合适。

Plan-and-Execute任务流程

v0.3版本

from langchain_experimental.plan_and_execute import PlanAndExecute, load_agent_executor, load_chat_planner
from langchain import SerpAPIWrapper
from langchain.agents import Tool
from langchain import LLMMathChain
from dotenv import load_dotenv

from models import get_lc_o_ali_model_client

load_dotenv()
# 配置API密钥和基础URL
llm = get_lc_o_ali_model_client()
# 创建工具
search = SerpAPIWrapper()
llm_math_chain = LLMMathChain(llm=llm, verbose=True)

# 定义工具列表
tools = [
    Tool(
        name="Search",
        func=search.run,
        description="用于回答关于当前事件的问题"
    ),
    Tool(
        name="Calculator",
        func=llm_math_chain.run,
        description="用于计算或解决问题，只能处理简单的数学表达式，不能处理列表或复杂数据结构"
    )
]

#1.0没有了
planner = load_chat_planner(llm)
executor = load_agent_executor(llm, tools, verbose=True)
agent = PlanAndExecute(planner=planner, executor=executor, verbose=True)
print(agent.invoke({"input": "在中国，100人民币能买几束玫瑰花？请用中文回答问题"}))

V1.0版本

import sys
import os
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from dotenv import load_dotenv
from langchain_classic.chains.llm_math.base import LLMMathChain
from langchain_community.utilities import SerpAPIWrapper
from langchain_core.tools import Tool

from models import get_lc_o_ali_model_client
# 导入我们自己实现的Plan-and-Execute框架
from cognitive.plan_and_execute_agent_for_lcV1 import create_plan_and_execute_agent

load_dotenv()

# 配置API密钥和基础URL
llm = get_lc_o_ali_model_client()

# 创建工具
search = SerpAPIWrapper()
llm_math_chain = LLMMathChain(llm=llm, verbose=True)

# 定义工具列表
tools = [
    Tool(
        name="Search",
        func=search.run,
        description="用于回答关于当前事件的问题"
    ),
    Tool(
        name="Calculator",
        func=llm_math_chain.run,
        description="用于计算或解决问题，只能处理简单的数学表达式，不能处理列表或复杂数据结构"
    )
]

# 使用我们自己实现的Plan-and-Execute框架
agent = create_plan_and_execute_agent(tools, verbose=True)
result = agent.run("在中国，100人民币能买几束玫瑰花？请用中文回答问题")
print(result["final_answer"])

class Step(BaseModel):
    """表示计划中的单个步骤"""
    id: int = Field(..., description="步骤的ID")
    description: str = Field(..., description="步骤的描述")
    status: str = Field("pending", description="步骤的状态 (pending, completed, failed)")


class Plan(BaseModel):
    """表示完整的计划"""
    steps: List[Step] = Field(default_factory=list, description="计划中的步骤列表")


class Planner:
    """负责生成计划的规划器"""
    
    def __init__(self, llm):
        self.llm = llm
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", "你是一个专业的任务规划师。你的任务是将用户的问题分解成具体的执行步骤。"
                       "请只输出JSON格式的计划，不要包含其他文本。\n\n"
                       "输出格式示例:\n"
                       "{{\"steps\": [{{\"id\": 1, \"description\": \"步骤1的描述\"}}, {{\"id\": 2, \"description\": \"步骤2的描述\"}}]}}"),
            ("human", "请为以下任务制定一个详细的执行计划:\n{input}")
        ])
        self.chain = self.prompt | self.llm

    def plan(self, input_str: str) -> Plan:
        """为给定的输入生成计划"""
        response = self.chain.invoke({"input": input_str})
        # 解析LLM的响应为Plan对象
        import json
        try:
            plan_data = json.loads(response.content)
            steps = [Step(**step) for step in plan_data.get("steps", [])]
            return Plan(steps=steps)
        except (json.JSONDecodeError, KeyError):
            # 如果解析失败，创建一个默认计划
            return Plan(steps=[Step(id=1, description=input_str)])


class Executor:
    """负责执行计划步骤的执行器"""
    
    def __init__(self, llm, tools: List[BaseTool]):
        self.llm = llm
        self.tools = tools
        
        # 使用LangChain 1.0的新推荐方式 create_agent
        self.agent = create_agent(
            model=self.llm,
            tools=self.tools,
            system_prompt="你是一个执行代理人。你的任务是执行给定的单个步骤。"
                         "请使用工具来帮助你完成任务，并给出清晰的结果总结。"
        )

    def execute_step(self, step_description: str, context: str = "") -> str:
        """执行单个步骤"""
        try:
            # 构造输入消息
            input_message = f"请执行以下步骤:\n{step_description}\n\n这是之前步骤的结果:\n{context}"
            
            # 使用新的invoke方式执行agent
            result = self.agent.invoke({
                "messages": [
                    {"role": "user", "content": input_message}
                ]
            })
            
            # 提取最终的回答内容
            if isinstance(result, dict) and "messages" in result:
                messages = result["messages"]
                if messages and hasattr(messages[-1], 'content'):
                    return messages[-1].content
                elif messages:
                    return str(messages[-1])
            
            return str(result)
        except Exception as e:
            return f"执行步骤时出错: {str(e)}"


class PlanAndExecuteAgent:
    """Plan-and-Execute Agent主类"""
    
    def __init__(self, planner: Planner, executor: Executor):
        self.planner = planner
        self.executor = executor

    def run(self, input_str: str) -> Dict[str, Any]:
        """运行整个Plan-and-Execute流程"""
        # 1. 制定计划
        print("🔧 正在制定计划...")
        plan = self.planner.plan(input_str)
        
        if not plan.steps:
            return {"error": "无法生成有效的计划"}

        print(f"📋 计划已生成，共 {len(plan.steps)} 个步骤:")
        for i, step in enumerate(plan.steps, 1):
            print(f"  {i}. {step.description}")

        # 2. 依次执行步骤
        context = ""
        results = []
        
        for i, step in enumerate(plan.steps):
            print(f"\n⚡ 正在执行步骤 {i+1}/{len(plan.steps)}: {step.description}")
            result = self.executor.execute_step(step.description, context)
            results.append(result)
            context += f"\n步骤 {i+1} 的结果: {result}"
            print(f"✅ 步骤 {i+1} 完成，结果: {result}")

        # 3. 生成最终答案
        print("\n🎯 正在生成最终答案...")
        final_prompt = ChatPromptTemplate.from_messages([
            ("system", "你是一个总结专家。你的任务是根据提供的步骤和结果，给出针对原始问题的最终答案。"),
            ("human", "原始问题: {input}\n\n执行过程和结果:\n{context}\n\n请给出最终答案:")
        ])
        
        final_chain = final_prompt | self.planner.llm
        final_response = final_chain.invoke({
            "input": input_str,
            "context": context
        })
        
        final_answer = final_response.content if hasattr(final_response, 'content') else str(final_response)
        
        return {
            "input": input_str,
            "plan": plan,
            "intermediate_steps": results,
            "final_answer": final_answer
        }


def create_plan_and_execute_agent(tools: List[BaseTool], **kwargs) -> PlanAndExecuteAgent:
    """创建一个Plan-and-Execute Agent的便捷函数"""
    llm = get_lc_o_ali_model_client()
    
    planner = Planner(llm)
    executor = Executor(llm, tools)
    
    return PlanAndExecuteAgent(planner, executor)

3.3、AI Agent回路验证：Self-Ask

自问自答（Self-Ask）框架这个允许大模型对自己提出问题并回答，来增强对问题的理解以提高回答质量，这个框架在需要深入分析或者提供创造性解决方案下比较适合，例如创意写作。

Self-Ask自问自答

V0.3版本

import os

from langchain_classic.agents import create_self_ask_with_search_agent, AgentExecutor
from langchain_tavily import TavilySearch
from dotenv import load_dotenv

from models import get_lc_o_ali_model_client, ALI_TONGYI_MAX_MODEL, ALI_TONGYI_DEEPSEEK_V3
import langchain
load_dotenv()
#开启调试模型
langchain.debug = True
"""
FIREWORKS_API_KEY = "fw_3ZdBAq9xu2sB59kyLQTD2kmj
TAVILY_API_KEY = "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"
"""

# from langchain_fireworks import ChatFireworks
# llm = ChatFireworks(
#     api_key=os.getenv("FIREWORKS_API_KEY"),
#     model="accounts/fireworks/models/deepseek-v3",
#     max_tokens=256,
# )
llm = get_lc_o_ali_model_client()

# prompt = hub.pull("hwchase17/self-ask-with-search")
# print("hwchase17/self-ask-with-search:",prompt)
from langchain_core.prompts import PromptTemplate
template = '''
Answer the following question. When needed, you can ask follow-up questions and answer them to reach the final answer. Always use English format for follow-up questions and answers:

Examples:
Question: Who lived longer, Muhammad Ali or Alan Turing?
Are follow up questions needed here: Yes.
Follow up: How old was Muhammad Ali when he died?
Intermediate answer: Muhammad Ali was 74 years old when he died.
Follow up: How old was Alan Turing when he died?
Intermediate answer: Alan Turing was 41 years old when he died.
So the final answer is: Muhammad Ali

Question: When was the founder of craigslist born?
Are follow up questions needed here: Yes.
Follow up: Who founded craigslist?
Intermediate answer: Craigslist was founded by Craig Newmark.
Follow up: When was Craig Newmark born?
Intermediate answer: Craig Newmark was born on December 6, 1952.
So the final answer is: December 6, 1952

Question: Who was George Washington's paternal grandfather?
Are follow up questions needed here: Yes.
Follow up: Who was George Washington's father?
Intermediate answer: George Washington's father was Augustine Washington.
Follow up: Who was Augustine Washington's father?
Intermediate answer: Augustine Washington's father was Lawrence Washington.
So the final answer is: Lawrence Washington

IMPORTANT: 
1. ONLY use the exact format shown above
2. DO NOT include any explanations or extra text
3. DO NOT use markdown formatting like **bold**
4. ONLY output in the specified format
5. End your response with "So the final answer is: [answer]"
6. For questions you don't know, you can use tools to get answers to them.

Question: {input}
Are follow up questions needed here: {agent_scratchpad}'''
prompt = PromptTemplate.from_template(template)

# template = '''
# 在回答用户问题时，可以自己提出问题并回答，来增强对问题的理解以提高回答质量。
# 示例一：
# 问题：谁活得更久，穆罕默德·阿里还是阿兰·图灵？
# 这里是否需要后续问题：是的。
# 追问：穆罕默德·阿里去世时几岁？
# 中间答案：穆罕默德·阿里去世时74岁。
# 追问：阿兰·图灵去世时几岁？
# 中间答案：阿兰·图灵去世时41岁。
# 所以最终的答案是：穆罕默德·阿里
# 示例二：
# 问：craigslist的创始人是什么时候出生的？
# 这里是否需要后续问题：是的。
# 追问：craigslist的创始人是谁？
# 中间答案：Craigslist是由克雷格·纽马克创立的。
# 追问：克雷格·纽马克什么时候出生的？
# 中间答案：克雷格·纽马克出生于1952年12月6日。
# 所以最终的答案是：1952年12月6日
# 示例三：
# 问：乔治·华盛顿的外祖父是谁？
# 这里是否需要后续问题：是的。
# 下一篇：谁是乔治·华盛顿的母亲？
# 中间答案：乔治·华盛顿的母亲是玛丽·鲍尔·华盛顿。
# 追问：玛丽·鲍尔·华盛顿的父亲是谁？
# 中间答案：玛丽·鲍尔华盛顿的父亲是约瑟夫·鲍尔。
# 所以最终的答案是：约瑟夫·鲍尔
#
# 用户问题: {input}
# 解决问题:{agent_scratchpad}'''
# prompt = PromptTemplate.from_template(template)
#上面的提示词会执行报错，因为LangChain的self_ask.py中对大模型的输出结果规定了英文格式

tools = [TavilySearch(max_results=1, include_raw_content=True,name="Intermediate Answer",tavily_api_key=os.getenv("TAVILY_API_KEY"))]

agent = create_self_ask_with_search_agent(llm, tools, prompt)

agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True,handle_parsing_errors=True)

print(agent_executor.invoke({"input": "Are both the directors of Jaws and Casino Royale from the same country?"}))
"""
示例四：
问：《大白鲨》和《皇家赌场》的导演都来自同一个国家吗？
这里是否需要后续问题：是的。
追问：《大白鲨》的导演是谁？
中间答案：《大白鲨》的导演是史蒂文·斯皮尔伯格。
追问：史蒂文·斯皮尔伯格来自哪里？
中间答案：美国。
追问：皇家赌场的导演是谁？
中间答案：皇家赌场的导演是马丁·坎贝尔。
追问：马丁·坎贝尔来自哪里？
中间答案：新西兰。
所以最终的答案是：不是
"""