9万字长文讲透大模型智能体全栈指南：Function Calling、RAG、多智能体编排与生产级部署（附聚合API实践）
·
博主智算菩萨，专注于人工智能、Python编程、音视频处理及UI窗体程序设计等方向。致力于以通俗易懂的方式拆解前沿技术，从零基础入门到高阶实战，陪伴开发者共同成长。目前已开设五大技术专栏，累计发布多篇原创技术文章，深受读者好评。
📌 专栏导航
- 人工智能前沿知识：深度剖析Transformer架构、生成式AI、强化学习、具身智能、神经符号系统、大模型及智能体（Agent）技术，系统性解析AI核心技术体系与前沿趋势。
- Python基础小白编程：从零开始，以保姆式教程讲解变量、数据类型、流程控制、函数等核心语法，配有大量实战代码与避坑指南，真正做到学以致用。
- 机器学习与深度学习：系统化拆解线性模型、决策树、随机森林、梯度提升树、神经网络等算法原理与工程实践，覆盖从公式推导到代码实现的全链路内容。
- 音频、图像与视频处理理论与实战：涵盖FFmpeg多媒体处理、audio_shop开源工具、ComfyUI-WanVideoWrapper视频生成等实用技术，从基础操作到高级应用一应俱全。
- UI窗体程序设计实战：深入讲解UI设计、动态窗体生成、游戏UI框架设计等实战技巧，提供从配置到编码的完整解决方案。
智算菩萨，以代码为经，以算法为纬，在人工智能的星辰大海中，做你前行路上最可靠的导航者。
1 摘要
本文将带你从零开始，深入理解智能体的核心原理，手把手教你调用AI大模型API构建属于自己的智能体系统。内容涵盖API调用原理、Python编程实现、工具调用、RAG检索增强、多智能体协作、前端集成以及国产AI API接入等全栈技术。
如果你需要一个便捷的API聚合平台来获取各类大模型的调用接口，推荐使用 大模型API聚合站，它聚合了多种主流模型，注册即可获取API Key，开箱即用。
2 引言：智能体时代的到来
2.1 什么是智能体
智能体（Agent）是人工智能领域中一个核心概念，指的是能够感知环境、自主决策并采取行动以实现特定目标的系统。与传统的规则驱动程序不同，基于大语言模型（LLM）的智能体具备自然语言理解与生成能力，能够根据用户的意图灵活地选择工具、规划步骤并执行任务，从而在开放域场景中展现出远超传统程序的适应性和智能水平。
从技术演进的角度来看，智能体的发展经历了三个重要阶段。第一阶段是基于规则的专家系统，这类系统依赖人工编写的if-then规则，虽然逻辑清晰但扩展性极差，面对复杂场景时规则数量呈指数级增长，维护成本极高。第二阶段是基于强化学习的智能体，通过与环境交互获得奖励信号来优化策略，在游戏、机器人控制等封闭环境中取得了显著成果，但难以直接处理自然语言交互和开放域知识推理。第三阶段即当前基于大语言模型的智能体，它将LLM作为"大脑"，结合外部工具、记忆系统和规划能力，形成了一个能够理解自然语言指令、自主分解任务、调用工具并整合结果的完整智能系统。
智能体的核心特征可以概括为以下四个方面：第一，自主性（Autonomy），智能体能够在没有人类持续干预的情况下独立运行，根据目标自主选择行动路径；第二，反应性（Reactivity），智能体能够及时感知环境变化并做出相应调整；第三，主动性（Proactivity），智能体不仅仅被动响应，还能主动发起行动以实现目标；第四，社会性（Sociality），多个智能体之间能够通过通信协作完成复杂任务。这四个特征使得基于LLM的智能体在实际应用中展现出强大的任务处理能力。
2.2 为什么需要自己构建智能体
随着ChatGPT、GLM、KIMI等大语言模型的普及，许多人可能会问：既然已经有了现成的AI对话工具，为什么还需要自己构建智能体？答案在于通用AI工具与定制化智能体之间存在本质差异，而这种差异在实际应用场景中往往决定了项目的成败。
首先，通用AI工具的能力边界是固定的。无论是ChatGPT还是其他对话式AI，它们只能提供文本生成和简单的联网搜索功能，无法直接操作你的业务数据库、调用你的内部API或执行特定领域的专业计算。而自建智能体可以将LLM的推理能力与你的业务系统深度集成，让AI真正成为业务流程的一部分而非旁观者。例如，在电商场景中，你可以构建一个智能客服智能体，它不仅能回答用户问题，还能直接查询订单系统、处理退款请求、修改配送地址，真正实现端到端的服务闭环。
其次，数据隐私和安全合规是自建智能体的另一个重要驱动力。将敏感业务数据发送到第三方AI平台可能面临数据泄露风险和合规挑战。通过自建智能体，你可以完全控制数据的流向和存储方式，确保敏感信息不离开你的基础设施。这对于金融、医疗、法律等对数据安全要求极高的行业尤为重要。
第三，成本控制是自建智能体的现实考量。通用AI工具通常按月收费或按token计费，对于高频使用场景来说成本可能非常高昂。而自建智能体可以根据实际需求选择最经济的模型组合——简单任务使用免费或低成本模型，复杂任务才调用高性能模型——从而在保证质量的前提下大幅降低运营成本。
最后，自建智能体赋予你完全的定制自由。你可以根据业务需求设计专属的提示词策略、工具集、记忆机制和交互流程，打造出真正贴合业务场景的AI助手，而非被迫适应通用工具的固定模式。
2.3 本文的目标与路线图
本文的目标是让读者从零开始，系统掌握基于大模型API构建智能体的完整技术栈。无论你是刚接触AI编程的新手，还是有一定经验想要深入理解智能体架构的开发者，都能从本文中获得实用的知识和可运行的代码。
为了帮助读者建立清晰的学习路径，下面用流程图展示本文的整体技术路线：
上图展示了从概念理解到生产部署的完整学习路径。每个阶段都建立在前一阶段的基础之上，形成循序渐进的知识体系。其中橙色标注的核心功能实现、工具调用和RAG是本文的重点章节，建议读者重点掌握。
3 智能体的核心原理与架构
3.1 大语言模型的工作原理
要构建基于LLM的智能体，首先需要理解大语言模型本身的工作原理。大语言模型本质上是一个基于Transformer架构的神经网络，通过在海量文本数据上进行自监督学习，获得了强大的语言理解和生成能力。
Transformer架构的核心创新是自注意力机制（Self-Attention），它允许模型在处理每个token时同时关注输入序列中的所有其他token，从而捕获长距离依赖关系。具体来说，自注意力机制通过计算Query、Key、Value三个矩阵的交互来决定每个token应该"关注"输入序列中的哪些部分。多头注意力（Multi-Head Attention）进一步增强了这一能力，不同的注意力头可以关注不同类型的语义关系，有的关注语法结构，有的关注语义相似性，有的关注指代关系。
大语言模型的训练过程可以分为三个阶段。第一阶段是预训练（Pre-training），模型在海量无标注文本上学习预测下一个token，这一阶段赋予了模型广泛的世界知识和语言能力。第二阶段是有监督微调（Supervised Fine-Tuning, SFT），在高质量的指令-回答对上训练，使模型学会遵循指令生成回答。第三阶段是强化学习人类反馈（RLHF），通过人类偏好数据训练奖励模型，再用奖励模型指导策略优化，使模型的输出更符合人类期望。
从API调用的角度来看，大语言模型的核心能力可以抽象为"输入文本序列，输出文本序列"。这个看似简单的能力，在精心设计的提示词和工具调用机制的加持下，可以衍生出极其丰富的智能行为。下面的表格总结了LLM的核心能力及其在智能体中的应用：
| LLM核心能力 | 技术原理 | 智能体应用场景 |
|---|---|---|
| 文本生成 | 自回归解码，逐token预测 | 对话回复、报告撰写、代码生成 |
| 指令遵循 | SFT训练，学习指令-回答映射 | 按用户要求执行特定任务 |
| 推理规划 | Chain-of-Thought涌现能力 | 任务分解、步骤规划、逻辑推理 |
| 工具调用 | Function Calling训练 | 调用外部API、数据库查询、代码执行 |
| 上下文理解 | 长上下文窗口与注意力机制 | 多轮对话、文档理解、知识整合 |
| 知识检索 | 预训练知识 + RAG增强 | 问答系统、知识库查询、信息整合 |
3.2 智能体的架构设计
基于LLM的智能体架构是本文的核心理论基础。当前业界最广泛认可的智能体架构可以概括为"LLM作为大脑 + 四大组件"的范式，这四大组件分别是：规划（Planning）、记忆（Memory）、工具（Tools）和行动（Action）。
规划模块负责将复杂任务分解为可执行的子任务序列，并制定执行策略。常见的规划策略包括ReAct（Reasoning + Acting）、Plan-and-Execute、Tree of Thoughts等。ReAct策略的核心思想是让模型交替进行推理（Thought）和行动（Action），每一步推理都基于前一步的观察结果（Observation），形成"思考-行动-观察"的循环。Plan-and-Execute策略则先让模型生成完整的执行计划，再逐步执行，适合步骤明确且依赖关系清晰的任务。Tree of Thoughts策略让模型探索多条推理路径，通过评估和回溯找到最优解，适合需要创造性思考的开放性问题。
记忆模块为智能体提供信息持久化和上下文管理能力。记忆系统通常分为短期记忆和长期记忆两个层次。短期记忆对应LLM的上下文窗口，存储当前对话的完整历史，其容量受限于模型的上下文长度（如8K、32K、128K tokens等）。长期记忆则通过外部存储（如向量数据库、关系数据库）实现，可以保存跨越多次会话的用户偏好、历史交互记录和领域知识。记忆的检索通常基于语义相似度匹配，即通过向量搜索找到与当前查询最相关的历史信息。
工具模块扩展了智能体与外部世界交互的能力。LLM本身只能生成文本，但通过Function Calling机制，模型可以"请求"调用预定义的外部工具，如搜索引擎、数据库查询、API调用、代码执行器等。工具的定义包括名称、描述和参数schema，模型根据用户意图和工具描述自主决定是否调用工具以及传递什么参数。这种设计使得智能体的能力不再局限于文本生成，而是可以真正执行操作、获取实时信息并产生实际影响。
行动模块负责执行规划模块确定的行动方案，包括调用工具、生成回复或执行其他操作。行动模块是智能体与外部环境交互的桥梁，它将LLM的决策转化为具体的行为，并将执行结果反馈给LLM进行下一轮推理。
3.3 感知-思考-行动循环
智能体的运行机制可以抽象为一个持续运行的"感知-思考-行动"循环（Perceive-Think-Act Loop），也称为PTA循环。这个循环是智能体自主运行的核心引擎，理解它对于构建任何类型的智能体都至关重要。
PTA循环的每一次迭代都包含三个阶段。感知阶段，智能体接收来自用户或环境的输入信息，并从记忆系统中检索相关的历史上下文。思考阶段，LLM基于当前输入、历史上下文和可用工具描述进行推理，决定下一步行动。行动阶段，智能体执行LLM决定的行动，可能是调用工具获取信息，也可能是直接生成回复。如果调用了工具，工具的返回结果将作为新的观察信息进入下一轮循环，直到LLM认为已经收集到足够信息可以给出最终回复。
这个循环的设计灵感来自于经典的ReAct框架。ReAct的核心洞见是：将推理和行动交织在一起，比纯推理（如Chain-of-Thought）或纯行动（如行为克隆）都更有效。推理帮助模型制定和调整计划，行动帮助模型获取新信息和验证假设，两者相辅相成。
3.4 记忆系统详解
记忆系统是智能体区别于简单聊天机器人的关键组件。一个设计良好的记忆系统能够让智能体在长程交互中保持连贯性，积累用户偏好，并利用历史经验提升决策质量。
短期记忆的实现直接依赖于LLM的上下文窗口。当前主流模型的上下文窗口大小差异较大，从8K到128K甚至1M tokens不等。在实际应用中，我们需要在上下文长度和成本之间做出权衡：更长的上下文意味着更多的信息可以传递给模型，但也意味着更高的token消耗和更长的响应时间。常见的短期记忆管理策略包括滑动窗口（只保留最近N轮对话）、摘要压缩（将早期对话压缩为摘要）和重要性过滤（只保留与当前任务相关的对话片段）。
长期记忆的实现则需要外部存储系统的支持。向量数据库是目前最主流的长期记忆存储方案，其工作原理是将文本通过嵌入模型（Embedding Model）转换为高维向量，然后存储在支持向量相似度搜索的数据库中。当需要检索记忆时，将查询文本同样转换为向量，通过余弦相似度或内积搜索找到最相关的记忆片段。主流的向量数据库包括ChromaDB、FAISS、Pinecone、Milvus等，它们各有特点，适用于不同的应用场景。
下面的表格对比了常见的记忆管理策略：
| 策略 | 实现方式 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|---|
| 滑动窗口 | 保留最近N轮对话 | 实现简单，成本低 | 丢失早期重要信息 | 简单问答、闲聊 |
| 摘要压缩 | LLM总结历史对话 | 保留关键信息，节省token | 摘要可能丢失细节 | 长对话、客服场景 |
| 重要性过滤 | 基于相关性评分筛选 | 保留最相关信息 | 需要额外的评分模型 | 知识密集型任务 |
| 向量检索 | 语义相似度搜索 | 语义匹配能力强 | 需要向量数据库 | RAG、知识库问答 |
| 混合策略 | 组合多种方法 | 兼顾各方面需求 | 实现复杂度高 | 生产级智能体 |
4 API调用基础理论
4.1 REST API与HTTP协议
在构建智能体之前，我们需要深入理解API调用的底层原理。当前主流的大模型API都基于REST（Representational State Transfer）架构风格设计，通过HTTP协议进行通信。理解这些基础知识不仅有助于正确使用API，更能在遇到问题时快速定位和解决。
REST API的核心设计原则包括：使用标准HTTP方法（GET、POST、PUT、DELETE）表示操作类型，使用URL路径表示资源，使用HTTP状态码表示操作结果，使用JSON格式传输数据。大模型API通常只使用POST方法，因为文本生成本质上是一个创建（Create）操作——你向服务器发送一个生成请求，服务器返回生成的文本。
HTTP请求由三个核心部分组成：请求行（包含方法和URL）、请求头（包含认证信息、内容类型等元数据）和请求体（包含实际的数据载荷）。对于大模型API调用来说，最关键的请求头是Authorization（携带API Key进行身份认证）和Content-Type（指定请求体为JSON格式）。请求体则包含模型参数和对话消息。
HTTP响应同样由三个部分组成：状态行（包含状态码和状态描述）、响应头（包含内容类型、速率限制等信息）和响应体（包含模型生成的结果）。状态码是判断请求是否成功的关键指标，200表示成功，400表示请求格式错误，401表示认证失败，429表示请求频率超限，500表示服务器内部错误。
下面的表格列出了大模型API调用中常见的HTTP状态码及其处理方式：
| 状态码 | 含义 | 常见原因 | 处理方式 |
|---|---|---|---|
| 200 | 成功 | 请求正确 | 解析响应体获取结果 |
| 400 | 请求错误 | 参数格式错误、缺少必填字段 | 检查请求体格式和参数 |
| 401 | 认证失败 | API Key无效或过期 | 检查API Key是否正确 |
| 403 | 权限不足 | 无权访问该模型 | 检查账户权限和模型可用性 |
| 429 | 请求过多 | 超出速率限制 | 实现指数退避重试 |
| 500 | 服务器错误 | 服务端异常 | 稍后重试，检查服务状态 |
| 502 | 网关错误 | 上游服务不可用 | 稍后重试 |
| 503 | 服务不可用 | 服务器过载 | 稍后重试 |
4.2 OpenAI兼容API规范
当前大模型API领域的事实标准是OpenAI制定的API规范。绝大多数模型提供商（包括国内外的各类模型）都提供了与OpenAI API兼容的接口，这意味着你只需要学习一套API规范，就能通过更换base URL和模型名称来调用不同的模型。这种兼容性极大地降低了开发者的学习成本和迁移成本。
OpenAI兼容API的核心端点是/v1/chat/completions，用于聊天补全。请求体的基本结构如下：
{
"model": "模型名称",
"messages": [
{"role": "system", "content": "系统提示词"},
{"role": "user", "content": "用户消息"}
],
"temperature": 0.7,
"max_tokens": 2048,
"stream": false
}
其中，model字段指定要调用的模型名称；messages字段是一个消息数组，每条消息包含role（角色：system/user/assistant/tool）和content（内容）；temperature控制输出的随机性，0表示确定性输出，1表示最大随机性；max_tokens限制生成的最大token数；stream决定是否使用流式输出。
响应体的基本结构如下：
{
"id": "chatcmpl-xxx",
"object": "chat.completion",
"created": 1234567890,
"model": "模型名称",
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": "模型生成的回复"
},
"finish_reason": "stop"
}
],
"usage": {
"prompt_tokens": 10,
"completion_tokens": 20,
"total_tokens": 30
}
}
choices数组包含模型生成的回复，通常只有一个选择；usage字段提供了token消耗统计，这对于成本监控非常重要；finish_reason表示生成结束的原因，stop表示正常结束，length表示达到max_tokens限制，tool_calls表示模型请求调用工具。
4.3 认证与密钥管理
API Key是调用大模型API的身份凭证，相当于你的账户密码。API Key的安全管理直接关系到你的账户安全和费用安全——一旦API Key泄露，他人可以冒用你的身份调用模型，产生高额费用。
API Key的管理最佳实践包括：第一，永远不要将API Key硬编码在代码中，应该使用环境变量或配置文件存储；第二，不要将API Key提交到Git仓库，应该在.gitignore中排除包含密钥的配置文件；第三，为不同项目使用不同的API Key，便于权限管理和费用追踪；第四，定期轮换API Key，降低泄露风险；第五，设置API Key的使用额度限制，即使泄露也能控制损失。
在Python中，推荐使用python-dotenv库管理环境变量。创建一个.env文件存储API Key：
API_KEY=sk-your-api-key-here
BASE_URL=https://api.aigc.bar/v1
然后在代码中通过os.getenv()读取：
import os
from dotenv import load_dotenv
load_dotenv()
api_key = os.getenv("API_KEY")
base_url = os.getenv("BASE_URL")
这种方式既安全又灵活，切换不同的API提供商只需要修改.env文件，无需改动代码。
4.4 请求与响应格式详解
深入理解请求和响应的格式细节，对于构建复杂的智能体系统至关重要。本节将详细解析每个字段的含义和使用方式。
messages字段的详细说明：messages数组是API调用中最重要的字段，它定义了完整的对话上下文。每条消息的role字段有四种取值：system用于设定AI的角色和行为规范，通常放在消息数组的第一位；user表示用户发送的消息；assistant表示AI之前生成的回复，用于维持多轮对话的连贯性；tool表示工具调用的返回结果，与Function Calling配合使用。
一个完整的多轮对话消息数组示例如下：
{
"messages": [
{"role": "system", "content": "你是一个专业的Python编程助手。"},
{"role": "user", "content": "如何读取CSV文件？"},
{"role": "assistant", "content": "可以使用pandas库的read_csv函数..."},
{"role": "user", "content": "如果CSV文件编码不是UTF-8怎么办？"},
{"role": "assistant", "content": "可以通过encoding参数指定编码..."},
{"role": "user", "content": "给我一个完整的示例代码"}
]
}
关键参数详解：
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| model | string | 必填 | 要调用的模型名称 |
| messages | array | 必填 | 对话消息数组 |
| temperature | float | 1 | 控制随机性，0-2之间，越低越确定 |
| top_p | float | 1 | 核采样参数，与temperature二选一 |
| max_tokens | integer | 模型默认 | 最大生成token数 |
| stream | boolean | false | 是否流式输出 |
| stop | string/array | null | 停止生成的标记 |
| presence_penalty | float | 0 | 存在惩罚，减少重复话题 |
| frequency_penalty | float | 0 | 频率惩罚，减少重复用词 |
| tools | array | null | 可调用的工具列表 |
| tool_choice | string/object | auto | 工具调用策略 |
理解这些参数的含义和作用，能够帮助你精细控制模型的输出行为。例如，在代码生成场景中，应该使用较低的temperature（如0.1-0.3）以获得确定性的输出；在创意写作场景中，可以使用较高的temperature（如0.7-1.0）以获得更多样化的结果。
5 环境搭建与快速入门
5.1 获取API密钥
在开始编码之前，你需要获取一个API密钥。通过前文提到的聚合平台注册账号后，在控制台中即可生成API Key。API Key通常以sk-开头，是一串随机字符，请妥善保管。
获取到API Key后，你需要了解当前可用的模型列表。不同的模型在能力、速度和价格上存在显著差异。下表列出了常用的模型及其特点：
| 模型名称 | 类型 | 速度 | 质量 | 适用场景 |
|---|---|---|---|---|
| gemini-2.5-flash-lite | 免费 | 极快 | 较低 | 快速测试、简单任务 |
| gpt-5.4 | 付费 | 中等 | 优秀 | 复杂推理、高质量生成 |
| claude-haiku-4-5-20251001-thinking | 付费 | 中等 | 优秀 | 长文本分析、代码生成 |
需要特别说明的是，在免费模型中，gemini-2.5-flash-lite是速度最快的模型，响应延迟极低，适合快速测试和简单任务。但其生成质量相对有限，在复杂推理、长文本生成和代码编写等场景中表现不够理想。如果你的应用对输出质量有较高要求，建议使用付费模型或转向国产AI的API（如KIMI、GLM），后者在中文场景中表现优异且价格相对亲民。
5.2 Python环境配置
Python是构建智能体的首选语言，其丰富的生态系统和简洁的语法使得API调用和系统集成变得非常高效。本节将指导你完成Python开发环境的配置。
首先，确保你的系统安装了Python 3.9或更高版本。可以通过以下命令检查：
python --version
# 或
python3 --version
接下来，创建项目目录和虚拟环境：
mkdir my-agent
cd my-agent
python -m venv venv
# Linux/Mac
source venv/bin/activate
# Windows
venv\Scripts\activate
安装必要的依赖包：
pip install openai python-dotenv requests httpx
这些包的作用分别是：openai是OpenAI官方Python SDK，支持所有兼容OpenAI规范的API；python-dotenv用于管理环境变量；requests和httpx是HTTP客户端库，在需要直接发送HTTP请求时使用。
创建项目文件结构：
my-agent/
├── .env # 环境变量（API Key等）
├── .gitignore # Git忽略文件
├── requirements.txt # 依赖列表
├── agent.py # 智能体主程序
├── tools.py # 工具定义
├── memory.py # 记忆管理
└── config.py # 配置管理
.gitignore文件中至少应包含：
.env
venv/
__pycache__/
*.pyc
requirements.txt文件内容：
openai>=1.0.0
python-dotenv>=1.0.0
requests>=2.31.0
httpx>=0.25.0
5.3 第一个API调用
环境搭建完成后，让我们编写第一个API调用程序，体验大模型的对话能力。这是构建智能体的第一步，也是最基础的一步。
首先创建.env文件，填入你的API Key和Base URL：
API_KEY=sk-your-api-key-here
BASE_URL=https://api.aigc.bar/v1
然后创建hello_llm.py，编写第一个API调用：
import os
from dotenv import load_dotenv
from openai import OpenAI
# 加载环境变量
load_dotenv()
# 初始化客户端
client = OpenAI(
api_key="你的密钥",
base_url="https://api.aigc.bar/v1/"
)
# 发送请求
response = client.chat.completions.create(
model="gemini-2.5-flash-lite",
messages=[
{"role": "system", "content": "你是一个友好的AI助手。"},
{"role": "user", "content": "你好！请介绍一下你自己。"}
],
temperature=0.7,
max_tokens=512
)
# 输出结果
print("模型回复：", response.choices[0].message.content)
print(f"Token消耗：{response.usage.total_tokens}")
print(f"完成原因：{response.choices[0].finish_reason}")
运行这个脚本：
python hello_llm.py
如果一切配置正确，你将看到模型生成的回复。恭喜你，已经成功完成了第一次API调用！
让我们进一步理解这段代码的每个部分。OpenAI客户端的初始化需要两个关键参数：api_key用于身份认证，base_url指定API的服务端点。通过将base_url配置为环境变量，你可以轻松切换不同的API提供商，而无需修改业务代码。
chat.completions.create()方法是API调用的核心，它向服务器发送一个聊天补全请求。model参数指定使用哪个模型，messages参数提供完整的对话上下文，temperature和max_tokens控制生成行为。
响应对象包含多个字段，其中最常用的是choices[0].message.content（模型生成的文本）和usage（token消耗统计）。finish_reason字段告诉你生成结束的原因，正常结束为stop，达到token限制为length。
5.4 使用requests直接调用API
虽然OpenAI SDK提供了便捷的调用方式，但理解底层的HTTP请求对于调试和定制化开发非常重要。下面展示如何使用requests库直接发送HTTP请求调用API：
import os
import json
import requests
from dotenv import load_dotenv
# Load environment variables from .env file if it exists.
# This is good practice even if you're primarily using config.json,
# as it can provide fallback values or other necessary configurations.
load_dotenv()
# 从 config.json 读取配置
CONFIG_FILE = "config.json"
def load_config():
"""从配置文件加载 API 设置"""
if os.path.exists(CONFIG_FILE):
try:
with open(CONFIG_FILE, "r", encoding="utf-8") as f:
config = json.load(f)
# Ensure all required keys are present
api_key = config.get("api_key")
base_url = config.get("base_url")
model = config.get("model")
if not api_key or not base_url:
print(f"错误：配置文件 '{CONFIG_FILE}' 缺失 'api_key' 或 'base_url'。")
return None, None, None
# If model is not provided, use a default or return None
# For this example, we'll return it as is, or None if not found.
return api_key, base_url, model
except json.JSONDecodeError:
print(f"错误：无法解析配置文件 '{CONFIG_FILE}'。请确保它是有效的 JSON。")
return None, None, None
except Exception as e:
print(f"读取配置文件 '{CONFIG_FILE}' 时发生未知错误: {e}")
return None, None, None
else:
print(f"错误：配置文件 '{CONFIG_FILE}' 不存在。")
return None, None, None
api_key, base_url, model = load_config()
# Check if configuration was loaded successfully
if not api_key or not base_url:
# If config.json failed, you might want to fall back to environment variables
# or exit if configuration is mandatory.
# For this example, we'll exit if config.json is not usable.
print("请确保 config.json 文件存在且包含有效的 'api_key' 和 'base_url'。")
exit(1)
# If model was not found in config.json, you can set a default here
# For example:
# if not model:
# model = "gemini-2.5-flash-lite" # Or any other default model
# 构造请求
url = f"{base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model if model else "gemini-2.5-flash-lite", # Use loaded model or default
"messages": [
{"role": "system", "content": "你是一个专业的技术顾问。"},
{"role": "user", "content": "什么是REST API？"}
],
"temperature": 0.7,
"max_tokens": 512
}
# 发送请求
try:
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status() # Raise an exception for bad status codes (4xx or 5xx)
# 检查状态码
result = response.json()
print("模型回复：", result["choices"][0]["message"]["content"])
if "usage" in result and "total_tokens" in result["usage"]:
print(f"Token消耗：{result['usage']['total_tokens']}")
else:
print("Token消耗信息未在响应中找到。")
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
if hasattr(e, 'response') and e.response is not None:
print(f"状态码：{e.response.status_code}")
print(f"错误信息：{e.response.text}")
except json.JSONDecodeError:
print("错误：无法解析服务器返回的 JSON。")
except KeyError as e:
print(f"错误：响应中缺少预期的键 {e}。")
except Exception as e:
print(f"发生未知错误: {e}")
这种方式虽然代码量更多，但让你对API调用有了完全的控制。你可以自定义请求头、处理特殊的认证方式、添加超时设置和重试逻辑。在实际生产环境中，很多团队会选择在OpenAI SDK的基础上封装自己的API客户端，以实现更精细的控制。
6 Python实现智能体核心功能
6.1 单轮对话实现
单轮对话是最简单的API调用模式，每次请求都是独立的，不依赖之前的对话历史。虽然功能简单，但它是理解API调用机制的基础，也是很多无状态应用场景（如文本翻译、内容摘要）的主要使用方式。
下面是一个完整的单轮对话封装：
from openai import OpenAI
class SimpleChatAgent:
"""简单的单轮对话智能体"""
def __init__(self, model="gemini-2.5-flash-lite", system_prompt="你是一个有帮助的AI助手。"):
# 直接填入你的密钥和接口地址，不再使用 os.getenv()
self.client = OpenAI(
api_key="你的密钥",
base_url="https://api.aigc.bar/v1"
)
self.model = model
self.system_prompt = system_prompt
def chat(self, user_message, temperature=0.7, max_tokens=1024):
"""发送单轮对话请求"""
try:
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": user_message}
],
temperature=temperature,
max_tokens=max_tokens
)
return {
"content": response.choices[0].message.content,
"model": response.model,
"tokens": response.usage.total_tokens,
"finish_reason": response.choices[0].finish_reason
}
except Exception as e:
return {"error": str(e)}
def translate(self, text, source_lang="中文", target_lang="英文"):
"""翻译功能"""
prompt = f"请将以下{source_lang}文本翻译为{target_lang}，只输出翻译结果：\n\n{text}"
return self.chat(prompt, temperature=0.3)
def summarize(self, text):
"""摘要功能"""
prompt = f"请对以下文本进行简洁的摘要，提取核心要点：\n\n{text}"
return self.chat(prompt, temperature=0.5)
def code_review(self, code):
"""代码审查功能"""
prompt = f"请审查以下代码，指出潜在问题并给出改进建议：\n\n```\n{code}\n```"
return self.chat(prompt, temperature=0.3)
# 使用示例
if __name__ == "__main__":
agent = SimpleChatAgent()
# 基础对话
result = agent.chat("请解释什么是机器学习")
print("对话结果：", result["content"])
# 翻译
result = agent.translate("今天天气真好，适合出去散步。")
print("翻译结果：", result["content"])
# 摘要
long_text = """
人工智能（Artificial Intelligence，简称AI）是计算机科学的一个分支，
致力于创建能够执行通常需要人类智能的任务的系统。这些任务包括学习、
推理、问题解决、感知和语言理解等。AI的发展经历了多次浪潮，从早期的
符号主义到连接主义，再到如今的深度学习时代。特别是2012年以来，
深度学习在图像识别、语音识别和自然语言处理等领域取得了突破性进展，
推动了AI技术的广泛应用。
"""
result = agent.summarize(long_text)
print("摘要结果：", result["content"])
这个SimpleChatAgent类封装了单轮对话的核心逻辑，并提供了翻译、摘要和代码审查等常用功能的快捷方法。通过修改system_prompt，你可以快速定制智能体的角色和行为。
6.2 多轮对话与上下文管理
多轮对话是智能体的基本能力，它要求系统能够维护完整的对话历史，并在每次请求时将历史消息一并发送给模型。这样模型才能理解对话的上下文，生成连贯的回复。
from openai import OpenAI
from typing import List, Dict
class ConversationAgent:
"""多轮对话智能体"""
def __init__(self, model="gemini-2.5-flash-lite", system_prompt="你是一个有帮助的AI助手。"):
# 直接填入你的密钥和接口地址，删除所有 os/dotenv 相关
self.client = OpenAI(
api_key="你的密钥",
base_url="https://api.aigc.bar/v1"
)
self.model = model
self.system_prompt = system_prompt
self.history: List[Dict] = []
self.max_history = 20 # 最多保留20轮对话
def chat(self, user_message: str, temperature=0.7, max_tokens=1024) -> str:
"""发送多轮对话请求"""
# 构造消息列表
messages = [{"role": "system", "content": self.system_prompt}]
messages.extend(self.history)
messages.append({"role": "user", "content": user_message})
try:
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens
)
assistant_message = response.choices[0].message.content
# 更新对话历史
self.history.append({"role": "user", "content": user_message})
self.history.append({"role": "assistant", "content": assistant_message})
# 管理历史长度
self._trim_history()
return assistant_message
except Exception as e:
return f"错误：{str(e)}"
def _trim_history(self):
"""修剪对话历史，保留最近的对话轮次"""
if len(self.history) > self.max_history * 2:
# 保留最近的消息
self.history = self.history[-(self.max_history * 2):]
def clear_history(self):
"""清空对话历史"""
self.history = []
def get_history_summary(self) -> str:
"""获取对话历史摘要"""
summary_lines = []
for msg in self.history:
role = "用户" if msg["role"] == "user" else "助手"
content = msg["content"][:50] + "..." if len(msg["content"]) > 50 else msg["content"]
summary_lines.append(f"{role}: {content}")
return "\n".join(summary_lines)
def run_interactive(self):
"""运行交互式对话"""
print("=" * 50)
print("智能体对话系统（输入 'quit' 退出，'clear' 清空历史）")
print("=" * 50)
while True:
user_input = input("\n你: ").strip()
if user_input.lower() == 'quit':
print("再见！")
break
elif user_input.lower() == 'clear':
self.clear_history()
print("对话历史已清空。")
continue
elif not user_input:
continue
response = self.chat(user_input)
print(f"\n助手: {response}")
# 使用示例
if __name__ == "__main__":
agent = ConversationAgent(
model="gemini-2.5-flash-lite",
system_prompt="你是一个专业的Python编程助手，擅长解答编程问题和提供代码示例。"
)
agent.run_interactive()
多轮对话的关键在于对话历史的管理。上面的实现使用一个列表来存储所有历史消息，每次请求时将完整历史发送给模型。_trim_history()方法确保历史不会无限增长，当超过设定的最大轮次时，自动丢弃最早的消息。这种滑动窗口策略简单有效，但可能会丢失重要的早期上下文。
对于更复杂的场景，我们可以实现基于摘要的历史管理策略：
from openai import OpenAI
from typing import List, Dict
# 基础多轮对话类（已内置密钥）
class ConversationAgent:
"""多轮对话智能体"""
def __init__(self, model="gemini-2.5-flash-lite", system_prompt="你是一个有帮助的AI助手。"):
self.client = OpenAI(
api_key="你的密钥",
base_url="https://api.aigc.bar/v1"
)
self.model = model
self.system_prompt = system_prompt
self.history: List[Dict] = []
self.max_history = 20 # 最多保留20轮对话
def chat(self, user_message: str, temperature=0.7, max_tokens=1024) -> str:
messages = [{"role": "system", "content": self.system_prompt}]
messages.extend(self.history)
messages.append({"role": "user", "content": user_message})
try:
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens
)
assistant_message = response.choices[0].message.content
self.history.append({"role": "user", "content": user_message})
self.history.append({"role": "assistant", "content": assistant_message})
self._trim_history()
return assistant_message
except Exception as e:
return f"错误：{str(e)}"
def _trim_history(self):
if len(self.history) > self.max_history * 2:
self.history = self.history[-(self.max_history * 2):]
def clear_history(self):
self.history = []
def get_history_summary(self) -> str:
summary_lines = []
for msg in self.history:
role = "用户" if msg["role"] == "user" else "助手"
content = msg["content"][:50] + "..." if len(msg["content"]) > 50 else msg["content"]
summary_lines.append(f"{role}: {content}")
return "\n".join(summary_lines)
def run_interactive(self):
print("=" * 50)
print("智能体对话系统（输入 'quit' 退出，'clear' 清空历史）")
print("=" * 50)
while True:
user_input = input("\n你: ").strip()
if user_input.lower() == 'quit':
print("再见！")
break
elif user_input.lower() == 'clear':
self.clear_history()
print("对话历史已清空。")
continue
elif not user_input:
continue
response = self.chat(user_input)
print(f"\n助手: {response}")
# 带自动摘要压缩的高级智能对话类（已修复异常处理）
class SmartConversationAgent(ConversationAgent):
"""带摘要压缩的智能对话智能体"""
def __init__(self, model="gemini-2.5-flash-lite", system_prompt="你是一个有帮助的AI助手。"):
super().__init__(model, system_prompt)
self.summary = "" # 早期对话的摘要
def chat(self, user_message: str, temperature=0.7, max_tokens=1024) -> str:
messages = [{"role": "system", "content": self.system_prompt}]
# 如果有摘要，加入上下文
if self.summary:
messages.append({
"role": "system",
"content": f"以下是之前对话的摘要：\n{self.summary}"
})
messages.extend(self.history)
messages.append({"role": "user", "content": user_message})
try:
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens
)
assistant_message = response.choices[0].message.content
self.history.append({"role": "user", "content": user_message})
self.history.append({"role": "assistant", "content": assistant_message})
# 当历史过长时，压缩早期对话
if len(self.history) > 10 * 2:
self._compress_history()
return assistant_message
except Exception as e:
return f"错误：{str(e)}"
def _compress_history(self):
"""将早期对话压缩为摘要"""
old_messages = self.history[:6] # 取前3轮
conversation_text = "\n".join([
f"{'用户' if m['role'] == 'user' else '助手'}: {m['content']}"
for m in old_messages
])
try:
summary_prompt = f"请用简洁的语言总结以下对话的核心内容：\n\n{conversation_text}"
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": summary_prompt}],
temperature=0.3,
max_tokens=256
)
new_summary = response.choices[0].message.content
self.summary += "\n" + new_summary if self.summary else new_summary
self.history = self.history[6:] # 保留剩余的历史
except Exception as e:
print(f"压缩失败：{str(e)}")
# 使用示例
if __name__ == "__main__":
agent = SmartConversationAgent(
model="gemini-2.5-flash-lite",
system_prompt="你是一个专业的Python编程助手，擅长解答编程问题和提供代码示例。"
)
agent.run_interactive()
摘要压缩策略的核心思想是：当对话历史超过一定长度时，使用LLM将早期对话压缩为摘要，然后只保留摘要和最近的对话历史。这样既保留了关键信息，又控制了token消耗。不过需要注意的是，摘要压缩本身也会消耗API调用次数和token，因此需要合理设置压缩的触发阈值。
6.3 流式输出实现
流式输出（Streaming）是提升用户体验的重要技术。在非流式模式下，用户需要等待模型生成完整回复后才能看到任何内容，对于长文本生成可能需要等待数十秒。而在流式模式下，模型的输出会逐token实时返回，用户可以立即看到生成过程，大大降低了感知延迟。
from openai import OpenAI
class StreamingAgent:
"""支持流式输出的智能体"""
def __init__(self, model="gemini-2.5-flash-lite", system_prompt="你是一个有帮助的AI助手。"):
# 直接填入你的密钥和接口地址
self.client = OpenAI(
api_key="sk-",
base_url="https://api.aigc.bar/v1"
)
self.model = model
self.system_prompt = system_prompt
self.history = []
def chat_stream(self, user_message: str, temperature=0.7, max_tokens=2048):
"""流式输出对话"""
messages = [{"role": "system", "content": self.system_prompt}]
messages.extend(self.history)
messages.append({"role": "user", "content": user_message})
try:
stream = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
stream=True # 开启流式输出
)
full_response = ""
print("助手: ", end="", flush=True)
for chunk in stream:
if chunk.choices[0].delta.content is not None:
content = chunk.choices[0].delta.content
print(content, end="", flush=True)
full_response += content
print() # 换行
# 更新历史
self.history.append({"role": "user", "content": user_message})
self.history.append({"role": "assistant", "content": full_response})
return full_response
except Exception as e:
print(f"\n错误：{str(e)}")
return None
def chat_stream_with_usage(self, user_message: str, temperature=0.7, max_tokens=2048):
"""带token统计的流式输出"""
messages = [{"role": "system", "content": self.system_prompt}]
messages.extend(self.history)
messages.append({"role": "user", "content": user_message})
try:
stream = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
stream=True
)
full_response = ""
print("助手: ", end="", flush=True)
for chunk in stream:
# 处理内容
if chunk.choices[0].delta.content is not None:
content = chunk.choices[0].delta.content
print(content, end="", flush=True)
full_response += content
# 检查是否有usage信息（部分API在最后一个chunk中返回）
if hasattr(chunk, 'usage') and chunk.usage is not None:
print(f"\n[Token消耗: {chunk.usage.total_tokens}]", flush=True)
print()
self.history.append({"role": "user", "content": user_message})
self.history.append({"role": "assistant", "content": full_response})
return full_response
except Exception as e:
print(f"\n错误：{str(e)}")
return None
# 使用示例
if __name__ == "__main__":
agent = StreamingAgent(
system_prompt="你是一个技术专家，擅长详细解释技术概念。"
)
print("=" * 50)
print("流式输出智能体（输入 'quit' 退出）")
print("=" * 50)
while True:
user_input = input("\n你: ").strip()
if user_input.lower() == 'quit':
break
if not user_input:
continue
agent.chat_stream(user_input)
流式输出的核心区别在于stream=True参数。开启流式后，API返回的不是单个响应对象，而是一个事件流（Server-Sent Events, SSE）。每个事件包含一个chunk对象，其中delta字段包含本次增量输出的内容。通过遍历事件流并逐个处理delta.content，就能实现逐字输出的效果。
6.4 错误处理与重试机制
在生产环境中，API调用不可避免地会遇到各种错误，包括网络超时、速率限制、服务器故障等。一个健壮的智能体系统必须具备完善的错误处理和重试机制。
import time
import logging
from openai import OpenAI
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("Agent")
class RobustAgent:
"""具备完善错误处理和重试机制的智能体"""
MAX_RETRIES = 5
RETRY_DELAYS = [1, 2, 4, 8, 16] # 指数退避延迟（秒）
def __init__(self, model="gemini-2.5-flash-lite", system_prompt="你是一个有帮助的AI助手。"):
# 直接填入你的密钥和接口地址，删除所有环境变量
self.client = OpenAI(
api_key="sk-",
base_url="https://api.aigc.bar/v1",
timeout=60.0, # 设置超时时间
max_retries=2 # SDK内置重试次数
)
self.model = model
self.system_prompt = system_prompt
self.history = []
def chat(self, user_message: str, temperature=0.7, max_tokens=1024) -> dict:
"""带重试机制的对话方法"""
messages = [{"role": "system", "content": self.system_prompt}]
messages.extend(self.history)
messages.append({"role": "user", "content": user_message})
last_error = None
for attempt in range(self.MAX_RETRIES):
try:
logger.info(f"第 {attempt + 1} 次请求...")
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens
)
result = {
"content": response.choices[0].message.content,
"model": response.model,
"tokens": response.usage.total_tokens,
"finish_reason": response.choices[0].finish_reason,
"attempts": attempt + 1
}
# 成功则更新历史
self.history.append({"role": "user", "content": user_message})
self.history.append({"role": "assistant", "content": result["content"]})
return result
except Exception as e:
last_error = e
error_type = type(e).__name__
logger.warning(f"请求失败 ({error_type}): {str(e)}")
# 判断是否需要重试
if self._is_retryable(e):
delay = self.RETRY_DELAYS[min(attempt, len(self.RETRY_DELAYS) - 1)]
logger.info(f"等待 {delay} 秒后重试...")
time.sleep(delay)
else:
logger.error(f"不可重试的错误，终止请求")
break
return {
"error": str(last_error),
"attempts": attempt + 1
}
def _is_retryable(self, error) -> bool:
"""判断错误是否可以重试"""
error_str = str(error).lower()
# 速率限制、服务器错误、超时等可以重试
retryable_keywords = [
"rate limit", "429", "500", "502", "503",
"timeout", "connection", "overloaded"
]
return any(keyword in error_str for keyword in retryable_keywords)
def chat_with_fallback(self, user_message: str, fallback_models=None, **kwargs):
"""带模型降级的对话方法"""
if fallback_models is None:
fallback_models = ["gpt-4o-mini", "gemini-2.5-flash-lite"]
# 尝试主模型
result = self.chat(user_message, **kwargs)
if "error" not in result:
return result
# 降级到备用模型
for fallback_model in fallback_models:
logger.info(f"降级到模型: {fallback_model}")
original_model = self.model
self.model = fallback_model
result = self.chat(user_message, **kwargs)
self.model = original_model
if "error" not in result:
result["fallback"] = True
result["original_model"] = original_model
return result
return {"error": "所有模型均不可用", "attempts": 0}
# 使用示例
if __name__ == "__main__":
agent = RobustAgent()
# 基础对话测试
print("\n=== 开始对话 ===")
res = agent.chat("你好，介绍一下你自己")
if "error" not in res:
print("回复：", res["content"])
print(f"Token消耗：{res['tokens']}")
else:
print("错误：", res["error"])
上面的RobustAgent实现了两层容错机制：指数退避重试和模型降级。指数退避重试在遇到可重试错误时，按照1秒、2秒、4秒、8秒、16秒的间隔进行重试，避免在服务端过载时雪崩式请求。模型降级则在主模型完全不可用时，自动切换到备用模型，确保服务可用性。
7 工具调用与Function Calling
7.1 Function Calling原理
Function Calling（函数调用）是LLM智能体的核心能力之一，它允许模型在生成文本回复之外，主动请求调用预定义的外部函数。这一能力是智能体从"只能聊天"进化为"能做事"的关键突破。
Function Calling的工作原理可以概括为以下流程：首先，开发者在API请求中定义一组可用工具（tools），每个工具包含名称、描述和参数schema；然后，模型根据用户意图和工具描述，决定是否需要调用工具以及传递什么参数；如果模型决定调用工具，API响应中会包含工具调用信息（tool_calls），开发者执行对应的函数并将结果返回给模型；最后，模型基于工具返回的结果生成最终回复。
需要特别强调的是，LLM本身并不直接执行任何工具函数。它只是根据工具的描述和用户的意图，生成结构化的工具调用请求（包括函数名和参数）。实际的函数执行由开发者的代码完成，执行结果再通过API返回给模型。这种设计既保证了安全性（模型无法直接执行危险操作），又提供了灵活性（开发者可以完全控制工具的实现）。
Function Calling的底层技术是模型经过特殊训练，学会了在适当的时候输出特定格式的JSON来表示工具调用意图，而不是普通的文本。当模型判断需要调用工具时，finish_reason会变为tool_calls，响应中的message对象会包含tool_calls字段而非content字段。
7.2 定义工具函数
工具的定义遵循JSON Schema规范，包含三个核心要素：工具名称（name）、工具描述（description）和参数模式（parameters）。工具描述的质量直接影响模型选择工具的准确性，因此需要清晰、准确地描述工具的功能和适用场景。
# 工具定义示例
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的当前天气信息，包括温度、湿度、天气状况等",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，如'北京'、'上海'"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位，默认为摄氏度"
}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "search_web",
"description": "在互联网上搜索信息，返回相关的搜索结果",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
},
"num_results": {
"type": "integer",
"description": "返回结果数量，默认5条",
"default": 5
}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "calculate",
"description": "执行数学计算，支持基本的四则运算和常用数学函数",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式，如'2+3*4'、'sqrt(16)'"
}
},
"required": ["expression"]
}
}
},
{
"type": "function",
"function": {
"name": "get_current_time",
"description": "获取当前的日期和时间",
"parameters": {
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "时区，如'Asia/Shanghai'、'US/Eastern'",
"default": "Asia/Shanghai"
}
},
"required": []
}
}
}
]
工具定义的最佳实践包括：第一，名称使用小写字母和下划线，简洁且语义明确；第二，描述要详细说明工具的功能、适用场景和注意事项，这是模型判断是否调用该工具的主要依据；第三，参数的类型和约束要尽可能精确，使用enum限制可选值，使用default设置默认值；第四，required字段准确标注必填参数，避免模型遗漏关键参数。
7.3 实现工具调用循环
工具调用循环是智能体的核心运行机制。当模型请求调用工具时，我们需要执行对应的函数，将结果返回给模型，然后让模型继续推理。这个过程可能需要多轮迭代，直到模型认为已经收集到足够信息可以给出最终回复。
import json
from datetime import datetime
from openai import OpenAI
class ToolCallingAgent:
"""支持工具调用的智能体"""
def __init__(self, model="gemini-2.5-flash-lite", system_prompt="你是一个有帮助的AI助手，可以使用工具来帮助用户。"):
# 直接填入你的密钥和接口地址
self.client = OpenAI(
api_key="sk-",
base_url="https://api.aigc.bar/v1"
)
self.model = model
self.system_prompt = system_prompt
self.tools = []
self.tool_functions = {}
self.history = []
self.max_iterations = 10 # 最大工具调用轮次
def register_tool(self, tool_definition, tool_function):
"""注册一个工具"""
self.tools.append(tool_definition)
func_name = tool_definition["function"]["name"]
self.tool_functions[func_name] = tool_function
def _execute_tool(self, tool_call):
"""执行工具调用"""
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
print(f" [工具调用] {func_name}({func_args})")
if func_name in self.tool_functions:
try:
result = self.tool_functions[func_name](**func_args)
print(f" [工具结果] {result}")
return json.dumps({"result": result}, ensure_ascii=False)
except Exception as e:
error_msg = f"工具执行错误: {str(e)}"
print(f" [工具错误] {error_msg}")
return json.dumps({"error": error_msg}, ensure_ascii=False)
else:
return json.dumps({"error": f"未知工具: {func_name}"}, ensure_ascii=False)
def chat(self, user_message: str, temperature=0.7, max_tokens=2048) -> str:
"""带工具调用的对话"""
messages = [{"role": "system", "content": self.system_prompt}]
messages.extend(self.history)
messages.append({"role": "user", "content": user_message})
iteration = 0
while iteration < self.max_iterations:
iteration += 1
print(f"\n--- 迭代 {iteration} ---")
# 调用API
kwargs = {
"model": self.model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
}
if self.tools:
kwargs["tools"] = self.tools
kwargs["tool_choice"] = "auto"
response = self.client.chat.completions.create(**kwargs)
message = response.choices[0].message
# 将助手消息加入历史
messages.append(message)
# 检查是否有工具调用
if message.tool_calls:
print(f" 模型请求调用 {len(message.tool_calls)} 个工具")
# 执行每个工具调用
for tool_call in message.tool_calls:
result = self._execute_tool(tool_call)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
# 继续循环，让模型处理工具结果
continue
# 没有工具调用，返回最终回复
final_response = message.content
# 更新对话历史（简化版，不包含工具调用的中间步骤）
self.history.append({"role": "user", "content": user_message})
self.history.append({"role": "assistant", "content": final_response})
return final_response
return "抱歉，工具调用轮次已达上限，无法完成请求。"
# ===== 工具函数实现 =====
def get_weather(city: str, unit: str = "celsius") -> str:
"""获取天气信息（模拟实现）"""
weather_data = {
"北京": {"temp": 22, "humidity": 45, "condition": "晴"},
"上海": {"temp": 26, "humidity": 72, "condition": "多云"},
"广州": {"temp": 30, "humidity": 85, "condition": "阵雨"},
"深圳": {"temp": 29, "humidity": 80, "condition": "多云"},
}
if city in weather_data:
data = weather_data[city]
if unit == "fahrenheit":
data["temp"] = data["temp"] * 9 / 5 + 32
return f"{city}当前天气：{data['condition']}，温度{data['temp']}°{'F' if unit == 'fahrenheit' else 'C'}，湿度{data['humidity']}%"
return f"未找到{city}的天气数据"
def search_web(query: str, num_results: int = 5) -> str:
"""搜索网页（模拟实现）"""
return f"搜索'{query}'返回了{num_results}条结果：1. 相关信息A 2. 相关信息B 3. 相关信息C"
def calculate(expression: str) -> str:
"""执行数学计算"""
allowed_names = {
"abs": abs, "round": round, "min": min, "max": max,
"pow": pow, "sum": sum, "len": len
}
try:
import math
for name in dir(math):
if not name.startswith("_"):
allowed_names[name] = getattr(math, name)
result = eval(expression, {"__builtins__": {}}, allowed_names)
return f"{expression} = {result}"
except Exception as e:
return f"计算错误: {str(e)}"
def get_current_time(timezone: str = "Asia/Shanghai") -> str:
"""获取当前时间"""
try:
import pytz
tz = pytz.timezone(timezone)
now = datetime.now(tz)
return now.strftime(f"%Y年%m月%d日 %H:%M:%S ({timezone})")
except:
return datetime.now().strftime("%Y年%m月%d日 %H:%M:%S")
# ===== 工具定义 =====
weather_tool = {
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的当前天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"], "description": "温度单位"}
},
"required": ["city"]
}
}
}
search_tool = {
"type": "function",
"function": {
"name": "search_web",
"description": "在互联网上搜索信息",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"},
"num_results": {"type": "integer", "description": "返回结果数量"}
},
"required": ["query"]
}
}
}
calculate_tool = {
"type": "function",
"function": {
"name": "calculate",
"description": "执行数学计算",
"parameters": {
"type": "object",
"properties": {
"expression": {"type": "string", "description": "数学表达式"}
},
"required": ["expression"]
}
}
}
time_tool = {
"type": "function",
"function": {
"name": "get_current_time",
"description": "获取当前的日期和时间",
"parameters": {
"type": "object",
"properties": {
"timezone": {"type": "string", "description": "时区"}
},
"required": []
}
}
}
# ===== 使用示例 =====
if __name__ == "__main__":
agent = ToolCallingAgent(
model="gemini-2.5-flash-lite",
system_prompt="你是一个智能助手，可以使用工具来获取实时信息。请根据用户的问题选择合适的工具。"
)
# 注册工具
agent.register_tool(weather_tool, get_weather)
agent.register_tool(search_tool, search_web)
agent.register_tool(calculate_tool, calculate)
agent.register_tool(time_tool, get_current_time)
# 测试工具调用
print(agent.chat("北京今天天气怎么样？"))
print("\n" + "=" * 50)
print(agent.chat("现在几点了？"))
print("\n" + "=" * 50)
print(agent.chat("帮我算一下 (235 + 678) * 12 的结果"))
上面的代码实现了一个完整的工具调用智能体。核心逻辑在chat方法中：每次API调用后，检查响应中是否包含tool_calls；如果有，执行对应的工具函数，将结果以tool角色的消息返回给模型；然后继续调用API，让模型基于工具结果进行下一轮推理。这个循环会持续进行，直到模型不再请求调用工具，直接生成文本回复。
7.4 常用工具示例
在实际应用中，智能体需要配备各种类型的工具来满足不同的任务需求。下面介绍几类常用的工具及其实现思路。
数据库查询工具：允许智能体查询结构化数据，是构建数据驱动型智能体的核心工具。实现时需要注意SQL注入风险，建议使用参数化查询而非字符串拼接。
import sqlite3
def query_database(sql: str, db_path: str = "data.db") -> str:
"""查询SQLite数据库"""
try:
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute(sql)
results = cursor.fetchall()
columns = [desc[0] for desc in cursor.description]
conn.close()
# 格式化为表格
output = " | ".join(columns) + "\n"
output += "-" * len(output) + "\n"
for row in results[:20]: # 限制返回行数
output += " | ".join(str(v) for v in row) + "\n"
return output if results else "查询结果为空"
except Exception as e:
return f"数据库查询错误: {str(e)}"
db_tool = {
"type": "function",
"function": {
"name": "query_database",
"description": "查询SQLite数据库，执行SQL查询语句并返回结果",
"parameters": {
"type": "object",
"properties": {
"sql": {"type": "string", "description": "SQL查询语句（仅支持SELECT）"},
"db_path": {"type": "string", "description": "数据库文件路径"}
},
"required": ["sql"]
}
}
}
文件操作工具：允许智能体读写文件，适用于文档处理、代码生成等场景。
def read_file(file_path: str) -> str:
"""读取文件内容"""
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
return content[:5000] if len(content) > 5000 else content # 限制长度
except Exception as e:
return f"读取文件错误: {str(e)}"
def write_file(file_path: str, content: str) -> str:
"""写入文件"""
try:
with open(file_path, 'w', encoding='utf-8') as f:
f.write(content)
return f"成功写入文件: {file_path}"
except Exception as e:
return f"写入文件错误: {str(e)}"
file_tools = [
{
"type": "function",
"function": {
"name": "read_file",
"description": "读取指定路径的文件内容",
"parameters": {
"type": "object",
"properties": {
"file_path": {"type": "string", "description": "文件路径"}
},
"required": ["file_path"]
}
}
},
{
"type": "function",
"function": {
"name": "write_file",
"description": "将内容写入指定路径的文件",
"parameters": {
"type": "object",
"properties": {
"file_path": {"type": "string", "description": "文件路径"},
"content": {"type": "string", "description": "要写入的内容"}
},
"required": ["file_path", "content"]
}
}
}
]
HTTP请求工具：允许智能体访问外部API，获取实时数据。
import requests
def http_get(url: str, headers: dict = None) -> str:
"""发送HTTP GET请求"""
try:
response = requests.get(url, headers=headers, timeout=10)
return response.text[:3000] # 限制返回长度
except Exception as e:
return f"HTTP请求错误: {str(e)}"
http_tool = {
"type": "function",
"function": {
"name": "http_get",
"description": "发送HTTP GET请求获取网页或API数据",
"parameters": {
"type": "object",
"properties": {
"url": {"type": "string", "description": "请求的URL地址"},
"headers": {"type": "object", "description": "请求头"}
},
"required": ["url"]
}
}
}
下面的表格总结了智能体常用的工具类型：
| 工具类型 | 典型功能 | 实现复杂度 | 安全风险 | 应用场景 |
|---|---|---|---|---|
| 信息查询 | 天气、搜索、新闻 | 低 | 低 | 问答、信息整合 |
| 数据库操作 | SQL查询、数据统计 | 中 | 高（SQL注入） | 数据分析、报表 |
| 文件操作 | 读写文件、目录管理 | 低 | 中（路径遍历） | 文档处理、代码生成 |
| HTTP请求 | API调用、网页抓取 | 中 | 中（SSRF） | 数据获取、服务集成 |
| 代码执行 | Python/Shell执行 | 高 | 高（任意代码） | 数据处理、自动化 |
| 邮件通知 | 发送邮件、消息推送 | 低 | 低 | 工作流自动化 |
| 日程管理 | 创建/查询日程 | 中 | 低 | 个人助手 |
8 RAG检索增强生成智能体
8.1 RAG原理与架构
RAG（Retrieval-Augmented Generation，检索增强生成）是解决大语言模型知识时效性和幻觉问题的重要技术方案。LLM的知识来源于训练数据，存在两个根本性局限：一是知识截止日期问题，模型无法获取训练之后的新信息；二是幻觉问题，模型可能生成看似合理但实际错误的信息。RAG通过在生成前先检索相关文档，将外部知识注入模型的上下文，从而有效缓解这两个问题。
RAG系统的工作流程可以分为三个阶段：索引阶段、检索阶段和生成阶段。索引阶段将文档集合处理成可检索的形式，包括文档切分、向量化（Embedding）和索引构建。检索阶段根据用户查询，从索引中找到最相关的文档片段。生成阶段将检索到的文档片段与用户查询一起输入LLM，让模型基于检索到的上下文生成回答。
RAG的核心优势在于它不需要重新训练或微调模型，就能让模型获取特定领域的知识。这意味着你可以快速构建一个"了解"你公司文档、产品手册或技术规范的AI助手，而无需承担微调模型的高昂成本和技术门槛。此外，RAG的另一个重要优势是可溯源——每个回答都可以追溯到具体的文档来源，用户可以验证信息的准确性，这对于企业级应用至关重要。
8.2 文档处理与向量化
文档处理是RAG系统的第一步，其质量直接影响后续检索和生成的效果。文档处理主要包括文档加载、文本切分和向量化三个步骤。
文档加载：不同格式的文档需要使用不同的加载器。PDF文档可以使用PyPDF2或pdfplumber，Word文档使用python-docx，网页使用BeautifulSoup，Markdown使用正则表达式或专用解析器。加载后的原始文本通常包含大量噪音（如页眉页脚、导航链接、特殊字符等），需要进行清洗。
文本切分：切分策略对RAG的效果影响极大。切分粒度过粗，单个片段包含过多信息，检索时噪声大；切分粒度过细，单个片段信息不完整，模型难以理解上下文。常见的切分策略包括固定长度切分（每段固定token数，带重叠）、按段落切分（以段落标记为分隔符）、按语义切分（使用模型判断语义边界）和递归切分（按多种分隔符递归切分）。
from typing import List
class TextSplitter:
"""文本切分器"""
def __init__(self, chunk_size=500, chunk_overlap=50):
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
def split_text(self, text: str) -> List[str]:
"""将文本切分为固定长度的片段"""
chunks = []
start = 0
while start < len(text):
end = start + self.chunk_size
chunk = text[start:end]
# 尝试在句子边界处切分
if end < len(text):
# 找到最近的句号、问号或感叹号
for sep in ['。', '！', '？', '.', '!', '?', '\n']:
last_sep = chunk.rfind(sep)
if last_sep > self.chunk_size * 0.5:
chunk = chunk[:last_sep + 1]
end = start + last_sep + 1
break
chunks.append(chunk.strip())
start = end - self.chunk_overlap # 重叠部分
return [c for c in chunks if c] # 过滤空片段
def split_by_paragraphs(self, text: str) -> List[str]:
"""按段落切分"""
paragraphs = text.split('\n\n')
chunks = []
current_chunk = ""
for para in paragraphs:
if len(current_chunk) + len(para) > self.chunk_size and current_chunk:
chunks.append(current_chunk.strip())
current_chunk = para
else:
current_chunk += "\n\n" + para if current_chunk else para
if current_chunk.strip():
chunks.append(current_chunk.strip())
return chunks
向量化（Embedding）：文本向量化是将文本转换为高维数值向量的过程，使得语义相似的文本在向量空间中距离相近。向量化通过Embedding模型实现，常用的Embedding模型包括OpenAI的text-embedding-3-small/large、BGE系列模型、M3E模型等。选择Embedding模型时需要考虑维度大小、多语言支持、推理速度和成本等因素。
import os
import json
import requests
from typing import List
class EmbeddingService:
"""文本向量化服务"""
def __init__(self, model="text-embedding-3-small"):
self.api_key = os.getenv("API_KEY")
self.base_url = os.getenv("BASE_URL")
self.model = model
def embed(self, text: str) -> List[float]:
"""将单个文本转换为向量"""
response = requests.post(
f"{self.base_url}/embeddings",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": self.model,
"input": text
}
)
if response.status_code == 200:
return response.json()["data"][0]["embedding"]
else:
raise Exception(f"Embedding请求失败: {response.status_code} {response.text}")
def embed_batch(self, texts: List[str]) -> List[List[float]]:
"""批量向量化"""
response = requests.post(
f"{self.base_url}/embeddings",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": self.model,
"input": texts
}
)
if response.status_code == 200:
data = response.json()["data"]
# 按index排序确保顺序正确
data.sort(key=lambda x: x["index"])
return [item["embedding"] for item in data]
else:
raise Exception(f"批量Embedding请求失败: {response.status_code}")
8.3 向量数据库
向量数据库是RAG系统的核心存储组件，它支持高效的向量相似度搜索。当用户提出查询时，查询文本被转换为向量，然后在向量数据库中搜索与之最相似的文档向量，返回对应的文档片段。
常见的向量数据库对比如下：
| 向量数据库 | 类型 | 特点 | 适用场景 |
|---|---|---|---|
| ChromaDB | 嵌入式 | 轻量级，Python原生，零配置 | 原型开发、小规模应用 |
| FAISS | 库 | Meta开源，性能极高，纯内存 | 大规模向量搜索 |
| Pinecone | 云服务 | 全托管，自动扩展 | 生产环境、免运维 |
| Milvus | 分布式 | 高性能，支持分布式部署 | 企业级大规模应用 |
| Qdrant | 独立服务 | Rust编写，性能好，过滤灵活 | 中等规模生产应用 |
| Weaviate | 独立服务 | 支持混合搜索，GraphQL API | 需要混合搜索的场景 |
对于学习和原型开发，推荐使用ChromaDB，它无需额外安装服务，纯Python即可运行。下面展示如何使用ChromaDB构建向量存储：
import chromadb
from chromadb.config import Settings
class VectorStore:
"""基于ChromaDB的向量存储"""
def __init__(self, collection_name="documents", persist_directory="./chroma_db"):
self.client = chromadb.Client(Settings(
chroma_db_impl="duckdb+parquet",
persist_directory=persist_directory
))
self.collection = self.client.get_or_create_collection(
name=collection_name,
metadata={"hnsw:space": "cosine"} # 使用余弦相似度
)
self.embedding_service = EmbeddingService()
def add_documents(self, texts: list, metadatas: list = None, ids: list = None):
"""添加文档到向量存储"""
if ids is None:
ids = [f"doc_{i}" for i in range(len(texts))]
if metadatas is None:
metadatas = [{"source": "unknown"} for _ in range(texts)]
# 批量向量化
embeddings = self.embedding_service.embed_batch(texts)
# 添加到ChromaDB
self.collection.add(
embeddings=embeddings,
documents=texts,
metadatas=metadatas,
ids=ids
)
def search(self, query: str, top_k: int = 5) -> list:
"""搜索相关文档"""
query_embedding = self.embedding_service.embed(query)
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=top_k
)
return [
{
"text": doc,
"metadata": meta,
"distance": dist
}
for doc, meta, dist in zip(
results["documents"][0],
results["metadatas"][0],
results["distances"][0]
)
]
8.4 完整RAG智能体实现
将前面的各个组件整合起来，我们就可以构建一个完整的RAG智能体。这个智能体能够根据用户的问题，自动检索相关文档，然后基于检索到的上下文生成准确的回答。
from openai import OpenAI
import numpy as np
# ====================== 内置缺失的依赖类（直接集成）======================
class TextSplitter:
"""简单文本分块器"""
def __init__(self, chunk_size=500, chunk_overlap=50):
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
def split_text(self, text):
chunks = []
start = 0
text_len = len(text)
while start < text_len:
end = min(start + self.chunk_size, text_len)
chunks.append(text[start:end].strip())
start = end - self.chunk_overlap
return chunks
class VectorStore:
"""简易向量库（模拟）"""
def __init__(self, collection_name="knowledge_base"):
self.collection_name = collection_name
self.documents = {}
def add_documents(self, texts, metadatas=None, ids=None):
for i, text in enumerate(texts):
doc_id = ids[i] if ids else f"doc_{i}"
self.documents[doc_id] = {
"text": text,
"metadata": metadatas[i] if metadatas else {}
}
def search(self, query, top_k=5):
"""简易相似度搜索"""
results = []
query_words = set(query.lower().split())
for doc_id, data in self.documents.items():
text = data["text"].lower()
score = sum(1 for w in query_words if w in text)
results.append({
"text": data["text"],
"metadata": data["metadata"],
"distance": 1 - (score / (len(query_words) or 1))
})
results = sorted(results, key=lambda x: x["distance"])[:top_k]
return results
# ====================== RAG 主类（已填密钥）======================
class RAGAgent:
"""RAG检索增强生成智能体"""
def __init__(self, model="gemini-2.5-flash-lite", collection_name="knowledge_base"):
# ✅ 直接填入你的密钥和地址
self.client = OpenAI(
api_key="sk-",
base_url="https://api.aigc.bar/v1"
)
self.model = model
self.vector_store = VectorStore(collection_name=collection_name)
self.splitter = TextSplitter(chunk_size=500, chunk_overlap=50)
self.conversation_history = []
def ingest_documents(self, documents: list, source_name: str = "uploaded"):
"""导入文档到知识库"""
all_chunks = []
all_metadatas = []
all_ids = []
for i, doc_text in enumerate(documents):
chunks = self.splitter.split_text(doc_text)
for j, chunk in enumerate(chunks):
all_chunks.append(chunk)
all_metadatas.append({"source": source_name, "doc_index": i, "chunk_index": j})
all_ids.append(f"{source_name}_{i}_{j}")
self.vector_store.add_documents(
texts=all_chunks,
metadatas=all_metadatas,
ids=all_ids
)
print(f"已导入 {len(documents)} 个文档，共 {len(all_chunks)} 个片段")
def ask(self, question: str, top_k: int = 5, temperature: float = 0.3) -> str:
"""基于RAG的问答"""
# 1. 检索相关文档
search_results = self.vector_store.search(question, top_k=top_k)
# 2. 组装上下文
context_parts = []
for i, result in enumerate(search_results):
context_parts.append(f"[文档{i + 1}] (来源: {result['metadata']['source']})\n{result['text']}")
context = "\n\n---\n\n".join(context_parts)
# 3. 构造提示词
system_prompt = """你是一个知识问答助手。请根据以下检索到的文档内容回答用户的问题。
要求：
1. 优先基于检索到的文档内容回答
2. 如果文档中没有相关信息，请明确说明
3. 引用信息时注明来源文档编号
4. 不要编造文档中没有的信息"""
user_prompt = f"""检索到的相关文档：
{context}
用户问题：{question}
请基于以上文档内容回答问题。"""
# 4. 调用LLM生成回答
messages = [
{"role": "system", "content": system_prompt},
]
messages.extend(self.conversation_history[-6:]) # 保留最近3轮
messages.append({"role": "user", "content": user_prompt})
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature,
max_tokens=2048
)
answer = response.choices[0].message.content
# 更新对话历史
self.conversation_history.append({"role": "user", "content": question})
self.conversation_history.append({"role": "assistant", "content": answer})
return answer
def ask_with_sources(self, question: str, top_k: int = 5) -> dict:
"""带来源信息的问答"""
search_results = self.vector_store.search(question, top_k=top_k)
answer = self.ask(question, top_k, temperature=0.3)
return {
"question": question,
"answer": answer,
"sources": [
{
"text": r["text"][:200] + "...",
"source": r["metadata"]["source"],
"relevance_score": 1 - r["distance"] # 转换为相似度
}
for r in search_results
]
}
# 使用示例
if __name__ == "__main__":
rag_agent = RAGAgent(model="gemini-2.5-flash-lite")
# 导入文档
sample_docs = [
"""Python是一种广泛使用的高级编程语言，由Guido van Rossum于1991年首次发布。
Python的设计哲学强调代码的可读性和简洁性，其语法允许程序员用更少的代码行来表达概念。
Python支持多种编程范式，包括面向对象、命令式、函数式和过程式编程。
Python拥有一个庞大且不断增长的标准库，以及丰富的第三方包生态系统。""",
"""机器学习是人工智能的一个子领域，它使计算机系统能够从数据中学习和改进，
而无需进行明确的编程。机器学习算法构建一个基于样本数据的数学模型，
称为训练数据，以便在没有明确编程来执行任务的情况下做出预测或决策。
机器学习主要分为三类：监督学习、无监督学习和强化学习。""",
"""深度学习是机器学习的一个子领域，基于人工神经网络的研究。
深度学习使用多层神经网络来学习数据的层次表示。
深度学习在图像识别、语音识别、自然语言处理等领域取得了突破性进展。
常见的深度学习框架包括TensorFlow、PyTorch和Keras。"""
]
rag_agent.ingest_documents(sample_docs, source_name="tech_intro")
# 问答
result = rag_agent.ask_with_sources("什么是深度学习？")
print("问题：", result["question"])
print("回答：", result["answer"])
print("\n参考来源：")
for src in result["sources"]:
print(f" - {src['source']} (相关度: {src['relevance_score']:.3f})")
RAG智能体的效果受多个因素影响，下面列出关键优化方向：
| 优化维度 | 具体方法 | 效果 |
|---|---|---|
| 文档切分 | 语义切分替代固定长度切分 | 减少信息碎片化 |
| Embedding模型 | 使用领域微调的Embedding模型 | 提升语义匹配精度 |
| 检索策略 | 混合检索（向量+关键词） | 兼顾语义和精确匹配 |
| 重排序 | 使用Cross-Encoder对检索结果重排 | 提升Top-K结果质量 |
| 上下文压缩 | 对检索文档进行压缩或摘要 | 减少噪声，降低token消耗 |
| 查询改写 | LLM改写用户查询 | 提升检索召回率 |
| 多路召回 | 多种检索策略并行，合并结果 | 提升召回率 |
9 多智能体协作系统
9.1 多智能体架构
随着任务复杂度的增加，单个智能体往往难以胜任所有工作。多智能体系统（Multi-Agent System, MAS）通过将不同能力的智能体组合协作，能够解决远超单个智能体能力范围的复杂任务。这种设计理念类似于人类团队协作——不同专业背景的人各司其职，通过沟通协调完成共同目标。
多智能体系统的架构模式主要有以下几种：
主从架构（Master-Worker）：一个主智能体负责任务分解和调度，多个工作智能体负责执行具体子任务。主智能体根据任务需求将工作分配给最合适的工作智能体，并整合各工作智能体的结果。这种架构简单直观，易于实现，但主智能体可能成为瓶颈。
管道架构（Pipeline）：多个智能体按照固定顺序依次处理任务，每个智能体的输出作为下一个智能体的输入。这种架构适合流程化的任务，如"研究→撰写→审核→发布"的内容生产流程。管道架构的缺点是灵活性较低，无法根据中间结果动态调整流程。
辩论架构（Debate）：多个智能体从不同角度分析同一问题，通过辩论达成共识。这种架构适合需要多角度思考的决策类任务，能够减少单一视角的偏见。辩论架构的关键挑战是如何设计有效的辩论终止条件和共识达成机制。
层级架构（Hierarchical）：智能体按照层级组织，高层智能体负责战略规划，中层智能体负责战术执行，底层智能体负责具体操作。这种架构适合复杂的大型任务，能够实现从宏观到微观的逐层细化。
9.2 角色分工与通信
多智能体系统的核心挑战之一是智能体之间的通信机制。高效的通信需要在信息完整性和通信成本之间取得平衡——传递过多信息会浪费token和增加延迟，传递过少信息则可能导致智能体之间理解不一致。
智能体之间的通信方式主要有三种：直接消息传递（一个智能体直接向另一个智能体发送消息）、共享记忆（所有智能体共享一个公共记忆空间，通过读写记忆进行间接通信）和黑板模式（一个公共的"黑板"区域，智能体可以发布和订阅特定类型的信息）。
import os
import json
from typing import List, Dict, Optional
from dotenv import load_dotenv
from openai import OpenAI
load_dotenv()
class AgentMessage:
"""智能体间通信消息"""
def __init__(self, sender: str, receiver: str, content: str, msg_type: str = "info"):
self.sender = sender
self.receiver = receiver
self.content = content
self.msg_type = msg_type # info, task, result, error
self.timestamp = None
def to_dict(self):
return {
"sender": self.sender,
"receiver": self.receiver,
"content": self.content,
"msg_type": self.msg_type
}
class BaseAgent:
"""基础智能体类"""
def __init__(self, name: str, role: str, model: str = "gemini-2.5-flash-lite"):
self.name = name
self.role = role
self.model = model
self.client = OpenAI(
api_key=os.getenv("API_KEY"),
base_url=os.getenv("BASE_URL")
)
self.history: List[Dict] = []
self.inbox: List[AgentMessage] = []
def process(self, message: str, system_prompt: str = None) -> str:
"""处理消息并生成回复"""
if system_prompt is None:
system_prompt = f"你是{self.name}，角色是{self.role}。请根据你的角色定位来处理任务。"
messages = [{"role": "system", "content": system_prompt}]
messages.extend(self.history[-10:]) # 保留最近5轮
messages.append({"role": "user", "content": message})
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=0.7,
max_tokens=2048
)
result = response.choices[0].message.content
self.history.append({"role": "user", "content": message})
self.history.append({"role": "assistant", "content": result})
return result
def send_message(self, receiver: str, content: str, msg_type: str = "info") -> AgentMessage:
"""发送消息给其他智能体"""
return AgentMessage(
sender=self.name,
receiver=receiver,
content=content,
msg_type=msg_type
)
def receive_message(self, message: AgentMessage):
"""接收消息"""
self.inbox.append(message)
def clear_history(self):
"""清空历史"""
self.history = []
self.inbox = []
9.3 实现多智能体系统
下面我们实现一个基于主从架构的多智能体协作系统，包含一个协调者和多个专业智能体：
class MultiAgentSystem:
"""多智能体协作系统"""
def __init__(self, coordinator_model="gemini-2.5-flash-lite", worker_model="gemini-2.5-flash-lite"):
self.coordinator = BaseAgent(
name="协调者",
role="任务分解和结果整合的协调者",
model=coordinator_model
)
self.workers: Dict[str, BaseAgent] = {}
self.worker_model = worker_model
self.message_log: List[Dict] = []
def register_worker(self, name: str, role: str, model: str = None):
"""注册工作智能体"""
agent = BaseAgent(
name=name,
role=role,
model=model or self.worker_model
)
self.workers[name] = agent
print(f"已注册智能体: {name} ({role})")
def run(self, task: str, max_rounds: int = 5) -> str:
"""执行多智能体协作任务"""
print(f"\n{'='*60}")
print(f"任务: {task}")
print(f"{'='*60}")
# 第一步：协调者分解任务
worker_list = "\n".join([f"- {name}: {agent.role}" for name, agent in self.workers.items()])
decomposition_prompt = f"""请将以下任务分解为子任务，并分配给合适的工作智能体。
可用的工作智能体：
{worker_list}
任务：{task}
请以JSON格式输出，格式如下：
{{
"subtasks": [
{{"task": "子任务描述", "assignee": "智能体名称"}},
...
]
}}
只输出JSON，不要其他内容。"""
decomposition = self.coordinator.process(decomposition_prompt)
print(f"\n[协调者] 任务分解结果：\n{decomposition}")
# 解析子任务
try:
# 提取JSON部分
json_str = decomposition
if "```json" in json_str:
json_str = json_str.split("```json")[1].split("```")[0]
elif "```" in json_str:
json_str = json_str.split("```")[1].split("```")[0]
subtasks = json.loads(json_str.strip())["subtasks"]
except (json.JSONDecodeError, KeyError) as e:
print(f"[系统] 任务分解解析失败: {e}")
return "任务分解失败，无法继续执行。"
# 第二步：分配并执行子任务
results = {}
for i, subtask in enumerate(subtasks):
assignee = subtask["assignee"]
task_desc = subtask["task"]
if assignee not in self.workers:
print(f"\n[系统] 智能体 '{assignee}' 不存在，跳过该子任务")
continue
print(f"\n--- 子任务 {i+1}/{len(subtasks)} ---")
print(f"分配给: {assignee}")
print(f"任务: {task_desc}")
result = self.workers[assignee].process(task_desc)
results[assignee] = result
print(f"\n[{assignee}] 完成结果：\n{result[:200]}...")
# 第三步：协调者整合结果
results_summary = "\n\n".join([
f"## {name}的结果\n{result}"
for name, result in results.items()
])
integration_prompt = f"""请整合以下各智能体的工作结果，生成一份完整的最终报告。
原始任务：{task}
各智能体的结果：
{results_summary}
请生成一份结构清晰、内容完整的最终报告。"""
final_report = self.coordinator.process(integration_prompt)
print(f"\n{'='*60}")
print(f"最终报告：\n{final_report}")
print(f"{'='*60}")
return final_report
# 使用示例
if __name__ == "__main__":
system = MultiAgentSystem()
# 注册专业智能体
system.register_worker("研究员", "负责信息搜索和资料收集，提供事实依据和背景信息")
system.register_worker("分析师", "负责数据分析和逻辑推理，提供专业见解和结论")
system.register_worker("撰写者", "负责将信息整合为结构清晰、表达流畅的文章或报告")
# 执行协作任务
system.run("分析2024年人工智能行业的发展趋势，并撰写一份简要的分析报告")
多智能体系统的设计需要考虑以下关键问题：
| 设计问题 | 考虑因素 | 推荐方案 |
|---|---|---|
| 智能体粒度 | 过粗则能力混杂，过细则通信开销大 | 按功能域划分，每个智能体专注一个领域 |
| 通信协议 | 信息完整性与通信成本的权衡 | 结构化消息格式，包含必要上下文 |
| 冲突解决 | 不同智能体给出矛盾结论 | 引入裁判智能体或投票机制 |
| 容错机制 | 单个智能体失败的影响 | 主智能体监控，失败时重新分配 |
| 终止条件 | 何时停止迭代 | 设定最大轮次和质量阈值 |
10 前端集成：HTML与JavaScript
10.1 Web智能体架构
将智能体部署为Web应用是使其真正可用的关键步骤。Web智能体架构通常采用前后端分离的设计：后端提供API服务，负责与LLM交互；前端提供用户界面，负责展示对话和收集输入。这种架构的优势在于前后端可以独立开发、部署和扩展，同时支持多种客户端（浏览器、移动端等）访问。
Web智能体的核心技术挑战在于流式输出的实现。在传统HTTP请求中，客户端发送请求后等待服务器返回完整响应。但对于LLM的流式输出，我们需要使用Server-Sent Events（SSE）或WebSocket技术，将模型生成的每个token实时推送到前端，实现逐字显示的效果。
10.2 后端API服务搭建
我们使用Flask框架搭建后端API服务，它轻量、灵活，非常适合智能体应用的快速开发。
# app.py - 后端API服务
import json
from flask import Flask, request, jsonify, Response, stream_with_context
from flask_cors import CORS
from openai import OpenAI
app = Flask(__name__)
CORS(app) # 允许跨域请求
# 初始化 OpenAI 客户端（已直接填入你的密钥和地址）
client = OpenAI(
api_key="sk-",
base_url="https://api.aigc.bar/v1"
)
# 简单的会话存储（生产环境应使用数据库）
sessions = {}
def get_or_create_session(session_id: str) -> list:
"""获取或创建会话"""
if session_id not in sessions:
sessions[session_id] = []
return sessions[session_id]
@app.route("/api/chat", methods=["POST"])
def chat():
"""普通对话接口"""
data = request.json
message = data.get("message", "")
session_id = data.get("session_id", "default")
model = data.get("model", "gemini-2.5-flash-lite")
system_prompt = data.get("system_prompt", "你是一个有帮助的AI助手。")
if not message:
return jsonify({"error": "消息不能为空"}), 400
# 获取会话历史
history = get_or_create_session(session_id)
# 构造消息
messages = [{"role": "system", "content": system_prompt}]
messages.extend(history)
messages.append({"role": "user", "content": message})
try:
response = client.chat.completions.create(
model=model,
messages=messages,
temperature=0.7,
max_tokens=2048
)
assistant_message = response.choices[0].message.content
# 更新会话历史
history.append({"role": "user", "content": message})
history.append({"role": "assistant", "content": assistant_message})
return jsonify({
"response": assistant_message,
"model": model,
"tokens": response.usage.total_tokens,
"session_id": session_id
})
except Exception as e:
return jsonify({"error": str(e)}), 500
@app.route("/api/chat/stream", methods=["POST"])
def chat_stream():
"""流式对话接口"""
data = request.json
message = data.get("message", "")
session_id = data.get("session_id", "default")
model = data.get("model", "gemini-2.5-flash-lite")
system_prompt = data.get("system_prompt", "你是一个有帮助的AI助手。")
if not message:
return jsonify({"error": "消息不能为空"}), 400
history = get_or_create_session(session_id)
messages = [{"role": "system", "content": system_prompt}]
messages.extend(history)
messages.append({"role": "user", "content": message})
def generate():
full_response = ""
try:
stream = client.chat.completions.create(
model=model,
messages=messages,
temperature=0.7,
max_tokens=2048,
stream=True
)
for chunk in stream:
if chunk.choices[0].delta.content is not None:
content = chunk.choices[0].delta.content
full_response += content
# SSE格式：data: {json}\n\n
yield f"data: {json.dumps({'content': content})}\n\n"
# 更新会话历史
history.append({"role": "user", "content": message})
history.append({"role": "assistant", "content": full_response})
# 发送结束标记
yield f"data: {json.dumps({'done': True, 'tokens': len(full_response)})}\n\n"
except Exception as e:
yield f"data: {json.dumps({'error': str(e)})}\n\n"
return Response(
stream_with_context(generate()),
mimetype="text/event-stream",
headers={
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no"
}
)
@app.route("/api/session/<session_id>/clear", methods=["POST"])
def clear_session(session_id):
"""清空会话"""
if session_id in sessions:
sessions[session_id] = []
return jsonify({"status": "ok"})
@app.route("/api/models", methods=["GET"])
def list_models():
"""列出可用模型"""
models = [
{"id": "gemini-2.5-flash-lite", "name": "Gemini Flash Lite", "type": "free", "speed": "极快"},
{"id": "gpt-4o-mini", "name": "GPT-4o Mini", "type": "paid", "speed": "快"},
{"id": "gpt-4o", "name": "GPT-4o", "type": "paid", "speed": "中等"},
{"id": "claude-sonnet-4-20250514", "name": "Claude Sonnet 4", "type": "paid", "speed": "中等"},
]
return jsonify(models)
if __name__ == "__main__":
app.run(debug=True, port=5000)
10.3 前端界面实现
前端界面是用户与智能体交互的窗口。我们使用纯HTML+CSS+JavaScript实现一个现代化的聊天界面，无需任何构建工具。
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>AI智能体对话</title>
<style>
* {
margin: 0;
padding: 0;
box-sizing: border-box;
}
body {
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif;
background: #f5f5f5;
height: 100vh;
display: flex;
flex-direction: column;
}
.header {
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
padding: 16px 24px;
display: flex;
align-items: center;
justify-content: space-between;
box-shadow: 0 2px 10px rgba(0,0,0,0.1);
}
.header h1 {
font-size: 20px;
font-weight: 600;
}
.header-controls {
display: flex;
gap: 12px;
align-items: center;
}
.header-controls select {
padding: 6px 12px;
border-radius: 6px;
border: none;
font-size: 14px;
cursor: pointer;
}
.btn {
padding: 8px 16px;
border: none;
border-radius: 6px;
cursor: pointer;
font-size: 14px;
transition: all 0.2s;
}
.btn-clear {
background: rgba(255,255,255,0.2);
color: white;
}
.btn-clear:hover {
background: rgba(255,255,255,0.3);
}
.chat-container {
flex: 1;
overflow-y: auto;
padding: 20px;
max-width: 800px;
width: 100%;
margin: 0 auto;
}
.message {
margin-bottom: 16px;
display: flex;
gap: 12px;
animation: fadeIn 0.3s ease;
}
@keyframes fadeIn {
from { opacity: 0; transform: translateY(10px); }
to { opacity: 1; transform: translateY(0); }
}
.message.user {
flex-direction: row-reverse;
}
.avatar {
width: 36px;
height: 36px;
border-radius: 50%;
display: flex;
align-items: center;
justify-content: center;
font-size: 16px;
flex-shrink: 0;
}
.message.user .avatar {
background: #667eea;
color: white;
}
.message.assistant .avatar {
background: #e8f5e9;
color: #2e7d32;
}
.bubble {
max-width: 70%;
padding: 12px 16px;
border-radius: 12px;
line-height: 1.6;
font-size: 15px;
white-space: pre-wrap;
word-break: break-word;
}
.message.user .bubble {
background: #667eea;
color: white;
border-bottom-right-radius: 4px;
}
.message.assistant .bubble {
background: white;
color: #333;
border-bottom-left-radius: 4px;
box-shadow: 0 1px 3px rgba(0,0,0,0.1);
}
.input-area {
padding: 16px 20px;
background: white;
border-top: 1px solid #e0e0e0;
max-width: 800px;
width: 100%;
margin: 0 auto;
}
.input-wrapper {
display: flex;
gap: 12px;
align-items: flex-end;
}
.input-wrapper textarea {
flex: 1;
padding: 12px 16px;
border: 2px solid #e0e0e0;
border-radius: 12px;
font-size: 15px;
resize: none;
outline: none;
font-family: inherit;
max-height: 120px;
transition: border-color 0.2s;
}
.input-wrapper textarea:focus {
border-color: #667eea;
}
.btn-send {
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
padding: 12px 24px;
border-radius: 12px;
font-size: 15px;
font-weight: 600;
white-space: nowrap;
}
.btn-send:hover {
opacity: 0.9;
}
.btn-send:disabled {
opacity: 0.5;
cursor: not-allowed;
}
.typing-indicator {
display: inline-block;
}
.typing-indicator span {
display: inline-block;
width: 8px;
height: 8px;
border-radius: 50%;
background: #999;
margin: 0 2px;
animation: bounce 1.4s infinite ease-in-out;
}
.typing-indicator span:nth-child(1) { animation-delay: 0s; }
.typing-indicator span:nth-child(2) { animation-delay: 0.2s; }
.typing-indicator span:nth-child(3) { animation-delay: 0.4s; }
@keyframes bounce {
0%, 80%, 100% { transform: scale(0); }
40% { transform: scale(1); }
}
</style>
</head>
<body>
<div class="header">
<h1>🤖 AI智能体</h1>
<div class="header-controls">
<select id="modelSelect">
<option value="gemini-2.5-flash-lite">Gemini Flash Lite (免费)</option>
<option value="gpt-4o-mini">GPT-4o Mini</option>
<option value="gpt-4o">GPT-4o</option>
</select>
<button class="btn btn-clear" onclick="clearChat()">清空对话</button>
</div>
</div>
<div class="chat-container" id="chatContainer">
<div class="message assistant">
<div class="avatar">🤖</div>
<div class="bubble">你好！我是AI智能体，有什么可以帮助你的吗？</div>
</div>
</div>
<div class="input-area">
<div class="input-wrapper">
<textarea id="messageInput" placeholder="输入你的消息..." rows="1"
onkeydown="handleKeyDown(event)"></textarea>
<button class="btn btn-send" id="sendBtn" onclick="sendMessage()">发送</button>
</div>
</div>
<script>
const API_BASE = 'http://localhost:5000/api';
let sessionId = 'session_' + Date.now();
let isStreaming = false;
// 自动调整文本框高度
const textarea = document.getElementById('messageInput');
textarea.addEventListener('input', function() {
this.style.height = 'auto';
this.style.height = Math.min(this.scrollHeight, 120) + 'px';
});
function handleKeyDown(event) {
if (event.key === 'Enter' && !event.shiftKey) {
event.preventDefault();
sendMessage();
}
}
function addMessage(role, content) {
const container = document.getElementById('chatContainer');
const messageDiv = document.createElement('div');
messageDiv.className = `message ${role}`;
const avatar = document.createElement('div');
avatar.className = 'avatar';
avatar.textContent = role === 'user' ? '👤' : '🤖';
const bubble = document.createElement('div');
bubble.className = 'bubble';
bubble.textContent = content;
messageDiv.appendChild(avatar);
messageDiv.appendChild(bubble);
container.appendChild(messageDiv);
container.scrollTop = container.scrollHeight;
return bubble;
}
function addTypingIndicator() {
const container = document.getElementById('chatContainer');
const messageDiv = document.createElement('div');
messageDiv.className = 'message assistant';
messageDiv.id = 'typingIndicator';
const avatar = document.createElement('div');
avatar.className = 'avatar';
avatar.textContent = '🤖';
const bubble = document.createElement('div');
bubble.className = 'bubble';
bubble.innerHTML = '<div class="typing-indicator"><span></span><span></span><span></span></div>';
messageDiv.appendChild(avatar);
messageDiv.appendChild(bubble);
container.appendChild(messageDiv);
container.scrollTop = container.scrollHeight;
}
function removeTypingIndicator() {
const indicator = document.getElementById('typingIndicator');
if (indicator) indicator.remove();
}
async function sendMessage() {
const input = document.getElementById('messageInput');
const message = input.value.trim();
if (!message || isStreaming) return;
// 显示用户消息
addMessage('user', message);
input.value = '';
input.style.height = 'auto';
// 禁用发送按钮
isStreaming = true;
document.getElementById('sendBtn').disabled = true;
// 显示打字指示器
addTypingIndicator();
const model = document.getElementById('modelSelect').value;
try {
// 使用流式接口
const response = await fetch(`${API_BASE}/chat/stream`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
message: message,
session_id: sessionId,
model: model
})
});
removeTypingIndicator();
// 创建助手消息气泡
const bubble = addMessage('assistant', '');
let fullContent = '';
const reader = response.body.getReader();
const decoder = new TextDecoder();
while (true) {
const { done, value } = await reader.read();
if (done) break;
const text = decoder.decode(value);
const lines = text.split('\n');
for (const line of lines) {
if (line.startsWith('data: ')) {
try {
const data = JSON.parse(line.slice(6));
if (data.content) {
fullContent += data.content;
bubble.textContent = fullContent;
document.getElementById('chatContainer').scrollTop =
document.getElementById('chatContainer').scrollHeight;
}
if (data.done) {
console.log('流式传输完成');
}
if (data.error) {
bubble.textContent = '错误: ' + data.error;
}
} catch (e) {
// 忽略解析错误
}
}
}
}
} catch (error) {
removeTypingIndicator();
addMessage('assistant', '连接失败，请检查后端服务是否启动。');
console.error('Error:', error);
}
isStreaming = false;
document.getElementById('sendBtn').disabled = false;
input.focus();
}
async function clearChat() {
const container = document.getElementById('chatContainer');
container.innerHTML = '';
addMessage('assistant', '对话已清空，有什么新的问题吗？');
await fetch(`${API_BASE}/session/${sessionId}/clear`, { method: 'POST' });
sessionId = 'session_' + Date.now();
}
</script>
</body>
</html>
10.4 完整Web智能体部署
将后端和前端整合在一起，我们可以创建一个完整的Web智能体应用。项目结构如下：
web-agent/
├── app.py # Flask后端
├── .env # 环境变量
├── requirements.txt # Python依赖
├── static/
│ └── index.html # 前端页面
└── templates/ # (可选) Jinja模板
在app.py中添加静态文件服务：
from flask import send_from_directory
@app.route("/")
def index():
return send_from_directory("static", "index.html")
安装依赖并启动服务：
pip install flask flask-cors openai python-dotenv
python app.py
打开浏览器访问http://localhost:5000即可使用智能体对话界面。
11 国产AI API接入实战
11.1 为什么选择国产AI API
在前面的章节中，我们主要使用的是国际主流模型的API。然而，在实际应用中，国产AI API具有独特的优势，特别是在中文场景下。
首先，国产模型在中文理解和生成方面具有天然优势。由于训练数据中中文语料占比更高，国产模型对中文的语法、语义和文化背景理解更深，生成的中文文本更加自然流畅。在中文问答、中文写作、中文代码注释等场景中，国产模型的表现往往优于同级别的国际模型。
其次，国产AI API的访问稳定性和延迟表现更优。由于服务器部署在国内，网络延迟更低，不会受到国际网络波动的影响。对于对响应速度敏感的实时应用（如在线客服、语音助手），这一优势尤为明显。
第三，国产AI API的价格通常更加亲民。国内厂商为了推广生态，往往提供更具竞争力的定价策略，部分模型甚至提供免费额度。对于初创团队和个人开发者来说，这大大降低了使用门槛。
第四，数据合规性是选择国产API的重要考量。根据中国的数据安全法规，某些行业的数据必须存储在境内，使用国产API可以更好地满足合规要求。
下面的表格对比了主流国产AI API的特点：
| 平台 | 代表模型 | 上下文长度 | 特色能力 | 免费额度 |
|---|---|---|---|---|
| 智谱AI (GLM) | glm-4-plus | 128K | 长文本理解、代码生成 | 有 |
| 月之暗面 (KIMI) | moonshot-v1 | 128K | 超长文本处理 | 有 |
| 百度文心 | ernie-4.0 | 128K | 知识增强、多模态 | 有 |
| 阿里通义 | qwen-max | 32K | 多语言、代码 | 有 |
| 讯飞星火 | spark-4.0 | 32K | 语音交互、教育 | 有 |
| DeepSeek | deepseek-chat | 64K | 代码、数学推理 | 有 |
11.2 KIMI API接入
KIMI（月之暗面）以其超长上下文处理能力著称，支持高达128K的上下文窗口，非常适合处理长文档、长对话等场景。KIMI的API同样兼容OpenAI规范，接入非常简单。
import os
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
class KimiAgent:
"""基于KIMI API的智能体"""
def __init__(self, model="moonshot-v1-8k"):
self.client = OpenAI(
api_key=os.getenv("KIMI_API_KEY"),
base_url="https://api.moonshot.cn/v1"
)
self.model = model
self.history = []
def chat(self, user_message: str, temperature=0.7, max_tokens=2048) -> str:
"""对话"""
messages = [
{"role": "system", "content": "你是一个专业的AI助手，擅长中文理解和生成。"}
]
messages.extend(self.history)
messages.append({"role": "user", "content": user_message})
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens
)
result = response.choices[0].message.content
self.history.append({"role": "user", "content": user_message})
self.history.append({"role": "assistant", "content": result})
return result
def analyze_document(self, document: str, question: str) -> str:
"""分析长文档"""
# KIMI支持超长上下文，可以直接将整个文档放入
messages = [
{
"role": "system",
"content": "你是一个文档分析助手。请根据提供的文档内容回答问题。"
},
{
"role": "user",
"content": f"文档内容：\n{document}\n\n问题：{question}"
}
]
response = self.client.chat.completions.create(
model="moonshot-v1-128k", # 使用128K上下文模型
messages=messages,
temperature=0.3,
max_tokens=4096
)
return response.choices[0].message.content
def stream_chat(self, user_message: str):
"""流式对话"""
messages = [
{"role": "system", "content": "你是一个专业的AI助手。"}
]
messages.extend(self.history)
messages.append({"role": "user", "content": user_message})
stream = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=0.7,
max_tokens=2048,
stream=True
)
full_response = ""
for chunk in stream:
if chunk.choices[0].delta.content:
content = chunk.choices[0].delta.content
print(content, end="", flush=True)
full_response += content
print()
self.history.append({"role": "user", "content": user_message})
self.history.append({"role": "assistant", "content": full_response})
return full_response
# 使用示例
if __name__ == "__main__":
agent = KimiAgent(model="moonshot-v1-8k")
# 普通对话
print(agent.chat("请介绍一下Python的装饰器模式"))
# 长文档分析
with open("long_document.txt", "r", encoding="utf-8") as f:
doc = f.read()
print(agent.analyze_document(doc, "这篇文档的核心观点是什么？"))
11.3 GLM API接入
GLM（智谱AI）是国内领先的大语言模型，在中文理解、代码生成和逻辑推理方面表现出色。GLM的API同样兼容OpenAI规范，接入方式与KIMI类似。
import os
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
class GlmAgent:
"""基于GLM API的智能体"""
def __init__(self, model="glm-4-flash"):
self.client = OpenAI(
api_key=os.getenv("GLM_API_KEY"),
base_url="https://open.bigmodel.cn/api/paas/v4"
)
self.model = model
self.history = []
def chat(self, user_message: str, temperature=0.7, max_tokens=2048) -> str:
"""对话"""
messages = [
{"role": "system", "content": "你是一个智能助手，擅长中文对话和知识问答。"}
]
messages.extend(self.history)
messages.append({"role": "user", "content": user_message})
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens
)
result = response.choices[0].message.content
self.history.append({"role": "user", "content": user_message})
self.history.append({"role": "assistant", "content": result})
return result
def chat_with_tools(self, user_message: str, tools: list = None):
"""带工具调用的对话"""
messages = [
{"role": "system", "content": "你是一个可以使用工具的智能助手。"}
]
messages.extend(self.history)
messages.append({"role": "user", "content": user_message})
kwargs = {
"model": self.model,
"messages": messages,
"temperature": 0.7,
"max_tokens": 2048,
}
if tools:
kwargs["tools"] = tools
kwargs["tool_choice"] = "auto"
response = self.client.chat.completions.create(**kwargs)
return response
def multi_model_chat(self, user_message: str) -> dict:
"""多模型对比对话"""
models = ["glm-4-flash", "glm-4-plus"]
results = {}
for model in models:
self.client.chat.completions.create = lambda **kw: self.client.chat.completions.create(**kw)
response = self.client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": "你是一个有帮助的AI助手。"},
{"role": "user", "content": user_message}
],
temperature=0.7,
max_tokens=1024
)
results[model] = response.choices[0].message.content
return results
# 使用示例
if __name__ == "__main__":
agent = GlmAgent(model="glm-5.1")
# 基础对话
print(agent.chat("请解释什么是Transformer架构"))
# 多模型对比
results = agent.multi_model_chat("Python中列表和元组的区别是什么？")
for model, answer in results.items():
print(f"\n[{model}]: {answer[:200]}...")
11.4 模型选择策略
面对众多可用的模型，如何选择最适合自己应用场景的模型是一个重要问题。本节提供一套系统的模型选择策略，帮助你在质量、速度和成本之间找到最佳平衡。
模型选择的核心原则是"够用就好"——不要为简单任务使用昂贵的模型，也不要为复杂任务使用能力不足的模型。在实际应用中，推荐采用"分层模型策略"：简单任务使用免费或低成本模型（如gemini-2.5-flash-lite、glm-4-flash），中等任务使用中等成本模型（如gpt-4o-mini、kimi），复杂任务使用高性能模型（如gpt-4o、glm-4-plus）。
下面的表格提供了不同应用场景的模型推荐：
| 应用场景 | 任务特点 | 推荐模型 | 理由 |
|---|---|---|---|
| 快速问答 | 简单、实时 | gemini-2.5-flash-lite | 免费、极快 |
| 中文对话 | 中文理解 | glm-4-flash / kimi | 中文优化、免费额度 |
| 代码生成 | 逻辑推理 | gpt-4o / deepseek-chat | 代码能力强 |
| 长文档分析 | 长上下文 | kimi-128k | 128K上下文 |
| 复杂推理 | 高质量 | gpt-4o / claude-sonnet | 推理能力强 |
| 日常助手 | 均衡 | gpt-4o-mini / glm-4-flash | 性价比高 |
| 数据分析 | 结构化输出 | gpt-4o / glm-4-plus | 格式遵循好 |
12 高级提示词工程
12.1 提示词设计原则
提示词（Prompt）是与大语言模型交互的核心接口，其质量直接决定了模型输出的质量。提示词工程（Prompt Engineering）是构建高质量智能体的必备技能，它不是简单的"写提示词"，而是一套系统化的方法论。
提示词设计的核心原则可以概括为"CRISP"框架：Clarity（清晰性）、Relevance（相关性）、Instruction（指令性）、Specificity（具体性）和Prioritization（优先级）。
清晰性要求提示词表述明确，避免歧义。模糊的提示词会导致模型输出不确定，例如"写一篇关于AI的文章"就比"写一篇800字的科普文章，面向非技术读者，介绍人工智能在医疗领域的三个具体应用案例"要模糊得多。后者明确了字数、受众、主题和结构要求，模型更容易生成符合预期的内容。
相关性要求提示词只包含与任务相关的信息，避免无关内容干扰模型的注意力。研究表明，当提示词中包含大量无关信息时，模型的输出质量会显著下降，这被称为"注意力稀释"效应。
指令性要求使用明确的指令动词，告诉模型应该做什么。常见的指令动词包括：分析、比较、总结、列举、解释、翻译、生成、评估等。指令动词应该放在提示词的开头，让模型第一时间明确任务类型。
具体性要求提供足够的细节和约束条件，包括输出格式、长度限制、语气风格、目标受众等。具体的约束条件能够有效减少模型输出的随机性，提高结果的可控性。
优先级要求将最重要的信息放在提示词的开头和结尾。研究表明，LLM对提示词开头和结尾的信息关注度更高，中间部分的信息容易被"遗忘"，这被称为"Lost in the Middle"现象。
12.2 系统提示词设计
系统提示词（System Prompt）是定义智能体角色和行为规范的核心机制。一个精心设计的系统提示词可以让模型在特定领域表现出专业水准，而一个糟糕的系统提示词则可能导致模型输出不一致、不可控。
# 优秀的系统提示词示例
CODING_ASSISTANT_PROMPT = """你是一位资深的Python编程专家，拥有10年以上的开发经验。
## 你的职责
- 帮助用户解决Python编程问题
- 提供高质量的代码示例和最佳实践
- 解释代码的工作原理和设计思路
## 行为规范
1. 代码必须遵循PEP 8规范
2. 优先使用Python标准库和主流第三方库
3. 每个代码示例都要包含注释说明
4. 如果有多种实现方式，比较它们的优缺点
5. 对于性能敏感的场景，提供时间复杂度分析
## 输出格式
- 代码使用```python代码块包裹
- 解释使用中文
- 关键术语保留英文原文
## 限制
- 不生成恶意代码
- 不提供绕过安全机制的方法
- 如果问题不明确，主动询问细节"""
RESEARCH_ASSISTANT_PROMPT = """你是一位学术研究助手，擅长文献综述和研究方法指导。
## 你的职责
- 帮助用户进行学术文献检索和分析
- 提供研究方法论建议
- 协助撰写学术论文的各个部分
## 行为规范
1. 回答必须基于学术证据，引用具体的研究
2. 区分事实和观点，明确标注不确定性
3. 使用学术语言，避免口语化表达
4. 提供多个视角，不偏袒特定观点
5. 如果超出知识范围，明确说明
## 输出格式
- 使用学术写作风格
- 重要论点提供参考文献标注
- 使用结构化的段落组织内容"""
CUSTOMER_SERVICE_PROMPT = """你是一位专业的客户服务代表，代表一家科技公司提供售后支持。
## 你的职责
- 解答客户关于产品使用的问题
- 处理客户的投诉和退换货请求
- 提供技术支持和故障排查指导
## 行为规范
1. 始终保持礼貌和耐心，即使客户情绪激动
2. 先理解客户的问题，再提供解决方案
3. 对于无法解决的问题，提供升级渠道
4. 不承诺无法兑现的解决方案
5. 记录关键信息以便后续跟进
## 回复模板
- 问候 + 确认理解客户问题
- 提供解决方案或排查步骤
- 确认问题是否解决
- 提供后续支持渠道"""
系统提示词的设计需要遵循以下最佳实践：第一，使用Markdown格式组织内容，层次清晰；第二，明确定义角色、职责、行为规范和限制条件；第三，提供具体的输出格式要求；第四，包含边界情况的处理指引；第五，定期根据实际使用反馈迭代优化。
12.3 Few-Shot与Chain-of-Thought
Few-Shot（少样本学习）和Chain-of-Thought（思维链）是两种强大的提示词技术，能够显著提升模型在特定任务上的表现。
Few-Shot通过在提示词中提供少量示例，让模型理解任务的具体要求和期望的输出格式。研究表明，即使只提供2-3个高质量示例，也能大幅提升模型的输出质量。Few-Shot的关键在于示例的质量而非数量——一个精心设计的示例比十个随意选择的示例更有效。
# Few-Shot提示词示例
few_shot_prompt = """请根据输入的情感文本，判断情感倾向。
示例1：
输入：这部电影太精彩了，演员的表演令人震撼！
输出：正面
示例2：
输入：服务态度很差，等了一个小时才上菜。
输出：负面
示例3：
输入：今天天气一般，不好不坏。
输出：中性
现在请判断：
输入：{user_input}
输出："""
Chain-of-Thought（CoT）通过引导模型展示推理过程，提升复杂推理任务的准确率。CoT的核心思想是：让模型"思考可见"，将推理步骤显式化，而不是直接给出答案。这种方式不仅提高了准确率，还使得推理过程可审查、可调试。
# Chain-of-Thought提示词示例
cot_prompt = """请一步步思考并回答以下问题。
问题：一个商店打8折促销，小明买了一件原价250元的衣服和一条原价180元的裤子，他一共需要付多少钱？
让我们一步步思考：
1. 首先，计算衣服的折扣价：250 × 0.8 = 200元
2. 然后，计算裤子的折扣价：180 × 0.8 = 144元
3. 最后，计算总价：200 + 144 = 344元
答案：小明一共需要付344元。
---
现在请用同样的方式回答以下问题：
{question}"""
CoT有两种使用方式：Zero-Shot CoT和Few-Shot CoT。Zero-Shot CoT只需在提示词末尾添加"让我们一步步思考"即可触发模型的推理能力，无需提供示例。Few-Shot CoT则在提示词中提供包含推理过程的示例，让模型学习如何进行结构化推理。
12.4 提示词模板管理
在构建复杂的智能体系统时，提示词的管理变得非常重要。好的提示词管理应该支持版本控制、A/B测试和动态组装。
from string import Template
from typing import Dict, Optional
class PromptTemplate:
"""提示词模板管理器"""
def __init__(self):
self.templates: Dict[str, str] = {}
def register(self, name: str, template: str):
"""注册模板"""
self.templates[name] = template
def render(self, name: str, **kwargs) -> str:
"""渲染模板"""
if name not in self.templates:
raise ValueError(f"模板 '{name}' 不存在")
template = Template(self.templates[name])
return template.safe_substitute(**kwargs)
def list_templates(self) -> list:
"""列出所有模板"""
return list(self.templates.keys())
# 注册常用模板
prompt_manager = PromptTemplate()
prompt_manager.register("chat", """你是一个$role。
## 职责
$responsibilities
## 行为规范
$rules
## 输出格式
$output_format""")
prompt_manager.register("rag_qa", """请根据以下检索到的文档内容回答用户的问题。
检索到的文档：
$context
用户问题：$question
要求：
1. 优先基于文档内容回答
2. 如果文档中没有相关信息，请明确说明
3. 引用信息时注明来源
4. 不要编造信息""")
prompt_manager.register("tool_calling", """你是一个可以使用工具的智能助手。
可用工具：
$tools_description
当需要获取实时信息或执行操作时，请使用相应的工具。
如果不需要工具就能回答，请直接回答。
用户问题：$question""")
prompt_manager.register("summarize", """请对以下内容进行摘要：
原始内容：
$content
要求：
- 摘要长度：$length
- 语言风格：$style
- 重点突出：$focus""")
prompt_manager.register("code_review", """请审查以下$language代码：
```$language
$code
请从以下维度进行审查：
1. 代码正确性：是否存在逻辑错误
2. 代码风格：是否符合编码规范
3. 性能优化：是否有性能瓶颈
4. 安全性：是否存在安全漏洞
5. 可维护性：代码是否易于理解和修改
请给出具体的问题描述和改进建议。""")
# 使用示例
system_prompt = prompt_manager.render(
"chat",
role="Python编程专家",
responsibilities="- 帮助用户解决Python编程问题\n- 提供高质量代码示例",
rules="1. 代码遵循PEP 8\n2. 包含注释说明",
output_format="代码使用```python代码块，解释使用中文"
)
rag_prompt = prompt_manager.render(
"rag_qa",
context="Python是一种高级编程语言...",
question="Python有什么特点？"
)
13 部署与生产化
13.1 容器化部署
将智能体应用容器化是现代化部署的最佳实践。Docker容器提供了环境一致性、快速部署和弹性扩展的能力，确保应用在开发、测试和生产环境中行为一致。
首先创建Dockerfile：
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
# 安装系统依赖
RUN apt-get update && apt-get install -y --no-install-recommends \
gcc \
&& rm -rf /var/lib/apt/lists/*
# 复制依赖文件并安装
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 复制应用代码
COPY . .
# 暴露端口
EXPOSE 5000
# 设置环境变量
ENV PYTHONUNBUFFERED=1
# 启动命令
CMD ["gunicorn", "--bind", "0.0.0.0:5000", "--workers", "4", "--timeout", "120", "app:app"]
创建docker-compose.yml：
version: '3.8'
services:
agent-api:
build: .
ports:
- "5000:5000"
environment:
- API_KEY=${API_KEY}
- BASE_URL=${BASE_URL}
env_file:
- .env
restart: unless-stopped
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:5000/api/models"]
interval: 30s
timeout: 10s
retries: 3
deploy:
resources:
limits:
memory: 512M
reservations:
memory: 256M
# 可选：Nginx反向代理
nginx:
image: nginx:alpine
ports:
- "80:80"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf
- ./static:/usr/share/nginx/html
depends_on:
- agent-api
restart: unless-stopped
生产环境的requirements.txt应该包含更多依赖：
flask>=3.0.0
flask-cors>=4.0.0
gunicorn>=21.2.0
openai>=1.0.0
python-dotenv>=1.0.0
requests>=2.31.0
chromadb>=0.4.0
pytesseract>=0.3.10
redis>=5.0.0
13.2 监控与日志
生产环境的智能体系统需要完善的监控和日志体系，以便及时发现和解决问题。监控的核心指标包括API调用延迟、成功率、token消耗、错误率和并发数。
import logging
import time
import json
from functools import wraps
from datetime import datetime
# 配置结构化日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("AgentMonitor")
class AgentMonitor:
"""智能体监控器"""
def __init__(self):
self.metrics = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"total_tokens": 0,
"total_latency": 0.0,
"model_usage": {}, # 模型使用统计
"error_types": {}, # 错误类型统计
}
def record_request(self, model: str, tokens: int, latency: float, success: bool, error_type: str = None):
"""记录请求指标"""
self.metrics["total_requests"] += 1
self.metrics["total_latency"] += latency
self.metrics["total_tokens"] += tokens
if success:
self.metrics["successful_requests"] += 1
else:
self.metrics["failed_requests"] += 1
if error_type:
self.metrics["error_types"][error_type] = self.metrics["error_types"].get(error_type, 0) + 1
# 模型使用统计
if model not in self.metrics["model_usage"]:
self.metrics["model_usage"][model] = {"count": 0, "tokens": 0, "latency": 0.0}
self.metrics["model_usage"][model]["count"] += 1
self.metrics["model_usage"][model]["tokens"] += tokens
self.metrics["model_usage"][model]["latency"] += latency
def get_summary(self) -> dict:
"""获取监控摘要"""
total = self.metrics["total_requests"]
if total == 0:
return {"message": "暂无数据"}
success_rate = self.metrics["successful_requests"] / total * 100
avg_latency = self.metrics["total_latency"] / total
return {
"total_requests": total,
"success_rate": f"{success_rate:.1f}%",
"avg_latency": f"{avg_latency:.2f}s",
"total_tokens": self.metrics["total_tokens"],
"model_usage": self.metrics["model_usage"],
"top_errors": sorted(
self.metrics["error_types"].items(),
key=lambda x: x[1],
reverse=True
)[:5]
}
def monitor_agent(monitor: AgentMonitor):
"""监控装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
try:
result = func(*args, **kwargs)
latency = time.time() - start_time
# 从结果中提取指标
tokens = result.get("tokens", 0) if isinstance(result, dict) else 0
model = kwargs.get("model", "unknown")
monitor.record_request(
model=model,
tokens=tokens,
latency=latency,
success=True
)
logger.info(json.dumps({
"event": "request_success",
"model": model,
"tokens": tokens,
"latency": f"{latency:.3f}s"
}))
return result
except Exception as e:
latency = time.time() - start_time
error_type = type(e).__name__
monitor.record_request(
model=kwargs.get("model", "unknown"),
tokens=0,
latency=latency,
success=False,
error_type=error_type
)
logger.error(json.dumps({
"event": "request_failed",
"error_type": error_type,
"error_message": str(e),
"latency": f"{latency:.3f}s"
}))
raise
return wrapper
return decorator
13.3 成本优化
API调用成本是智能体运营的主要开支之一。有效的成本优化策略能够在不牺牲服务质量的前提下大幅降低运营成本。
成本优化的核心策略包括以下几个方面：
模型分层策略：根据任务复杂度选择合适的模型，避免"杀鸡用牛刀"。简单任务使用低成本模型，复杂任务才调用高性能模型。可以通过规则引擎或分类模型自动判断任务复杂度，动态选择模型。
缓存策略：对于重复性高的查询，使用缓存避免重复调用API。缓存可以基于查询的语义相似度实现——当新查询与缓存中的查询语义相似度超过阈值时，直接返回缓存结果。语义缓存的实现需要Embedding模型和向量数据库的支持。
上下文压缩：在多轮对话中，通过摘要压缩、重要性过滤等方式减少发送给模型的token数量。每减少1000个输入token，就能节省相应的费用。对于长上下文场景，上下文压缩的节省效果尤为显著。
批量处理：对于非实时任务，将多个请求合并为批量请求，减少API调用次数。部分API提供商对批量请求提供折扣价格。
Token预算控制：为每个用户或会话设置token预算上限，防止异常使用导致成本失控。同时设置告警机制，当消耗接近预算时及时通知。
class CostOptimizer:
"""成本优化器"""
# 模型价格表（每1K tokens的价格，单位：元）
MODEL_PRICING = {
"gemini-2.5-flash-lite": {"input": 0.0, "output": 0.0}, # 免费
"gpt-4o-mini": {"input": 0.01, "output": 0.03},
"gpt-4o": {"input": 0.05, "output": 0.15},
"glm-4-flash": {"input": 0.0, "output": 0.0}, # 免费额度
"glm-4-plus": {"input": 0.05, "output": 0.05},
"moonshot-v1-8k": {"input": 0.01, "output": 0.01},
}
def __init__(self, daily_budget: float = 10.0):
self.daily_budget = daily_budget
self.daily_spent = 0.0
self.cache = {} # 简单的查询缓存
def calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""计算单次请求成本"""
pricing = self.MODEL_PRICING.get(model, {"input": 0.05, "output": 0.05})
cost = (input_tokens / 1000 * pricing["input"]) + (output_tokens / 1000 * pricing["output"])
return cost
def select_model(self, task_complexity: str, budget_remaining: float = None) -> str:
"""根据任务复杂度和预算选择模型"""
if budget_remaining is None:
budget_remaining = self.daily_budget - self.daily_spent
if task_complexity == "simple":
return "gemini-2.5-flash-lite"
elif task_complexity == "medium":
if budget_remaining > 1.0:
return "gpt-4o-mini"
else:
return "glm-4-flash"
elif task_complexity == "complex":
if budget_remaining > 5.0:
return "gpt-4o"
elif budget_remaining > 1.0:
return "glm-4-plus"
else:
return "gpt-4o-mini"
return "gemini-2.5-flash-lite"
def check_cache(self, query: str) -> str | None:
"""检查缓存"""
# 简单的精确匹配缓存
# 生产环境应使用语义缓存
return self.cache.get(query)
def update_cache(self, query: str, response: str):
"""更新缓存"""
# 限制缓存大小
if len(self.cache) > 1000:
# 删除最早的缓存项
oldest_key = next(iter(self.cache))
del self.cache[oldest_key]
self.cache[query] = response
def check_budget(self, estimated_cost: float) -> bool:
"""检查预算"""
return (self.daily_spent + estimated_cost) <= self.daily_budget
def record_spending(self, cost: float):
"""记录花费"""
self.daily_spent += cost
14 综合实战：构建完整的智能体应用
14.1 项目需求分析
经过前面章节的学习，我们已经掌握了构建智能体所需的各种核心技术。现在，让我们将这些技术整合起来，构建一个完整的智能体应用——“智能编程助手”。
这个智能编程助手需要具备以下核心功能：代码问答（回答编程相关问题）、代码审查（审查用户提交的代码）、文档检索（基于技术文档的RAG问答）、多语言支持（支持Python、JavaScript、Java等多种编程语言）和交互式对话（支持多轮对话和流式输出）。
14.2 完整代码实现
下面是智能编程助手的完整实现代码。这个实现整合了前面章节的所有核心技术，包括多轮对话、工具调用、RAG、流式输出和成本优化。
# smart_coding_assistant.py
import json
import time
import logging
import os
import subprocess
from typing import List, Dict, Optional
from openai import OpenAI
logger = logging.getLogger("SmartCodingAssistant")
class SmartCodingAssistant:
"""智能编程助手 - 综合实战"""
def __init__(self, model="gemini-2.5-flash-lite", premium_model="gpt-4o-mini"):
# ✅ 直接填入你的密钥和地址，删除所有环境变量
self.client = OpenAI(
api_key="sk-",
base_url="https://api.aigc.bar/v1"
)
self.model = model
self.premium_model = premium_model
self.history: List[Dict] = []
self.max_history = 20
# 系统提示词
self.system_prompt = """你是一位资深的全栈编程专家，精通Python、JavaScript、Java等多种编程语言。
## 核心能力
- 代码编写：提供高质量、可运行的代码示例
- 代码审查：发现潜在问题并提供改进建议
- 问题诊断：帮助定位和解决编程问题
- 架构设计：提供系统设计和技术选型建议
## 行为规范
1. 代码必须可运行，包含必要的import语句
2. 遵循各语言的编码规范（Python PEP 8, JavaScript ESLint等）
3. 关键代码添加注释说明
4. 提供多种实现方案时，比较优缺点
5. 安全相关的代码要特别标注注意事项
## 输出格式
- 代码使用对应语言的代码块包裹
- 解释使用中文，技术术语保留英文
- 复杂问题分步骤解答"""
# 工具定义
self.tools = [
{
"type": "function",
"function": {
"name": "execute_python",
"description": "执行Python代码并返回输出结果",
"parameters": {
"type": "object",
"properties": {
"code": {"type": "string", "description": "要执行的Python代码"}
},
"required": ["code"]
}
}
},
{
"type": "function",
"function": {
"name": "search_documentation",
"description": "搜索技术文档和API参考",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"},
"language": {"type": "string", "description": "编程语言",
"enum": ["python", "javascript", "java", "go"]}
},
"required": ["query"]
}
}
}
]
# 工具函数映射
self.tool_functions = {
"execute_python": self._execute_python,
"search_documentation": self._search_documentation,
}
def _execute_python(self, code: str) -> str:
"""执行Python代码（安全沙箱版本）"""
try:
result = subprocess.run(
["python", "-c", code],
capture_output=True,
text=True,
timeout=10,
env={"PATH": os.environ.get("PATH", "")}
)
output = result.stdout if result.stdout else result.stderr
return output[:2000] if output else "（无输出）"
except subprocess.TimeoutExpired:
return "错误：代码执行超时（10秒限制）"
except Exception as e:
return f"执行错误：{str(e)}"
def _search_documentation(self, query: str, language: str = "python") -> str:
"""搜索文档（模拟实现）"""
doc_cache = {
"python": {
"list comprehension": "列表推导式: [expr for item in iterable if condition]",
"decorator": "装饰器: @decorator\ndef func(): ...",
"async": "异步: async def func(): await coroutine()",
},
"javascript": {
"promise": "Promise: new Promise((resolve, reject) => {...})",
"async await": "async/await: async function() { await promise }",
"arrow function": "箭头函数: const fn = (args) => expression",
}
}
lang_docs = doc_cache.get(language, {})
results = []
for key, value in lang_docs.items():
if query.lower() in key.lower():
results.append(f"{key}: {value}")
return "\n".join(results) if results else f"未找到关于'{query}'的{language}文档"
def chat(self, user_message: str, stream: bool = False) -> str:
"""主对话方法"""
messages = [{"role": "system", "content": self.system_prompt}]
messages.extend(self.history[-self.max_history:])
messages.append({"role": "user", "content": user_message})
max_iterations = 5
iteration = 0
while iteration < max_iterations:
iteration += 1
kwargs = {
"model": self.model,
"messages": messages,
"temperature": 0.7,
"max_tokens": 4096,
"tools": self.tools,
"tool_choice": "auto"
}
try:
response = self.client.chat.completions.create(**kwargs)
except Exception as e:
logger.error(f"API调用失败: {e}")
return f"抱歉，服务暂时不可用：{str(e)}"
message = response.choices[0].message
messages.append(message)
# 处理工具调用
if message.tool_calls:
for tool_call in message.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
logger.info(f"工具调用: {func_name}({func_args})")
if func_name in self.tool_functions:
result = self.tool_functions[func_name](**func_args)
else:
result = f"未知工具: {func_name}"
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(result)
})
continue
# 无工具调用，返回最终回复
final_response = message.content
self.history.append({"role": "user", "content": user_message})
self.history.append({"role": "assistant", "content": final_response})
return final_response
return "抱歉，处理过程中工具调用次数过多，请简化您的问题。"
def chat_stream(self, user_message: str):
"""流式对话"""
messages = [{"role": "system", "content": self.system_prompt}]
messages.extend(self.history[-self.max_history:])
messages.append({"role": "user", "content": user_message})
stream = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=0.7,
max_tokens=4096,
stream=True
)
full_response = ""
for chunk in stream:
if chunk.choices[0].delta.content:
content = chunk.choices[0].delta.content
full_response += content
yield content
self.history.append({"role": "user", "content": user_message})
self.history.append({"role": "assistant", "content": full_response})
def review_code(self, code: str, language: str = "python") -> str:
"""代码审查"""
prompt = f"""请审查以下{language}代码，从正确性、风格、性能、安全性和可维护性五个维度进行评估：
```{language}
{code}
请给出：
1. 总体评分（1-10）
2. 每个维度的具体问题和建议
3. 改进后的代码示例"""
return self.chat(prompt)
def explain_code(self, code: str, language: str = "python") -> str:
"""代码解释"""
prompt = f"""请详细解释以下{language}代码的工作原理：
```{language}
{code}
请逐行或逐段解释，包括：
1. 整体功能和目的
2. 关键逻辑的解释
3. 使用的设计模式或技巧"""
return self.chat(prompt)
def generate_code(self, description: str, language: str = "python") -> str:
"""代码生成"""
prompt = f"""请根据以下描述生成{language}代码：
{description}
要求：
1. 代码完整可运行
2. 包含必要的注释
3. 遵循{language}的编码规范
4. 包含错误处理"""
return self.chat(prompt)
# 命令行交互
if __name__ == "__main__":
assistant = SmartCodingAssistant()
print("=" * 60)
print("🤖 智能编程助手")
print("命令: /review <代码> | /explain <代码> | /generate <描述> | /clear | /quit")
print("=" * 60)
while True:
user_input = input("\n你: ").strip()
if not user_input:
continue
if user_input == "/quit":
print("再见！")
break
elif user_input == "/clear":
assistant.history = []
print("对话已清空。")
continue
elif user_input.startswith("/review "):
code = user_input[8:]
print("\n助手: ", end="")
for chunk in assistant.chat_stream(f"请审查以下代码：\n{code}"):
print(chunk, end="", flush=True)
print()
elif user_input.startswith("/explain "):
code = user_input[9:]
print("\n助手: ", end="")
for chunk in assistant.chat_stream(f"请解释以下代码：\n{code}"):
print(chunk, end="", flush=True)
print()
elif user_input.startswith("/generate "):
desc = user_input[10:]
print("\n助手: ", end="")
for chunk in assistant.chat_stream(f"请生成代码：\n{desc}"):
print(chunk, end="", flush=True)
print()
else:
print("\n助手: ", end="")
for chunk in assistant.chat_stream(user_input):
print(chunk, end="", flush=True)
print()
14.3 智能体效果评估
构建智能体之后，评估其效果是持续改进的基础。智能体的评估维度包括准确性、相关性、完整性和效率。
准确性评估关注智能体的回答是否正确，是否存在事实性错误或幻觉。评估方法包括人工标注（专家对回答进行正确性判断）、自动对比（与标准答案对比）和交叉验证（多个模型对同一问题的回答进行对比）。
相关性评估关注智能体的回答是否与用户的问题相关，是否偏题或答非所问。评估方法包括语义相似度计算（回答与问题的语义相关性）和人工评分。
完整性评估关注智能体的回答是否完整地覆盖了用户问题的所有方面，是否遗漏了重要信息。评估方法包括关键信息覆盖率计算和人工检查。
效率评估关注智能体的响应速度和资源消耗，包括延迟、token消耗和工具调用次数。
class AgentEvaluator:
"""智能体评估器"""
def __init__(self, agent):
self.agent = agent
self.evaluation_results = []
def evaluate_qa(self, test_cases: list) -> dict:
"""评估问答能力"""
results = {
"total": len(test_cases),
"scores": [],
"details": []
}
for case in test_cases:
question = case["question"]
expected_keywords = case.get("keywords", [])
# 获取智能体回答
answer = self.agent.chat(question)
# 评估关键词覆盖率
covered = sum(1 for kw in expected_keywords if kw.lower() in answer.lower())
coverage = covered / len(expected_keywords) if expected_keywords else 0
results["scores"].append(coverage)
results["details"].append({
"question": question,
"answer": answer[:200],
"keyword_coverage": f"{coverage:.1%}",
"expected_keywords": expected_keywords
})
results["average_score"] = sum(results["scores"]) / len(results["scores"]) if results["scores"] else 0
return results
def evaluate_latency(self, test_messages: list, rounds: int = 3) -> dict:
"""评估响应延迟"""
latencies = []
for msg in test_messages:
for _ in range(rounds):
start = time.time()
self.agent.chat(msg)
latency = time.time() - start
latencies.append(latency)
return {
"avg_latency": f"{sum(latencies)/len(latencies):.3f}s",
"min_latency": f"{min(latencies):.3f}s",
"max_latency": f"{max(latencies):.3f}s",
"p95_latency": f"{sorted(latencies)[int(len(latencies)*0.95)]:.3f}s"
}
15 总结与展望
15.1 本文核心要点回顾
本文从智能体的基本概念出发，系统讲解了基于大模型API构建智能体的完整技术栈。让我们回顾一下核心要点：
第一，智能体的本质是"LLM作为大脑 + 规划/记忆/工具/行动四大组件"的架构范式。理解这个架构是构建任何智能体的理论基础。规划模块负责任务分解和策略制定，记忆模块负责信息持久化和上下文管理，工具模块负责扩展与外部世界的交互能力，行动模块负责执行决策。
第二，API调用是智能体与LLM交互的基础。OpenAI兼容API规范是当前的事实标准，掌握这套规范就能调用绝大多数模型。关键概念包括消息角色（system/user/assistant/tool）、流式输出（SSE）、工具调用（Function Calling）和token管理。
第三，Function Calling是智能体从"只能聊天"进化为"能做事"的关键能力。通过定义工具并让模型自主决定何时调用，智能体可以获取实时信息、执行操作和访问外部系统。工具调用循环（推理→调用→观察→推理）是智能体自主运行的核心引擎。
第四，RAG技术通过检索增强解决了LLM的知识时效性和幻觉问题。RAG系统的三大阶段——索引、检索和生成——各有优化空间，从文档切分策略到Embedding模型选择，从检索算法到上下文压缩，每个环节的优化都能提升最终效果。
第五，多智能体协作通过角色分工和通信机制，使智能体系统能够处理远超单智能体能力范围的复杂任务。主从架构、管道架构、辩论架构和层级架构各有适用场景，选择合适的架构模式是多智能体系统设计的关键。
第六，国产AI API（如KIMI、GLM）在中文场景中具有独特优势，包括更好的中文理解能力、更低的访问延迟、更亲民的价格和更好的数据合规性。在实际应用中，应该根据任务特点灵活选择国际模型和国产模型。
15.2 智能体技术展望
智能体技术正在快速发展，以下几个方向值得关注：
自主性提升：未来的智能体将具备更强的自主决策能力，能够在更少的人类干预下完成复杂任务。这需要更强大的规划能力、更可靠的自纠错机制和更完善的安全约束。当前的智能体在长程任务中容易出现"迷失"现象——在多步推理后偏离原始目标——解决这一问题需要新的架构设计。
多模态融合：当前大多数智能体主要处理文本信息，未来的智能体将能够理解和生成图像、音频、视频等多种模态的内容。多模态智能体可以处理更丰富的输入（如截图、语音指令），提供更直观的输出（如生成图表、语音回复），从而在更广泛的应用场景中发挥作用。
个性化与持续学习：未来的智能体将能够根据用户的偏好和使用习惯持续优化自身行为，实现真正的个性化服务。这需要高效的在线学习算法和安全的记忆更新机制，确保智能体既能适应用户需求，又不会产生偏见或安全问题。
智能体生态：随着智能体数量的增加，智能体之间的协作和竞争将形成复杂的生态系统。智能体市场、智能体编排平台和智能体安全审计等基础设施将逐步完善，推动智能体技术从实验室走向大规模商业应用。
安全与对齐：随着智能体能力的增强，确保其行为安全、可控、符合人类价值观变得越来越重要。智能体安全研究包括对抗性攻击防御、输出过滤、权限控制和行为审计等方向，这些研究对于智能体的可信部署至关重要。
构建智能体是一项充满挑战但也极具价值的工作。希望本文能够帮助你迈出第一步，在智能体的世界中探索无限可能。记住，最好的学习方式是动手实践——从最简单的API调用开始，逐步添加工具、记忆和规划能力，最终构建出属于你自己的智能体系统。
声明：本文所有代码示例均基于OpenAI兼容API规范编写，可通过配置不同的base_url和api_key来切换不同的模型提供商。文中涉及的模型名称和价格信息可能随时间变化，请以各平台官方信息为准。代码由AI生成，不一定与你现有的Python版本兼容，可以借助AI对话助手进行修改，修改和调校后以兼容大家开发智能体。
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
20
0- 0
已为社区贡献161条内容
所有评论(0)