1.CASE1：本地知识智能客服

1.1Agent设计

AGENT_TMPL = "按照给定的格式回答以下问题。你可以使
用下面这些工具：
{tools}
回答时需要遵循以下用括起来的格式：

Question: 我需要回答的问题
Thought: 回答这个上述我需要做些什么
Action: “{tool_names}” 中的一个工具名
Action Input: 选择这个工具所需要的输入
Observation: 选择这个工具返回的结果
…（这个 思考/行动/行动输入/观察 可以重复N次）
Thought: 我现在知道最终答案
Final Answer: 原始输入问题的最终答案
现在开始回答，记得在给出最终答案前，需要按照指定格式
进行一步一步的推理。

Question: {input}
{agent_scratchpad}
“”
这个模版中，最重要的是：
Thought
Action
Action Input
这 3 个部分是 AI 在Chain中进行自我思考的过程

1.2Agent搭建

import re
from typing import List, Union
# Python内置模块，用于格式化和包装文本
import textwrap
import time

from langchain.agents import (
    Tool,          # 可用工具
    AgentExecutor, # Agent执行
    LLMSingleActionAgent, # 定义Agent
    AgentOutputParser, # 输出结果解析
)
from langchain.prompts import StringPromptTemplate
# LLMChain，包含一个PromptTemplate和一个LLM
from langchain_community.llms import Tongyi  # 导入通义千问Tongyi模型
from langchain import LLMChain
# Agent执行，Agent结束
from langchain.schema import AgentAction, AgentFinish
# PromptTemplate: 管理LLMs的Prompts
from langchain.prompts import PromptTemplate
from langchain.llms.base import BaseLLM

# 定义了LLM的Prompt Template
CONTEXT_QA_TMPL = """
根据以下提供的信息，回答用户的问题
信息：{context}

问题：{query}
"""
CONTEXT_QA_PROMPT = PromptTemplate(
    input_variables=["query", "context"],
    template=CONTEXT_QA_TMPL,
)

# 输出结果显示，每行最多60字符，每个字符显示停留0.1秒（动态显示效果）
def output_response(response: str) -> None:
    if not response:
        exit(0)
    # 每行最多60个字符
    for line in textwrap.wrap(response, width=60):
        for word in line.split():
            for char in word:
                print(char, end="", flush=True)
                time.sleep(0.1)  # Add a delay of 0.1 seconds between each character
            print(" ", end="", flush=True)  # Add a space between each word
        print()  # Move to the next line after each line is printed
    # 遇到这里，这个问题的回答就结束了
    print("----------------------------------------------------------------")

# 模拟公司产品和公司介绍的数据源
class TeslaDataSource:
    def __init__(self, llm: BaseLLM):
        self.llm = llm

    # 工具1：产品描述
    def find_product_description(self, product_name: str) -> str:
        """模拟公司产品的数据库"""
        product_info = {
            "Model 3": "具有简洁、动感的外观设计，流线型车身和现代化前脸。定价23.19-33.19万",
            "Model Y": "在外观上与Model 3相似，但采用了更高的车身和更大的后备箱空间。定价26.39-36.39万",
            "Model X": "拥有独特的翅子门设计和更加大胆的外观风格。定价89.89-105.89万",
        }
        # 基于产品名称 => 产品描述
        return product_info.get(product_name, "没有找到这个产品")

    # 工具2：公司介绍
    def find_company_info(self, query: str) -> str:
        """模拟公司介绍文档数据库，让llm根据信息回答问题"""
        context = """
        特斯拉最知名的产品是电动汽车，其中包括Model S、Model 3、Model X和Model Y等多款车型。
        特斯拉以其技术创新、高性能和领先的自动驾驶技术而闻名。公司不断推动自动驾驶技术的研发，并在车辆中引入了各种驾驶辅助功能，如自动紧急制动、自适应巡航控制和车道保持辅助等。
        """
        # prompt模板 = 上下文context + 用户的query
        prompt = CONTEXT_QA_PROMPT.format(query=query, context=context)
        # 使用LLM进行推理
        return self.llm(prompt)


