基于ReAct Agent 实现智能客服

在深入学习并实际操作 ReAct 框架之后，针对 Function Calling 未能解决的智能客服案例，我们将尝试采用 ReAct 框架来构建解决方案。首先，整体的项目架构如下图所示：

需要说明的是：AI Agent 的效果非常依赖于大模型的原生能力，所以如果使用小参数量的模型无法复现项目是正常现象。

本文可以带领新人做出实在可用的agent项目，后续我将再将基于框架的agent，请关注
客服服务代理类：

class CustomerServiceAgent:
    def __init__(self, client, config):
        self.client = client
        self.config = config
        self.messages = []
        self.system_prompt = """
        You are a Intelligent customer service assistant for e-commerce platform. It is necessary to answer the user's consultation about the product in a timely manner. If it has nothing to do with the specific product, you can answer it directly.
        output it as Answer: [Your answer here].

        Example :
        Answer: Is there anything else I can help you with

        If specific information about the product is involved, You run in a loop of Thought, Action, Observation.
        Use Thought to describe your analysis process.
        Use Action to run one of the available tools - then wait for an Observation.
        When you have a final answer, output it as Answer: [Your answer here].

        Available tools:
        1. query_by_product_name: Query the database to retrieve a list of products that match or contain the specified product name. This function can be used to assist customers in finding products by name via an online platform or customer support interface
        2. read_store_promotions: Read the store's promotion document to find specific promotions related to the provided product name. This function scans a text document for any promotional entries that include the product name.
        3. calculate: Calculate the final transaction price by combining the selling price and preferential information of the product

        IMPORTANT: When answering questions related to product prices and promotional information, you MUST use the available tools to obtain the data. You are NOT allowed to make assumptions or use your own inference to generate these details. The answers regarding prices and promotions should be based solely on the results retrieved from the tools.

        When using an Action, always format it as:
        Action: tool_name: argument1, argument2, ...

        Example :
        Human: Do you sell football in your shop? If you sell soccer balls, what are the preferential policies now? If I buy it now, how much will I get in the end?
        Thought: To answer this question, I need to check the database of the background first.
        Action: query_by_product_name: football

        Observation: At present, I have checked that the ball is in stock, and I know its price is 120 yuan.

        Thought: I need to further inquire about the preferential policy of football
        Action: read_store_promotions: football

        Observation: The current promotional policy for this ball is: 10% discount upon purchase

        Thought: Now I need to combine the selling price and preferential policies of the ball to calculate the final transaction price
        Action: calculate: 120 * 0.9

        Observation: The final price of the ball was 108.0 yuan

        Thought: I now have all the information needed to answer the question.
        Answer:  According to your enquiry, we do sell soccer balls in our store, the current price is 120 yuan. At present, we offer a 10% discount on the purchase of football. Therefore, if you buy now, the final transaction price will be 108 yuan.

        Note: You must reply to the final result in Chinese

        Now it's your turn:
        """.strip()
        self.messages.append({"role": "system", "content": self.system_prompt})

    def __call__(self, message):
        self.messages.append({"role": "user", "content": message})
        response = self.execute()
        if not isinstance(response, str):
            raise TypeError(f"Expected string response from execute, got {type(response)}")
        self.messages.append({"role": "assistant", "content": response})
        return response

    def execute(self):
        completion = self.client.chat.completions.create(
            model=self.config['openai']['model_name'],
            messages=self.messages,
        )
        response = completion.choices[0].message.content
        if response != None:
            return completion.choices[0].message.content
        else:
            return "当前没有正常的生成回复，请重新思考当前的问题，并再次进行尝试"

主循环逻辑：

def main():
    config = load_config()
    try:
        client = get_client(config)

        agent = CustomerServiceAgent(client, config)
    except Exception as e:
        print(f"Error initializing the AI client: {str(e)}")
        print("Please check your configuration and ensure the AI service is running.")
        return

    tools = {
        "query_by_product_name": query_by_product_name,
        "read_store_promotions": read_store_promotions,
        "calculate": calculate,
    }

您暂时无权查看此隐藏内容！
            except Exception as e:
                print(f"An error occurred while processing the query: {str(e)}")
                print("Please check your configuration and ensure the AI service is running.")
                break

            iteration += 1

        if iteration == max_iterations:
            print("Reached maximum number of iterations without a final answer.")

完整代码：

我这里使用了deepseek作为模型支持，总体表现还不错。