agent是要解决什么问题?

主要解决ai只会说不会做的问题。

大模型=大模型
function calling=大模型+skills
agent=大模型+skills+推理循环

这样是不是就理解了。

进化过程

例如查看目录的功能。

进化过程-关键词匹配

最原始的方式，如：
case1：用户说"帮我看看当前文件夹有啥" → 匹配失败 ❌
case2：用户说"显示一下文件列表" → 匹配失败 ❌
需要穷举所有可能的表达方式，这显然不可能。

进化过程-生产最佳实践

让LLM自己决定是否需要调用工具，以及调用哪个工具。
链路：
将tools赛给大模型 | 第一次调用大模型 | 看模型返回的数据中是否有工具相关信息，有的化则代理调用工具，将信息再拼接到上下文中 | 第二次调用大模型。

共调用了两次大模型：
第一次 # 大模型判断是否需要使用工具，需要的话返回工具列表
第二次 # 拼接了工具拿到的信息后，大模型再次生成

要不要用skills.md

这里的最佳实践并没有用到skills。
为什么没用到呢?skills不是很火吗?
是很火，只不过那也只是一种实现方式而已，对于非开发人员来说，skills更容易操作。
对于开发人员来说，纯pydantic类更好，更标准，也省去了维护两个位置的代码(python+skill.md)。

生产最佳实践

1、在tools文件夹下创建cube_tool.py
2、在services文件夹下创建tool_registry.py
3、调用模型时的嵌入
4、如果要新增工具类，在tools下维护即可自动注册上

1、在tools文件夹下创建cube_tool.py

代码：

from pydantic import BaseModel, Field

class CubeCalculatorInput(BaseModel):
    """立方计算器的输入参数模型"""
    number: float = Field(..., description="需要计算立方的数字，支持整数和小数")

def execute(input_data: CubeCalculatorInput) -> dict:
    """
    计算一个数字的立方（三次方）。
    当用户询问某个数的立方、三次方时使用此工具。
    """
    try:
        print("自定义立方计算方法开始")
        result = input_data.number ** 3
        return {
            "success": True,
            "result": result,
            "output": f"{input_data.number} 的立方是 {result}"
        }
    except Exception as e:
        return {"success": False, "output": f"计算失败: {str(e)}"}

2、在services文件夹下创建tool_registry.py

代码：

import importlib
import pkgutil
import inspect
import tools
from pydantic import BaseModel
from core.logger import log


class ToolRegistry:
    def __init__(self):
        self._tools = {}
        self._discover_tools()

    def _discover_tools(self):
        """自动发现并加载 tools 目录下所有包含 execute 函数的模块"""
        for importer, modname, ispkg in pkgutil.iter_modules(tools.__path__, tools.__name__ + "."):
            try:
                module = importlib.import_module(modname)
                if hasattr(module, 'execute'):
                    self._register_module(module, modname)
            except Exception as e:
                log.error(f"加载工具模块 {modname} 失败: {e}")

    def _register_module(self, module, modname):
        """解析模块并注册工具"""
        func = module.execute
        sig = inspect.signature(func)
        params = list(sig.parameters.values())

        # 约定：第一个参数必须是 Pydantic BaseModel
        if not params or not issubclass(params[0].annotation, BaseModel):
            log.warning(f"模块 {modname} 的 execute 函数第一个参数不是 Pydantic BaseModel，已跳过")
            return

        input_model = params[0].annotation
        tool_name = getattr(module, 'TOOL_NAME', modname.split('.')[-1])  # 允许自定义名称

        self._tools[tool_name] = {
            "module": module,
            "function": func,
            "input_model": input_model,
            "definition": {
                "type": "function",
                "function": {
                    "name": tool_name,
                    "description": func.__doc__ or "No description provided.",
                    "parameters": input_model.model_json_schema()
                }
            }
        }
        log.info(f"✅ 工具已自动注册: {tool_name}")

    def get_all_definitions(self):
        """获取所有工具的 JSON Schema 定义"""
        return [info["definition"] for info in self._tools.values()]

    def execute(self, tool_name: str, arguments: dict):
        """
        执行工具
        :param tool_name: 工具名称
        :param arguments: LLM 返回的参数字典
        """
        if tool_name not in self._tools:
            return {"success": False, "output": f"未知工具: {tool_name}"}

        try:
            tool_info = self._tools[tool_name]
            # ⭐ 关键：利用 Pydantic 进行严格的参数校验和类型转换
            validated_input = tool_info["input_model"](**arguments)
            result = tool_info["function"](validated_input)
            return result
        except Exception as e:
            log.error(f"工具 {tool_name} 执行异常: {e}")
            return {"success": False, "output": f"执行出错: {str(e)}"}