AGENT_TMPL = """按照给定的格式回答以下问题。你可以使用下面这些工具：

{tools}

回答时需要遵循以下用---括起来的格式：

---
Question: 我需要回答的问题
Thought: 回答这个上述我需要做些什么
Action: "{tool_names}" 中的一个工具名
Action Input: 选择这个工具所需要的输入
Observation: 选择这个工具返回的结果
...（这个 思考/行动/行动输入/观察 可以重复N次）
Thought: 我现在知道最终答案
Final Answer: 原始输入问题的最终答案
---

现在开始回答，记得在给出最终答案前，需要按照指定格式进行一步一步的推理。

Question: {input}
{agent_scratchpad}
"""


class CustomPromptTemplate(StringPromptTemplate):
    template: str  # 标准模板
    tools: List[Tool]  # 可使用工具集合

    def format(self, **kwargs) -> str:
        """
        按照定义的 template，将需要的值都填写进去。
        Returns:
            str: 填充好后的 template。
        """
        # 取出中间步骤并进行执行
        intermediate_steps = kwargs.pop("intermediate_steps")  
        print('intermediate_steps=', intermediate_steps)
        print('='*30)
        thoughts = ""
        for action, observation in intermediate_steps:
            thoughts += action.log
            thoughts += f"\nObservation: {observation}\nThought: "
        # 记录下当前想法 => 赋值给agent_scratchpad
        kwargs["agent_scratchpad"] = thoughts  
        # 枚举所有可使用的工具名+工具描述
        kwargs["tools"] = "\n".join(
            [f"{tool.name}: {tool.description}" for tool in self.tools]
        )  
        # 枚举所有的工具名称
        kwargs["tool_names"] = ", ".join(
            [tool.name for tool in self.tools]
        )  
        cur_prompt = self.template.format(**kwargs)
        #print(cur_prompt)
        return cur_prompt

"""
    对Agent返回结果进行解析，有两种可能：
    1）还在思考中 AgentAction
    2）找到了答案 AgentFinal
"""
class CustomOutputParser(AgentOutputParser):
    def parse(self, llm_output: str) -> Union[AgentAction, AgentFinish]:
        """
        解析 llm 的输出，根据输出文本找到需要执行的决策。
        Args:
            llm_output (str): _description_
        Raises:
            ValueError: _description_
        Returns:
            Union[AgentAction, AgentFinish]: _description_
        """
        # 如果句子中包含 Final Answer 则代表已经完成
        if "Final Answer:" in llm_output:  
            return AgentFinish(
                return_values={"output": llm_output.split("Final Answer:")[-1].strip()},
                log=llm_output,
            )

        # 需要进行 AgentAction
        regex = r"Action\s*\d*\s*:(.*?)\nAction\s*\d*\s*Input\s*\d*\s*:[\s]*(.*)"  # 解析 action_input 和 action
        match = re.search(regex, llm_output, re.DOTALL)
        if not match:
            raise ValueError(f"Could not parse LLM output: `{llm_output}`")
        action = match.group(1).strip()
        action_input = match.group(2)
        # Agent执行
        return AgentAction(
            tool=action, tool_input=action_input.strip(" ").strip('"'), log=llm_output
        )

# 设置通义千问API密钥
DASHSCOPE_API_KEY = 'sk-'

