​
一、本文知识可以涉及到的知识：

• python
• langchain
• langchain-agent
• langchain-tool
• langchain-retriever
• 向量数据库Qdrant
• HunYuan
• HunYuan Embeeding

二、本文所需要解决的场景

0. 尽量少的花费tokens（翻译一下啊：少花钱）
1. 一场进行中的会议，不断实时输出会议纪要
2. 一个人员入会后进行问题问答
3. 在我入会之前说了什么内容（这是一个总结性的问题，需要全文总结在其入会之前的内容）
4. 张书记说了什么内容（这是一个检索再总结的问题，需要先找到张书记的内容，在进行总结；当然也可以进行类似2.1的操作全文总结，但是会比较耗费token）
5. 问答返回需要流式返回
6. 用户的问题可能是各种各样的，比如我想知道会议说了什么、我想知道会议内容是什么等等都是要总结会议
   带着以上的问题和需求，我们本次的探索

三、方案设计
我们先看下设计图：

总体设计翻译如下：

我们准备了多个工具用于应对不同的问题，设计中需要的Tool（langchain tool）其实可以分为两类：
需要做全文总结的Tool
可以做RAG检索的Tool
我们使用模型的function_call用来分析客户的问题应该使用哪种工具，是全文总结还是RAG检索
四、代码实现
4.1 创建工具

from typing import Optional

from langchain_core.callbacks import CallbackManagerForToolRun
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import BaseTool
from langchain_openai import ChatOpenAI


class summary(BaseTool):
    name: str = "summary"
    description: str = "我可以总结会议纪要或者总结代办内容"
    return_direct: bool = True

    def _run(self, question: str, run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
        print("我来了，我是个总结小能手："+question)
        return "这是我总结的会议纪要"

class DoudiTool(BaseTool):
    name: str = "doudi"
    description: str = "我是兜底工具，其他function不能解决的问题都可以找我"
    return_direct: bool = True

    def _run(self, question: str, run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
        print("我来了，我是个万能助手："+question)
        return "我是万能助手"


class RAGTool(BaseTool):
    name: str = "Calculator"
    description: str = "useful for when you need to answer questions about math"
    # args_schema: Type[BaseModel] = CalculatorInput
    return_direct: bool = True

    def _run(self, question: str, run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
        """Use the tool."""
        print("我来了呀呀呀呀呀呀"+run_manager.metadata["name"])
        llm = ChatOpenAI(
            api_key="sk-xxxx", # 混元 APIKey
            base_url="https://api.hunyuan.cloud.tencent.com/v1", # 混元 endpoint
            model_name="hunyuan-pro",
            streaming=True
        )

        prompt = ChatPromptTemplate.from_messages(
            [("human", "{answer_style}")]
        )

        chain = prompt | llm | StrOutputParser()
        return chain.invoke({"answer_style":question})

4.2 创建agent

import asyncio

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableConfig
from langchain_openai import ChatOpenAI

from RAGTool import RAGTool, summary, DoudiTool

# 构造 client
chat = ChatOpenAI(
    api_key="sk-xxxxx",  # 混元 APIKey
    base_url="https://api.hunyuan.cloud.tencent.com/v1",  # 混元 endpoint
    model_name="hunyuan-functioncall",
    disable_streaming="tool_calling"
)

tools = [DoudiTool(),RAGTool(),summary()]

prompt = ChatPromptTemplate.from_messages(
    [
        ("system","你只能从给出的function中选择一个function，禁止自己回答问题"),
        ("user", "你只能从给出的function中选择一个function，禁止自己回答问题，请回答以下问题: {input}"),
        ("placeholder", "{agent_scratchpad}")
    ]
)

agent = create_tool_calling_agent(chat, tools, prompt)

# ------------创建Agent Executor-----------
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 执行查询
async def getData()-> None:
    runConfig = RunnableConfig(configurable={"name": "anan"})
    res = agent_executor.astream_events({"input": "我在干什么"}, config=runConfig, version="v2")
    async for event in res:
        if event["event"] == "on_chat_model_stream":
            print(event["data"]["chunk"])
        elif event["event"] == "on_chain_end":
            print(event)

if __name__ == '__main__':
    asyncio.run(getData())

4.3 运行
4.3.1 尝试使用【计算器】工具
入参则为：{“input”: “计算1+1=?”}

执行结果如下：

如上所示，正确的调用了计算函数，并且能够进行流式返回

4.3.2 尝试使用【兜底】工具

入参为：{“input”: “我在干什么”}

执行结果如下：

至此整个程序完成

idea插件推荐
AnNote - IntelliJ IDEs Plugin | Marketplace
75 折折扣：

MGRYF-TJW4N-WZMSJ-MZDLD-LVGJH

BTKQ8-XZLPH-L3QH3-MPKBH-BP9RR

本文由博客群发一文多发等运营工具平台 OpenWrite 发布