# 全局单例
tool_registry = ToolRegistry()

3、调用模型时的嵌入

这部分代码仅示意，代码：

 # 3. 获取所有已注册的工具定义 (自动从 tools 目录扫描)
        tools = tool_registry.get_all_definitions()

        # 4. 第一轮调用：让 LLM 决策
        response = client.chat.completions.create(
            model="qwen-plus",
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )

        assistant_message = response.choices[0].message
        tool_output = None
        tool_used = None

        # 5. 处理工具调用
        if assistant_message.tool_calls:
            tool_call = assistant_message.tool_calls[0]
            function_name = tool_call.function.name
            function_args = json.loads(tool_call.function.arguments)

            log.info(f"LLM 调用工具: {function_name}, 参数: {function_args}")

            # ⭐ 统一通过注册中心执行，自动完成参数校验
            tool_result = tool_registry.execute(function_name, function_args)
            tool_output = tool_result
            tool_used = function_name

            # 将结果返回给 LLM
            messages.append({"role": "assistant", "content": None, "tool_calls": [tool_call.dict()]})
            messages.append(
                {"role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(tool_result, ensure_ascii=False)})

            # 第二轮调用：生成最终回答
            final_response = client.chat.completions.create(model="qwen-plus", messages=messages, tools=tools)
            answer = final_response.choices[0].message.content
        else:
            answer = assistant_message.content or ""

环境

python #3.11

沿用之前的项目和环境。

langchain示例

步骤

不用换项目，在项目下新建文件夹agent_demo。
agent_demo文件夹下，再新建data.txt，main.py。

新建data.txt，输入内容

公司名称：未来科技 (FutureTech)
CEO: 艾琳·陈 (Erin Chen)
核心产品：量子助手 (QuantumAssistant)
成立时间：2024年
总部地点：上海张江

新建python文件

代码：

import os
import re
from typing import List, Dict, Any

# ================= 0. 网络环境修复 (关键) =================
# 强制清空代理环境变量，防止因为开启了全局代理导致无法连接阿里云国内节点
os.environ['HTTP_PROXY'] = ''
os.environ['HTTPS_PROXY'] = ''
os.environ['ALL_PROXY'] = ''
print("🚀 已清除环境变量代理，确保直连...")

# 基础导入
from langchain_core.tools import Tool
from langchain_core.runnables import RunnableLambda
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import CharacterTextSplitter

# OpenAI 库导入
from openai import OpenAI

# ================= 1. 配置 API KEY 和 客户端 =================

# 按照你的要求获取 Key
API_KEY = os.getenv("DASHSCOPE_API_KEY", "YOUR API KEY")
print(f"🔍 代码读取到的API_KEY：{API_KEY}")

# 打印 Key 的前8位进行调试确认（保护隐私）
key_preview = API_KEY[:8] + "..." if len(API_KEY) > 8 else "Key太短或无效"
print(f"🔍 代码读取到的API_KEY：{key_preview}")

# 初始化 OpenAI 客户端，指向阿里云百炼的兼容接口
client = OpenAI(
    api_key=API_KEY,  # 替换为你的实际 Key (sk-...)
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

# 指定模型 ID
# 注意：在百炼兼容模式下，建议使用短名称，去掉 "Qwen/" 前缀
MODEL_ID = "qwen2.5-72b-instruct"

print(f"🔄 正在连接模型: {MODEL_ID} ...")


# ================= 2. 定义工具函数 =================

def simple_calculator(expression: str) -> str:
    """计算数学表达式。"""
    try:
        if re.match(r'^[0-9\.\s\+\-\*\/\(\)\*\*]+$', expression):
            result = eval(expression)
            return str(result)
        else:
            return "无效的数学表达式"
    except Exception as e:
        return f"计算错误: {str(e)}"


def setup_rag_tool():
    if not os.path.exists("data.txt"):
        return None
    try:
        loader = TextLoader("data.txt", encoding="utf-8")
        docs = CharacterTextSplitter(chunk_size=500, chunk_overlap=50).split_documents(loader.load())
        embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")
        vectorstore = Chroma.from_documents(documents=docs, embedding=embeddings, persist_directory="./chroma_db")

        def search_func(query: str) -> str:
            results = vectorstore.similarity_search(query, k=2)
            return "\n".join([d.page_content for d in results])

        return Tool(
            name="CompanySearch",
            func=search_func,
            description="用于查询公司内部信息，如CEO、产品、成立时间等。"
        )
    except Exception as e:
        print(f"⚠️ 知识库工具创建失败: {e}")
        return None


# 创建工具列表
tools: List[Tool] = []
tools.append(Tool(name="Calculator", func=simple_calculator, description="用于计算数学表达式。"))
rag_tool = setup_rag_tool()
if rag_tool:
    tools.append(rag_tool)


# ================= 3. 手动实现 ReAct 逻辑 (OpenAI 格式) =================

def create_simple_agent(client, model_id, tools: List[Tool]):
    tool_map = {tool.name: tool.func for tool in tools}
    tool_names = "[" + ", ".join([tool.name for tool in tools]) + "]"

    # 系统提示词
    system_prompt = f"""你是一个智能助手。请使用以下工具来回答问题：

可用工具：{tool_names}

请严格按照以下格式回答：
问题：你需要回答的问题
思考：你应该总是思考该做什么
行动：要采取的行动，必须是以下之一：{tool_names}
行动输入：行动的输入
观察：行动的结果
...（这个思考/行动/行动输入/观察可以重复N次）
思考：我现在知道最终答案了
最终答案：对原始问题的最终答案

开始！"""

    def agent_logic(inputs: Dict[str, Any]) -> str:
        question = inputs["input"]
        # 初始化消息历史
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"问题：{question}\n思考："}
        ]

        thought_history = ""

        max_iterations = 5
        for _ in range(max_iterations):
            try:
                # 调用 OpenAI 接口
                response = client.chat.completions.create(
                    model=model_id,
                    messages=messages + [{"role": "assistant", "content": thought_history}],
                    temperature=0.7,
                    max_tokens=512
                )

                new_text = response.choices[0].message.content
                thought_history += new_text

                # 检查是否需要调用工具
                if "行动：" in new_text:
                    action_match = re.search(r"行动：\s*(\w+)", new_text)
                    action_input_match = re.search(r"行动输入：\s*(.+?)(?:\n|$)", new_text)

                    if action_match and action_input_match:
                        action_name = action_match.group(1)
                        action_input = action_input_match.group(1).strip()

                        if action_name in tool_map:
                            try:
                                observation = tool_map[action_name](action_input)
                                thought_history += f"\n观察：{observation}\n思考："
                            except Exception as e:
                                thought_history += f"\n观察：工具执行错误: {str(e)}\n思考："
                        else:
                            thought_history += f"\n观察：未知工具 '{action_name}'\n思考："
                    else:
                        thought_history += "\n思考："
                elif "最终答案：" in new_text:
                    final_answer = new_text.split("最终答案：")[-1].strip()
                    return final_answer
                else:
                    thought_history += "\n思考："

            except Exception as e:
                print(f"❌ API 调用错误: {e}")
                return f"调用模型失败: {e}"

        return thought_history

    return RunnableLambda(agent_logic)


# 创建智能体
agent = create_simple_agent(client, MODEL_ID, tools)

# ================= 4. 运行智能体 =================
print("\n🤖 云端智能体已启动! (输入 'quit' 退出)")

while True:
    try:
        user_input = input("\n👤 你: ").strip()
        if not user_input or user_input.lower() in ["quit", "exit"]:
            print("👋 再见！")
            break

        response = agent.invoke({"input": user_input})
        print(f"🤖 AI: {response}")

    except KeyboardInterrupt:
        print("\n👋 强制退出")
        break
    except Exception as e:
        print(f"❌ 发生错误: {e}")

不使用langchain，原代码实现agent

使用方法：
1、直接运行该代码，跑的是第一个问题，应该会
2、然后注释掉最后的第一个问题，换第二个问题运行下有正确答案。

import re
from openai import OpenAI
import os

# =================配置区=================
# API_KEY = "ms-你的Key"
# BASE_URL = "https://api.modelscope.cn/v1"
MODEL_NAME = "qwen2.5-72b-instruct"
# client = OpenAI(api_key=API_KEY, base_url=BASE_URL)

API_KEY = os.getenv("DASHSCOPE_API_KEY", "YOUR API KEY")
print(f"🔍 代码读取到的API_KEY：{API_KEY}")
client = OpenAI(
    api_key=API_KEY,  # 替换为你的实际 Key (sk-...)
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)




# =================1. 定义工具（这是手脚）=================
# 我们手动定义两个简单的工具，方便演示
def search_tool(query):
    """搜索工具：用于查询实时信息"""
    print(f"   🔍 [执行日志] 正在调用搜索引擎，查询关键词：'{query}'...")
    # 模拟网络延迟和返回结果
    return f"搜索结果：{query} 是未来科技的核心产品，发布于2024年。"


def calculator_tool(expression):
    """计算工具：用于数学计算"""
    print(f"   🧮 [执行日志] 正在调用计算器，计算公式：'{expression}'...")
    try:
        result = eval(expression)  # 注意：实际生产中 eval 有风险，这里仅做演示
        return f"计算结果：{result}"
    except:
        return "计算错误"


# 将工具注册到一个字典里，方便通过名字查找
tools_map = {
    "Search": search_tool,
    "Calculator": calculator_tool
}

# =================2. 定义提示词（这是大脑）=================
system_prompt = """
你是一个智能助手。你可以使用以下工具来回答用户的问题：

1. Search(query): 当你需要查询公司信息或实时新闻时使用。
2. Calculator(expression): 当你需要进行数学计算时使用。

请严格按照以下格式回答：

思考：你需要先思考该怎么做。
行动：只能从 [Search, Calculator] 中选择一个工具名称。
行动输入：工具所需的输入参数。
观察：等待系统返回结果（这一步由系统自动完成，你不需要写）。

如果不需要使用工具，直接回答：
思考：我可以直接回答。
最终答案：你的回答。
"""


# =================3. 核心执行循环（这是神经系统）=================
def run_agent_loop(user_question):
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_question}
    ]

    print(f"\n👤 用户提问: {user_question}")
    print("-" * 30)

    # 最多循环 5 次，防止死循环
    for step in range(5):
        print(f"\n--- 第 {step + 1} 轮思考 ---")

        # A. 调用大模型获取思考结果
        response = client.chat.completions.create(
            model=MODEL_NAME,
            messages=messages
        )
        ai_content = response.choices[0].message.content
        print(f"🧠 AI思考内容:\n{ai_content}")

        # B. 解析 AI 的回复，看它想干什么
        # 使用正则表达式提取 "行动：" 和 "行动输入："
        action_match = re.search(r"行动：(.*?)\n行动输入：(.*?)\n", ai_content)

        # 如果没有找到行动指令，说明它想直接回答（或者出错了）
        if not action_match:
            final_answer_match = re.search(r"最终答案：(.*)", ai_content, re.DOTALL)
            if final_answer_match:
                print(f"\n✅ 最终回答: {final_answer_match.group(1)}")
            else:
                print("\n❌ 无法理解 AI 的回复")
            break

        # C. 提取工具名称和参数
        tool_name = action_match.group(1).strip()
        tool_args = action_match.group(2).strip()

        print(f"🛠️ 识别到工具调用: {tool_name}({tool_args})")

        # D. 真正执行 Python 函数
        if tool_name in tools_map:
            tool_func = tools_map[tool_name]
            observation = tool_func(tool_args)  # <--- 这里就是真正的“执行”

            print(f"👁️ 观察结果: {observation}")

            # E. 将观察结果塞回给 AI，让它进行下一轮思考
            messages.append({"role": "assistant", "content": ai_content})
            messages.append({"role": "system", "content": f"观察：{observation}"})
        else:
            print(f"❌ 错误：AI 试图使用不存在的工具 {tool_name}")
            break


# =================运行=================
# 试着问一个需要查资料的问题
run_agent_loop("未来科技的核心产品是什么？")

# 试着问一个需要计算的问题
# run_agent_loop("帮我算一下 123 * 456 等于多少")

如何理解智能体

逻辑模块	代码中的位置	核心作用
工具库	tools: List[Tool]	提供外部能力（计算、查文档），打破模型知识限制。
大脑指令	system_prompt	强制模型遵循“先想后做”的逻辑，而不是直接瞎编答案。
感知-行动环	while 循环 + re.search	这是智能体的灵魂。它不断： 1. 听 (API Response) 2. 做 (Run Tool) 3. 回传 (Update History) 直到得出最终结论。

总结：
这段代码没有直接调用模型问“1+1等于几”然后直接返回，而是通过“思考 -> 计算 -> 观察 -> 回答”的闭环，证明了它是一个能调用工具解决问题的智能体，而不仅仅是一个聊天机器人。