if __name__ == "__main__":
    # 定义LLM
    llm = Tongyi(model_name="qwen-turbo", dashscope_api_key=DASHSCOPE_API_KEY)  # 使用通义千问qwen-turbo模型
    # 自有数据
    tesla_data_source = TeslaDataSource(llm)
    # 定义的Tools
    tools = [
        Tool(
            name="查询产品名称",
            func=tesla_data_source.find_product_description,
            description="通过产品名称找到产品描述时用的工具，输入的是产品名称",
        ),
        Tool(
            name="公司相关信息",
            func=tesla_data_source.find_company_info,
            description="当用户询问公司相关的问题，可以通过这个工具了解公司信息",
        ),
    ]
    
    # 用户定义的模板
    agent_prompt = CustomPromptTemplate(
        template=AGENT_TMPL,
        tools=tools,
        input_variables=["input", "intermediate_steps"],
    )
    # Agent返回结果解析
    output_parser = CustomOutputParser()
    # 最常用的Chain, 由LLM + PromptTemplate组成
    llm_chain = LLMChain(llm=llm, prompt=agent_prompt)
    # 定义的工具名称
    tool_names = [tool.name for tool in tools]
    # 定义Agent = llm_chain + output_parser + tools_names
    agent = LLMSingleActionAgent(
        llm_chain=llm_chain,
        output_parser=output_parser,
        stop=["\nObservation:"],
        allowed_tools=tool_names,
    )
    # 定义Agent执行器 = Agent + Tools
    agent_executor = AgentExecutor.from_agent_and_tools(
        agent=agent, tools=tools, verbose=True
    )

    # 主过程：可以一直提问下去，直到Ctrl+C
    while True:
        try:
            user_input = input("请输入您的问题：")
            response = agent_executor.run(user_input)
            output_response(response)
        except KeyboardInterrupt:
            break

LangChain和Dashscope的区别？
LangChain = LLM application
Dashscope 保存了很多LLM的调用API
工具是LLM执行的吗？不是Agent执行的吗？
Agent是智能体，我们会给Agent配置各种tool, memory, prompt, LLM
LLM是Agent的大脑，Agent如果要决定使用哪个工具，是通过LLM来进行推理。

如何在LangChain中自定义工具？
1）在tools中添加我们的Tool
tools = [
Tool(
name=“查询产品名称”,
func=tesla_data_source.find_product_description,
description=“通过产品名称找到产品描述时用的工具，输入的是产品名称”,
),
Tool(
name=“公司相关信息”,
func=tesla_data_source.find_company_info,
description=“当用户询问公司相关的问题，可以通过这个工具了解公司信息”,
),
]
2) 实现该tool的函数
find_product_description

tools在langchain中，你可以自己定义自己的tool，也可以使用langchain已经集成好的tool

2.ReAct范式

ReAct: Synergizing Reasoning and Acting in Language
Models, 2022 https://arxiv.org/abs/2210.03629
• 将推理和动作相结合，克服LLM胡言乱语的问题，同时提高了结果的可解释性和可信赖度
Thinking：作者对人类的行为的洞察：
人们在从事一项需要多个步骤的任务时，在步骤和步骤之间，或者动作和动作之间，一般都会有推理过程。

Thinking1：LLM选择工具也是用相似度来选择的吗？
就是通过LLM的prompt进行提问，来选择的
用户的需求是XXX
我现在有XXX工具，
如果需要，我可以选择其中一个工具来执行；
（是通过LLM的推理来完成的）

• Agent 的核心是把 LLM 当作推理引擎，让它能使用外部工具，以及自己的长期记忆，从而完成灵活的决策步骤，进行复杂任务
• LangChain 里的 Chain 的概念，是由人来定义的一套流程步骤来让 LLM 执行，可以看成是把LLM 当成了一个强大的多任务工具典型的 Agent 逻辑（比如 ReAct）：
• 由 LLM 选择工具。
• 执行工具后，将输出结果返回给 LLM
• 不断重复上述过程，直到达到停止条件，通常是 LLM 自己认为找到答案了

from langchain.agents import Tool, AgentExecutor, create_react_agent
from langchain.prompts import PromptTemplate
from langchain.schema import AgentAction, AgentFinish
from langchain_community.llms import Tongyi
from langchain.memory import ConversationBufferMemory
import re
import json
from typing import List, Union, Dict, Any
import os

# 设置通义千问API密钥
DASHSCOPE_API_KEY = 'sk-'

# 自定义工具1：文本分析工具
class TextAnalysisTool:
    """文本分析工具，用于分析文本内容"""
    
    def __init__(self):
        self.name = "文本分析"
        self.description = "分析文本内容，提取字数、字符数和情感倾向"
    
    def run(self, text: str) -> str:
        """分析文本内容
        
        参数:
            text: 要分析的文本
        返回:
            分析结果
        """
        # 简单的文本分析示例
        word_count = len(text.split())
        char_count = len(text)
        
        # 简单的情感分析（示例）
        positive_words = ["好", "优秀", "喜欢", "快乐", "成功", "美好"]
        negative_words = ["差", "糟糕", "讨厌", "悲伤", "失败", "痛苦"]
        
        positive_count = sum(1 for word in positive_words if word in text)
        negative_count = sum(1 for word in negative_words if word in text)
        
        sentiment = "积极" if positive_count > negative_count else "消极" if negative_count > positive_count else "中性"
        
        return f"文本分析结果:\n- 字数: {word_count}\n- 字符数: {char_count}\n- 情感倾向: {sentiment}"

# 自定义工具2：数据转换工具
class DataConversionTool:
    """数据转换工具，用于在不同格式之间转换数据"""
    
    def __init__(self):
        self.name = "数据转换"
        self.description = "在不同数据格式之间转换，如JSON、CSV等"
    
    def run(self, input_data: str, input_format: str, output_format: str) -> str:
        """转换数据格式
        
        参数:
            input_data: 输入数据
            input_format: 输入格式
            output_format: 输出格式
        返回:
            转换后的数据
        """
        try:
            if input_format.lower() == "json" and output_format.lower() == "csv":
                # JSON到CSV的转换示例
                data = json.loads(input_data)
                if isinstance(data, list):
                    if not data:
                        return "空数据"
                    
                    # 获取所有可能的列
                    headers = set()
                    for item in data:
                        headers.update(item.keys())
                    headers = list(headers)
                    
                    # 创建CSV
                    csv = ",".join(headers) + "\n"
                    for item in data:
                        row = [str(item.get(header, "")) for header in headers]
                        csv += ",".join(row) + "\n"
                    
                    return csv
                else:
                    return "输入数据必须是JSON数组"
            
            elif input_format.lower() == "csv" and output_format.lower() == "json":
                # CSV到JSON的转换示例
                lines = input_data.strip().split("\n")
                if len(lines) < 2:
                    return "CSV数据至少需要标题行和数据行"
                
                headers = lines[0].split(",")
                result = []
                
                for line in lines[1:]:
                    values = line.split(",")
                    if len(values) != len(headers):
                        continue
                    
                    item = {}
                    for i, header in enumerate(headers):
                        item[header] = values[i]
                    result.append(item)
                
                return json.dumps(result, ensure_ascii=False, indent=2)
            
            else:
                return f"不支持的转换: {input_format} -> {output_format}"
        
        except Exception as e:
            return f"转换失败: {str(e)}"

# 自定义工具3：文本处理工具
class TextProcessingTool:
    """文本处理工具，用于处理文本内容"""
    
    def __init__(self):
        self.name = "文本处理"
        self.description = "处理文本内容，如查找、替换、统计等"
    
    def run(self, operation: str, content: str, **kwargs) -> str:
        """处理文本内容
        
        参数:
            operation: 操作类型
            content: 文本内容
            **kwargs: 其他参数
        返回:
            处理结果
        """
        if operation == "count_lines":
            return f"文本共有 {len(content.splitlines())} 行"
        
        elif operation == "find_text":
            search_text = kwargs.get("search_text", "")
            if not search_text:
                return "请提供要查找的文本"
            
            lines = content.splitlines()
            matches = []
            
            for i, line in enumerate(lines):
                if search_text in line:
                    matches.append(f"第 {i+1} 行: {line}")
            
            if matches:
                return f"找到 {len(matches)} 处匹配:\n" + "\n".join(matches)
            else:
                return f"未找到文本 '{search_text}'"
        
        elif operation == "replace_text":
            old_text = kwargs.get("old_text", "")
            new_text = kwargs.get("new_text", "")
            
            if not old_text:
                return "请提供要替换的文本"
            
            new_content = content.replace(old_text, new_text)
            count = content.count(old_text)
            
            return f"替换完成，共替换 {count} 处。\n新内容:\n{new_content}"
        
        else:
            return f"不支持的操作: {operation}"

# 创建工具链
def create_tool_chain():
    """创建工具链"""
    # 创建工具
    text_analysis = TextAnalysisTool()
    data_conversion = DataConversionTool()
    text_processing = TextProcessingTool()
    
    # 组合工具
    tools = [
        Tool(
            name=text_analysis.name,
            func=text_analysis.run,
            description="分析文本内容，提取字数、字符数和情感倾向"
        ),
        Tool(
            name=data_conversion.name,
            func=data_conversion.run,
            description="在不同数据格式之间转换，如JSON、CSV等"
        ),
        Tool(
            name=text_processing.name,
            func=text_processing.run,
            description="处理文本内容，如查找、替换、统计等"
        )
    ]
    
    # 初始化语言模型
    llm = Tongyi(model_name="qwen-turbo", dashscope_api_key=DASHSCOPE_API_KEY)
    
    # 创建提示模板
    prompt = PromptTemplate.from_template(
        """你是一个有用的AI助手，可以使用以下工具:
{tools}
可用工具名称: {tool_names}

使用以下格式:
问题: 你需要回答的问题
思考: 你应该始终思考要做什么
行动: 要使用的工具名称，必须是 [{tool_names}] 中的一个
行动输入: 工具的输入
观察: 工具的结果
... (这个思考/行动/行动输入/观察可以重复 N 次)
思考: 我现在已经有了最终答案
回答: 对原始问题的最终回答

开始!
问题: {input}
思考: {agent_scratchpad}"""
    )
    
    # 创建Agent
    agent = create_react_agent(llm, tools, prompt)
    
    # 创建Agent执行器
    agent_executor = AgentExecutor.from_agent_and_tools(
        agent=agent,
        tools=tools,
        memory=ConversationBufferMemory(memory_key="chat_history"),
        verbose=True,
        handle_parsing_errors=False  # 关闭自动重试, True会严格检查重试
    )
    
    return agent_executor

# 示例：使用工具链处理任务
def process_task(task_description):
    """
    使用工具链处理任务
    
    参数:
        task_description: 任务描述
    返回:
        处理结果
    """
    try:
        agent_executor = create_tool_chain()
        response = agent_executor.invoke({"input": task_description})
        return response["output"]  # 从返回的字典中提取输出
    except Exception as e:
        return f"处理任务时出错: {str(e)}"

# 示例用法
if __name__ == "__main__":
    # 示例1: 文本分析与处理
    task1 = "分析以下文本的情感倾向，并统计其中的行数：'这个产品非常好用，我很喜欢它的设计，使用体验非常棒！\n价格也很合理，推荐大家购买。\n客服态度也很好，解答问题很及时。'"
    print("任务1:", task1)
    print("结果:", process_task(task1))
    
    # 示例2: 数据格式转换
    task2 = "将以下CSV数据转换为JSON格式：'name,age,comment\n张三,25,这个产品很好\n李四,30,服务态度差\n王五,28,性价比高'"
    print("\n任务2:", task2)
    print("结果:", process_task(task2))

dify私有化部署，coze是可视化界面。

3.LCEL构建任务链

LCEL 是 LangChain 推出的链式表达式语言，支持用"|“操作符将各类单元（如Prompt、LLM、Parser等）组合。
每个”|"左侧的输出会自动作为右侧的输入，实现数据流式传递。
优势：
• 代码简洁，逻辑清晰，易于多步任务编排。
• 支持多分支、条件、并行等复杂链路。
• 易于插拔、复用和调试每个子任务。
典型用法：
• 串联：A | B | C，A的输出传给B，B的输出传给C。 • 分支：{"x": A, "y": B}，并行执行A和B。
• 支持流式：如 .stream() 方法可边生成边消费。