最近在尝试将大语言模型(LLM)应用到实际工作流中时,你是否也遇到过这样的困扰:模型虽然能说会道,但让它真正执行一个多步骤的复杂任务,比如“帮我分析一下上周的销售数据并生成报告”,它要么直接拒绝,要么给出一个笼统的建议,无法真正“动手”完成。这正是当前AI应用从“聊天”走向“实干”的关键瓶颈。

而“智能体(Agent)”技术的出现,正在打破这一僵局。它让大语言模型拥有了规划、执行、使用工具和反思的能力,从一个被动的知识库,转变为一个能主动解决问题的“数字员工”。作为国内领先的AI应用,豆包也推出了自己的Agent平台,让开发者能够基于豆包大模型,快速构建具备强大行动力的智能应用。

本文将以一个完整的实战项目为例,手把手带你从零开始,深入豆包Agent的开发世界。无论你是想为自己的业务添加自动化能力,还是对AI Agent开发充满好奇的开发者,都能通过本文掌握从环境搭建、技能定义、到复杂任务编排的全流程。我们将构建一个能够自动进行网络搜索、信息整理并生成简报的“信息助理”Agent,让你真切感受到“豆包真能干活了!”的魅力。

1. 智能体(Agent)核心概念与豆包平台定位

在深入代码之前,我们有必要厘清几个核心概念,理解豆包Agent在整个技术图谱中的位置。

1.1 什么是AI智能体(Agent)?

简单来说,一个AI智能体是一个能够感知环境、自主决策并执行行动以实现目标的系统。在LLM的语境下,智能体通常由以下几部分构成:

  1. 核心大脑(LLM) :负责理解任务、进行规划、做出决策。豆包大模型就扮演了这个角色。
  2. 规划能力 :将复杂的用户指令(如“写一份行业分析报告”)拆解成一系列可执行的子任务(如“搜索行业趋势”、“查找头部公司”、“总结关键数据”)。
  3. 工具使用能力 :智能体可以调用外部工具来获取信息或执行操作。这是其“能干实事”的关键。工具可以是搜索引擎API、数据库查询、代码执行器、甚至控制软件鼠标键盘的自动化脚本。
  4. 记忆与反思 :智能体能够记住对话历史和之前执行步骤的结果,并能对执行过程进行反思,如果当前步骤失败了,它会尝试另一种方法。

传统的聊天模型是“一问一答,答完即止”,而智能体是“接受目标,自主规划,调用工具,持续执行,直至完成”。

1.2 豆包Agent开放平台是什么?

豆包Agent开放平台是字节跳动推出的,基于豆包大模型构建智能体应用的一站式平台。它为广大开发者和企业提供了低门槛、高效率的智能体创建、调试、部署与管理能力。其核心优势在于:

  • 模型即服务 :直接集成强大的豆包系列模型,无需自行训练或部署大模型,降低了技术门槛和成本。
  • 丰富的工具生态 :平台提供了预置的官方工具(如联网搜索、知识库查询、文本处理等),并支持开发者自定义工具(通过API、函数等方式),极大地扩展了智能体的能力边界。
  • 可视化编排 :提供了低代码/无代码的工作流编排界面,可以通过拖拽方式定义智能体的决策逻辑和执行流程,使得复杂逻辑的构建更加直观。
  • 便捷部署与集成 :创建的智能体可以轻松发布为API、网页应用或嵌入到其他产品中。

1.3 相关技术栈与学习路线

对于开发者而言,入门豆包Agent开发,建议遵循以下学习路径:

  1. 基础 :熟悉Python编程语言,了解基本的HTTP API调用(Requests库)和JSON数据处理。
  2. 核心 :掌握豆包Agent平台的官方文档,理解其核心概念: Agent Skill (技能/工具)、 Workflow (工作流)、 Memory (记忆)。
  3. 实践 :从创建一个简单的、能调用单个工具的Agent开始,逐步过渡到多工具协作、具备复杂工作流的Agent。
  4. 进阶 :探索自定义工具开发、长程记忆管理、复杂决策逻辑优化以及与现有业务系统的深度集成。

接下来,我们将进入实战环节,从环境准备开始。

2. 环境准备与豆包开发者账号配置

工欲善其事,必先利其器。开发豆包Agent的第一步是准备好开发环境并获取必要的权限和凭证。

2.1 注册豆包开放平台账号

  1. 访问豆包开放平台官方网站。
  2. 使用字节跳动账号(如抖音、头条账号)登录。如果没有,需先注册。
  3. 完成开发者实名认证,这是创建应用和调用API的必要步骤。

2.2 创建应用与获取API Key

登录开放平台后,你需要创建一个应用来管理你的Agent。

  1. 在控制台找到“应用管理”或“创建应用”入口。
  2. 填写应用名称、描述等信息。应用类型选择“智能体”或“API调用”。
  3. 创建成功后,进入应用详情页,找到“凭证与安全”或“API Key”管理页面。
  4. 生成API Key :创建一个新的Key,并立即妥善保存。这个Key是程序调用豆包模型和Agent服务的通行证, 一旦泄露可能造成资源盗用 。通常它看起来像一串由字母数字组成的令牌。

2.3 本地Python开发环境搭建

我们将使用Python作为主要开发语言,通过官方SDK来调用服务。

  1. 安装Python :确保你的系统已安装Python 3.8或更高版本。可以在终端输入 python --version python3 --version 检查。
  2. 安装豆包Python SDK :官方SDK封装了API调用,使用起来更简便。
    pip install volcengine-python-sdk
    # 或者指定安装包含豆包服务的版本,具体包名请以官方文档为准
    # pip install volcengine-python-sdk[maas] 
    
    注意:SDK的具体名称和安装方式可能随官方更新而变化,请以豆包开放平台最新文档为准。
  3. 准备代码编辑器 :推荐使用VSCode、PyCharm等现代IDE,它们对Python有很好的支持。

环境就绪后,我们就可以开始编写第一个能“干活”的豆包Agent了。

3. 豆包Agent核心组件与API初探

在动手编码前,我们先通过官方SDK,了解构建一个Agent所需的核心组件和基本调用模式。

3.1 核心组件:Agent, Skill, Memory

  • Agent(智能体) :智能体的本体,承载了LLM核心和基础配置(如系统指令、温度参数等)。你可以通过平台可视化创建,也可以通过API定义。
  • Skill(技能) :即智能体可以使用的“工具”。一个Skill本质上是一个可被LLM理解和调用的函数。它包含:
    • name : 技能名称,LLM根据名称决定何时调用。
    • description : 技能描述,LLM根据描述理解技能的功能和适用场景。
    • parameters : 技能所需的输入参数定义(JSON Schema格式)。
    • function : 实际执行该技能的函数。
  • Memory(记忆) :使Agent拥有上下文对话能力。包括短期记忆(当前会话)和长期记忆(可持久化存储的历史)。本文示例主要使用会话记忆。

3.2 使用SDK进行基础对话

让我们先完成一个最简单的验证:使用SDK调用豆包模型进行一次对话,确保API Key和网络连通性正常。

# 文件:test_basic_chat.py
from volcengine.maas import MaasService, MaasException, ChatRole

# 1. 配置服务信息
service = MaasService('maas-api.ml-platform-cn-beijing.volces.com', 'cn-beijing') # 服务地址和区域
service.set_ak('你的AccessKey ID') # 替换为你的AK
service.set_sk('你的Secret Access Key') # 替换为你的SK

# 2. 定义请求参数
req = {
    "model": {
        "name": "ep-20250225141520-ftlpj" # 替换为你想使用的具体模型名称,例如豆包-Pro
    },
    "messages": [
        {
            "role": ChatRole.USER,
            "content": "你好,请介绍一下你自己。"
        }
    ],
    "parameters": {
        "max_tokens": 1000, # 生成的最大token数
        "temperature": 0.8, # 创造性,越高越随机
    }
}

# 3. 发起调用
try:
    resp = service.chat(req)
    print(resp.choice.message.content)
except MaasException as e:
    print(e)
except Exception as e:
    print(e)

运行与解释

  • 将代码中的 AccessKey ID , Secret Access Key model.name 替换为你自己在平台获取的值。
  • 运行脚本 python test_basic_chat.py ,你应该能收到豆包模型的自我介绍回复。
  • 这个例子展示了直接调用模型API,还不是Agent。Agent在此基础上增加了对 tools (即Skill)参数的支持和自动调用逻辑。

4. 实战:构建“信息助理”智能体(单技能版)

现在,我们来构建本文的核心案例——一个能联网搜索的“信息助理”。首先从单一技能开始:让Agent能调用一个搜索工具。

4.1 设计技能:模拟搜索工具

由于直接调用真实的搜索引擎API需要额外的注册和密钥,为了简化教程,我们首先模拟一个搜索工具。在后续进阶部分,我们会替换为真实工具。

# 文件:skills/search_skill.py
import json

def mock_web_search(query: str, max_results: int = 3) -> str:
    """
    模拟网络搜索工具。
    
    Args:
        query (str): 搜索查询词。
        max_results (int): 返回的最大结果数量。
    
    Returns:
        str: 格式化的搜索结果字符串。
    """
    # 这是一个模拟的搜索结果数据库
    mock_data = {
        "大语言模型发展趋势 2024": [
            {"title": "多模态能力成为标配", "snippet": "2024年,主流LLM纷纷加强图像、音频理解与生成能力。"},
            {"title": "Agent智能体爆发", "snippet": "让LLM能规划、使用工具、执行任务的Agent框架成为研发热点。"},
            {"title": "上下文窗口持续增长", "snippet": "百万token级别的上下文窗口使得长文档处理成为可能。"},
        ],
        "Python异步编程": [
            {"title": "asyncio核心概念", "snippet": "理解事件循环、协程、Task和Future是掌握asyncio的关键。"},
            {"title": "async/await语法", "snippet": "用于定义协程,使异步代码编写更同步化,易于理解。"},
            {"title": "高性能网络应用", "snippet": "aiohttp, FastAPI等框架充分利用异步IO提升并发性能。"},
        ]
    }
    
    results = mock_data.get(query, [])
    results = results[:max_results]
    
    if not results:
        return f"未找到关于 '{query}' 的搜索结果。"
    
    formatted_results = [f"{i+1}. {res['title']}: {res['snippet']}" for i, res in enumerate(results)]
    return f"搜索 '{query}' 的结果如下:\n" + "\n".join(formatted_results)

# 测试这个技能
if __name__ == "__main__":
    print(mock_web_search("大语言模型发展趋势 2024"))
    print("\n---\n")
    print(mock_web_search("未知主题"))

4.2 将技能封装给Agent使用

豆包Agent的SDK要求技能以特定的格式(符合OpenAI Tool Call格式)提供。我们需要将上面的函数包装成合适的格式。

# 文件:agent_core.py
from volcengine.maas import MaasService, MaasException, ChatRole
import json
from skills.search_skill import mock_web_search

class SimpleInfoAssistant:
    def __init__(self, api_key, secret_key, model_name="ep-20250225141520-ftlpj"):
        self.service = MaasService('maas-api.ml-platform-cn-beijing.volces.com', 'cn-beijing')
        self.service.set_ak(api_key)
        self.service.set_sk(secret_key)
        self.model_name = model_name
        # 定义可用的工具(技能)列表
        self.tools = [
            {
                "type": "function",
                "function": {
                    "name": "mock_web_search",
                    "description": "根据用户提供的查询词,模拟进行网络搜索并返回摘要结果。用于获取最新、最相关的网络信息。",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "query": {
                                "type": "string",
                                "description": "需要搜索的关键词或问题,例如:'大语言模型最新进展'"
                            },
                            "max_results": {
                                "type": "integer",
                                "description": "希望返回的最大结果数量,默认是3",
                                "default": 3
                            }
                        },
                        "required": ["query"]
                    }
                }
            }
        ]
        # 工具名称到实际函数的映射
        self.tool_functions = {
            "mock_web_search": mock_web_search
        }
        self.conversation_history = [] # 简单的会话记忆

    def _execute_tool(self, tool_call):
        """执行被模型调用的工具。"""
        function_name = tool_call.function.name
        function_args = json.loads(tool_call.function.arguments)
        
        if function_name in self.tool_functions:
            print(f"[Agent 正在执行工具] {function_name},参数:{function_args}")
            result = self.tool_functions[function_name](**function_args)
            return result
        else:
            return f"错误:未知的工具 '{function_name}'。"

    def chat(self, user_input):
        """与智能体进行一轮对话。"""
        # 1. 将用户输入加入历史
        self.conversation_history.append({"role": ChatRole.USER, "content": user_input})
        
        # 2. 准备请求,这次带上工具定义
        req = {
            "model": {"name": self.model_name},
            "messages": self.conversation_history,
            "tools": self.tools, # 关键:告诉模型有哪些工具可用
            "tool_choice": "auto", # 让模型自动决定是否调用工具
            "parameters": {
                "max_tokens": 2000,
                "temperature": 0.7,
            }
        }
        
        try:
            # 3. 发起首次请求,模型可能会决定调用工具
            resp = self.service.chat(req)
            assistant_message = resp.choice.message
            self.conversation_history.append(assistant_message.to_dict()) # 保存模型回复
            
            tool_calls = getattr(assistant_message, 'tool_calls', None)
            
            # 4. 如果模型要求调用工具
            if tool_calls:
                tool_responses = []
                for tool_call in tool_calls:
                    # 执行工具
                    tool_result = self._execute_tool(tool_call)
                    # 将工具执行结果作为一条特殊消息追加到历史中
                    tool_response_msg = {
                        "role": "tool",
                        "content": json.dumps(tool_result, ensure_ascii=False),
                        "tool_call_id": tool_call.id
                    }
                    tool_responses.append(tool_response_msg)
                
                # 将工具执行结果全部加入历史
                self.conversation_history.extend(tool_responses)
                
                # 5. 再次请求模型,让它基于工具结果生成最终回复
                req_final = {
                    "model": {"name": self.model_name},
                    "messages": self.conversation_history,
                    "parameters": {
                        "max_tokens": 2000,
                        "temperature": 0.7,
                    }
                }
                final_resp = self.service.chat(req_final)
                final_message = final_resp.choice.message
                self.conversation_history.append(final_message.to_dict())
                return final_message.content
            else:
                # 模型没有调用工具,直接返回回复
                return assistant_message.content
                
        except MaasException as e:
            return f"请求API时发生错误:{e}"
        except Exception as e:
            return f"发生未知错误:{e}"

    def clear_history(self):
        """清空对话历史。"""
        self.conversation_history = []

4.3 运行与测试单技能Agent

创建一个主程序来测试我们的初级信息助理。

# 文件:main_single_skill.py
from agent_core import SimpleInfoAssistant
import os

# 从环境变量或直接填写你的密钥
API_KEY = os.getenv("DOUBAO_API_KEY", "你的_API_KEY")
SECRET_KEY = os.getenv("DOUBAO_SECRET_KEY", "你的_SECRET_KEY")
MODEL_NAME = "ep-20250225141520-ftlpj" # 替换为你的模型名

def main():
    assistant = SimpleInfoAssistant(API_KEY, SECRET_KEY, MODEL_NAME)
    
    print("=== 豆包信息助理(单技能版)已启动 ===")
    print("输入 'quit' 或 'exit' 退出程序。\n")
    
    while True:
        try:
            user_input = input("\n你:")
            if user_input.lower() in ['quit', 'exit', '退出']:
                print("再见!")
                break
            if not user_input.strip():
                continue
                
            print("\n助理:", end="", flush=True)
            response = assistant.chat(user_input)
            print(response)
            
        except KeyboardInterrupt:
            print("\n\n程序被中断。")
            break
        except Exception as e:
            print(f"\n发生错误:{e}")

if __name__ == "__main__":
    main()

运行与观察

  1. 在终端运行 python main_single_skill.py
  2. 尝试提问:“最近大语言模型有什么发展趋势?”
  3. 观察控制台输出。你应该会看到类似 [Agent 正在执行工具] mock_web_search,参数:{'query': '大语言模型发展趋势 2024', 'max_results': 3} 的日志,然后Agent会返回整合了搜索结果的回答。
  4. 再尝试一个不需要搜索的问题:“你好,今天天气怎么样?” 观察Agent是否会直接回答而不调用工具。

至此,你已经成功创建了一个具备基础工具调用能力的豆包Agent!它已经能根据你的问题,自主判断是否需要搜索,并整合信息回复你。

5. 进阶:构建多技能与工作流驱动的智能体

单一技能远远不够。一个真正能“干活”的助理,需要协调多种技能。接下来,我们为Agent添加信息整理和摘要生成的技能,并引入简单的工作流逻辑。

5.1 新增技能:文本摘要与整理

我们添加一个本地处理的技能,用于对长文本进行摘要。

# 文件:skills/summarize_skill.py

def summarize_text(text: str, max_length: int = 200) -> str:
    """
    对提供的文本进行摘要,保留核心信息。
    这是一个简化的模拟实现。在实际应用中,可以调用更专业的摘要API或模型。
    
    Args:
        text (str): 需要摘要的原始文本。
        max_length (int): 摘要的最大长度(字符数)。
    
    Returns:
        str: 生成的摘要文本。
    """
    if len(text) <= max_length:
        return text
    
    # 模拟摘要逻辑:取开头、中间和结尾的部分句子。
    sentences = text.replace('。', '。\n').split('\n')
    sentences = [s.strip() for s in sentences if s.strip()]
    
    if len(sentences) <= 3:
        return text[:max_length] + "..."
    
    # 选取首、中、尾句
    selected = [sentences[0]]
    mid_idx = len(sentences) // 2
    selected.append(sentences[mid_idx])
    selected.append(sentences[-1])
    
    summary = '。'.join(selected) + '。'
    if len(summary) > max_length:
        summary = summary[:max_length-3] + "..."
    return summary

# 测试
if __name__ == "__main__":
    long_text = "人工智能是研究、开发用于模拟、延伸和扩展人的智能的理论、方法、技术及应用系统的一门新的技术科学。人工智能领域的研究包括机器人、语言识别、图像识别、自然语言处理和专家系统等。人工智能从诞生以来,理论和技术日益成熟,应用领域也不断扩大,可以设想,未来人工智能带来的科技产品,将会是人类智慧的‘容器’。人工智能可以对人的意识、思维的信息过程的模拟。人工智能不是人的智能,但能像人那样思考、也可能超过人的智能。"
    print(summarize_text(long_text, 100))

5.2 升级Agent核心:集成多技能与简单工作流

现在,我们升级 agent_core.py ,集成搜索和摘要两个技能,并实现一个简单的工作流:当用户请求“报告”或“简报”时,自动执行“搜索 -> 摘要”的流程。

# 文件:agent_core_advanced.py
from volcengine.maas import MaasService, MaasException, ChatRole
import json
import re
from skills.search_skill import mock_web_search
from skills.summarize_skill import summarize_text

class AdvancedInfoAssistant:
    def __init__(self, api_key, secret_key, model_name="ep-20250225141520-ftlpj"):
        self.service = MaasService('maas-api.ml-platform-cn-beijing.volces.com', 'cn-beijing')
        self.service.set_ak(api_key)
        self.service.set_sk(secret_key)
        self.model_name = model_name
        
        # 定义多个工具
        self.tools = [
            {
                "type": "function",
                "function": {
                    "name": "mock_web_search",
                    "description": "根据查询词模拟网络搜索,返回相关信息摘要。用于获取事实性、时效性信息。",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "query": {"type": "string", "description": "搜索关键词"},
                            "max_results": {"type": "integer", "default": 3}
                        },
                        "required": ["query"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "summarize_text",
                    "description": "对长文本进行浓缩摘要,提取核心内容。适用于整理冗长的搜索结果或文档。",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "text": {"type": "string", "description": "需要摘要的原始文本"},
                            "max_length": {"type": "integer", "default": 200, "description": "摘要最大长度"}
                        },
                        "required": ["text"]
                    }
                }
            }
        ]
        
        self.tool_functions = {
            "mock_web_search": mock_web_search,
            "summarize_text": summarize_text
        }
        self.conversation_history = []
        
        # 简单的工作流触发器关键词
        self.report_keywords = ["报告", "简报", "总结一下", "汇总"]

    def _contains_report_request(self, text):
        """简单判断用户是否要求生成报告。"""
        return any(keyword in text for keyword in self.report_keywords)

    def _execute_workflow_report(self, topic):
        """执行一个简单的‘搜索->摘要’工作流。"""
        print(f"[工作流触发] 检测到报告请求,主题:'{topic}'")
        
        # 步骤1: 搜索
        print("  -> 步骤1: 执行搜索...")
        search_results = mock_web_search(topic, max_results=5)
        
        # 步骤2: 摘要
        print("  -> 步骤2: 对结果进行摘要...")
        # 这里简单地将所有搜索结果作为文本进行摘要
        summary = summarize_text(search_results, max_length=300)
        
        workflow_result = f"**关于『{topic}』的简报**\n\n" \
                         f"**搜索摘要:**\n{summary}\n\n" \
                         f"**原始搜索结果(供参考):**\n{search_results}"
        return workflow_result

    def chat(self, user_input):
        """与智能体进行对话,支持简单工作流。"""
        # 工作流判断:如果用户输入包含报告关键词,则触发工作流
        if self._contains_report_request(user_input):
            # 尝试从输入中提取主题(这里用简单正则,实际可用更复杂的NLP)
            topic_match = re.search(r'[“”"](.+?)[“”"]|关于(.+?)的|总结(.+?)$', user_input)
            topic = "相关主题"
            if topic_match:
                topic = next((g for g in topic_match.groups() if g), "相关主题").strip()
            
            workflow_output = self._execute_workflow_report(topic)
            # 将工作流输出作为一次“工具调用结果”插入历史,并让模型生成友好回复
            self.conversation_history.append({"role": ChatRole.USER, "content": user_input})
            self.conversation_history.append({
                "role": "tool",
                "content": json.dumps({"workflow_result": workflow_output}, ensure_ascii=False),
                "name": "report_workflow"
            })
            
            # 请求模型基于工作流结果生成最终回复
            req = {
                "model": {"name": self.model_name},
                "messages": self.conversation_history,
                "parameters": {"max_tokens": 1500, "temperature": 0.7}
            }
            try:
                resp = self.service.chat(req)
                final_reply = resp.choice.message.content
                self.conversation_history.append(resp.choice.message.to_dict())
                return final_reply
            except MaasException as e:
                return f"工作流执行后生成回复失败:{e}"
        
        # 非工作流请求,走标准的工具调用流程
        self.conversation_history.append({"role": ChatRole.USER, "content": user_input})
        
        req = {
            "model": {"name": self.model_name},
            "messages": self.conversation_history,
            "tools": self.tools,
            "tool_choice": "auto",
            "parameters": {"max_tokens": 2000, "temperature": 0.7}
        }
        
        try:
            resp = self.service.chat(req)
            assistant_message = resp.choice.message
            self.conversation_history.append(assistant_message.to_dict())
            
            tool_calls = getattr(assistant_message, 'tool_calls', None)
            if tool_calls:
                tool_responses = []
                for tool_call in tool_calls:
                    tool_result = self._execute_tool(tool_call)
                    tool_response_msg = {
                        "role": "tool",
                        "content": json.dumps(tool_result, ensure_ascii=False),
                        "tool_call_id": tool_call.id
                    }
                    tool_responses.append(tool_response_msg)
                
                self.conversation_history.extend(tool_responses)
                
                req_final = {
                    "model": {"name": self.model_name},
                    "messages": self.conversation_history,
                    "parameters": {"max_tokens": 2000, "temperature": 0.7}
                }
                final_resp = self.service.chat(req_final)
                final_message = final_resp.choice.message
                self.conversation_history.append(final_message.to_dict())
                return final_message.content
            else:
                return assistant_message.content
                
        except MaasException as e:
            return f"请求API时发生错误:{e}"
        except Exception as e:
            return f"发生未知错误:{e}"

    def _execute_tool(self, tool_call):
        """执行工具(与之前相同)"""
        function_name = tool_call.function.name
        function_args = json.loads(tool_call.function.arguments)
        print(f"[Agent 调用工具] {function_name},参数:{function_args}")
        if function_name in self.tool_functions:
            result = self.tool_functions[function_name](**function_args)
            return result
        else:
            return f"错误:未知的工具 '{function_name}'。"

    def clear_history(self):
        self.conversation_history = []

5.3 测试多技能与工作流Agent

# 文件:main_advanced.py
from agent_core_advanced import AdvancedInfoAssistant
import os

API_KEY = os.getenv("DOUBAO_API_KEY", "你的_API_KEY")
SECRET_KEY = os.getenv("DOUBAO_SECRET_KEY", "你的_SECRET_KEY")
MODEL_NAME = "ep-20250225141520-ftlpj"

def main():
    assistant = AdvancedInfoAssistant(API_KEY, SECRET_KEY, MODEL_NAME)
    
    print("=== 豆包高级信息助理(多技能+工作流)已启动 ===")
    print("试试以下指令:")
    print("  - '搜索一下Python异步编程'")
    print("  - '帮我生成一份关于大语言模型的简报'")
    print("  - '总结一下刚才的对话'")
    print("输入 'quit' 退出。\n")
    
    test_queries = [
        "搜索一下Python异步编程",
        "帮我生成一份关于大语言模型的简报",
        "你好,今天天气怎么样?",
    ]
    
    for query in test_queries:
        print(f"\n你:{query}")
        print("\n助理:", end="", flush=True)
        response = assistant.chat(query)
        print(response)
        print("-" * 50)
    
    # 交互模式
    assistant.clear_history()
    while True:
        try:
            user_input = input("\n你:")
            if user_input.lower() in ['quit', 'exit', '退出']:
                break
            print("\n助理:", end="", flush=True)
            response = assistant.chat(user_input)
            print(response)
        except KeyboardInterrupt:
            break

if __name__ == "__main__":
    main()

运行这个程序,你会看到当输入包含“简报”时,Agent会触发预设的工作流,自动执行搜索和摘要,并生成格式化的报告。而对于普通问题,它依然会智能地判断是否需要调用搜索或摘要工具。

6. 常见问题、排查与优化

在实际开发和运行中,你可能会遇到一些问题。下面是一些常见场景的排查思路和优化建议。

6.1 常见错误与排查

问题现象 可能原因 排查步骤与解决方案
认证失败 (InvalidAccessKeyId, SignatureDoesNotMatch) 1. API Key或Secret Key错误。
2. Key未启用或权限不足。
3. 请求的Region(地域)与服务地址不匹配。
1. 检查控制台,确认复制的Key无误,且未包含多余空格。
2. 在豆包开放平台控制台检查该应用/Key的状态是否为“启用”。
3. 确认 MaasService 初始化时传入的Region与创建Key时选择的区域一致。
模型不存在 (ModelNotFound) 1. 模型名称填写错误。
2. 该模型在当前区域不可用。
1. 登录控制台,在“模型广场”或“我的模型”中查找正确的模型Endpoint名称。
2. 尝试更换为其他可用模型,或检查区域支持情况。
工具调用不被触发 1. 工具描述不够清晰,模型无法理解何时调用。
2. 用户问题过于简单,模型认为无需工具。
3. tool_choice 参数设置问题。
1. 优化工具描述,明确使用场景和输入输出。
2. 在系统指令中明确要求Agent积极使用工具。
3. 尝试将 tool_choice 设为 "required" 强制测试工具调用(生产环境慎用)。
4. 在对话历史中提供工具调用示例(Few-shot Learning)。
工具调用参数错误 1. 工具的参数JSON Schema定义有误。
2. 模型生成的参数不符合Schema要求。
1. 仔细检查 parameters 的定义,确保类型、必填字段正确。
2. 在 _execute_tool 函数中添加更健壮的参数解析和错误处理。
3. 打印出模型生成的 function.arguments 进行调试。
响应速度慢 1. 网络延迟。
2. 模型推理本身耗时。
3. 进行了多次来回交互(多轮工具调用)。
1. 确保服务器区域选择正确(如国内用户选cn-beijing)。
2. 适当调整 max_tokens ,避免生成过长内容。
3. 对于复杂任务,考虑在应用层设计更高效的工作流,减少与模型的交互轮次。
上下文长度超限 对话历史(含工具调用结果)过长,超过了模型上下文窗口。 1. 定期清理 conversation_history ,只保留最近N轮对话。
2. 对历史消息进行摘要后再存入记忆。
3. 使用平台提供的更高级的Memory管理功能。

6.2 性能与效果优化建议

  1. 系统指令(System Prompt)优化 : 在创建Agent时,可以通过 messages 列表开头插入一个 role system 的消息来设定系统指令。这是控制Agent行为最有效的方式之一。

    messages = [
        {"role": "system", "content": "你是一个专业的信息助理,善于利用搜索和摘要工具获取并整理信息。在回答用户问题时,应优先考虑使用工具来获取最新、最准确的数据。你的回答应简洁、专业、有条理。"},
        {"role": "user", "content": user_input}
    ]
    
  2. 工具描述工程 : 工具的描述 ( description ) 至关重要。描述应清晰说明工具的 用途 适用场景 以及 输入参数的含义 。好的描述能极大提升模型调用工具的准确率。

  3. 结构化输出 : 对于需要后续程序处理的场景,可以要求模型以JSON等特定格式输出。这可以通过在系统指令中说明,或使用豆包平台可能提供的“结构化输出”功能(请查阅最新文档)来实现。

  4. 流式输出 : 对于生成内容较长的场景,可以使用SDK的流式接口,实现打字机效果,提升用户体验。

  5. 错误处理与降级 : 在 _execute_tool 函数中做好异常捕获。当工具调用失败时,应返回明确的错误信息,并让模型有机会尝试其他方法或向用户说明情况。

7. 生产环境部署与安全最佳实践

当你准备将豆包Agent集成到真实业务中时,需要关注以下方面:

7.1 安全与权限管理

  • API Key管理 :切勿将API Key硬编码在代码或前端。应使用环境变量、密钥管理服务(如KMS)或配置文件(并加入.gitignore)。在服务器上设置严格的文件权限。
  • 用量监控与限流 :在豆包开放平台控制台设置用量告警和限流策略,防止意外超支或恶意调用。
  • 输入输出过滤 :对用户的输入进行必要的清洗和过滤,防止Prompt注入攻击。对模型的输出,尤其是涉及外部工具调用(如执行系统命令)时,要进行严格的校验和沙箱隔离。
  • 最小权限原则 :为Agent配置的工具,其背后的API或函数应遵循最小权限原则。例如,一个只读的搜索Agent不应拥有写入数据库的权限。

7.2 部署架构建议

  • 后端服务化 :将Agent核心逻辑封装成RESTful API或gRPC服务,供前端或其他业务系统调用。这有助于解耦、扩展和统一管理。
  • 异步处理 :对于耗时长(如需要多次工具调用)的任务,应采用异步处理模式。接收请求后立即返回一个任务ID,通过WebSocket或轮询告知用户任务进度和结果。
  • 会话状态管理 :对于需要多轮对话的Agent,会话状态( conversation_history )应存储在Redis等外部缓存或数据库中,而不是内存中,以支持多实例部署和会话持久化。
  • 日志与监控 :记录详细的日志,包括用户输入、模型请求/响应、工具调用详情和结果。这便于问题排查、效果分析和成本核算。集成APM工具监控服务性能。

7.3 集成真实工具

将模拟工具替换为真实工具,你的Agent能力将产生质的飞跃。

  1. 替换搜索工具 :注册并获取SerpAPI、Google Custom Search JSON API等服务的Key,替换 mock_web_search 函数。
  2. 添加计算工具 :集成WolframAlpha API进行数学计算。
  3. 添加业务工具 :连接你公司的内部API,让Agent可以查询订单、生成报表、发送通知等。
  4. 使用代码解释器 :豆包平台可能提供代码执行沙箱,可以让Agent编写并运行Python代码来处理数据、绘制图表,这是极其强大的能力。

7.4 成本控制

  • 缓存 :对频繁且结果不变的查询(如“北京的面积”),可以在应用层添加缓存,避免重复调用模型和工具。
  • 精简上下文 :如前所述,管理好对话历史长度。
  • 选择合适模型 :根据任务复杂度选择合适的模型。简单的分类、提取任务可能使用轻量级模型就足够了。
  • 设置预算 :在平台层面为应用设置每日/每月预算上限。

通过本文的教程,你已经掌握了使用豆包Agent开放平台构建功能型智能体的核心流程:从概念理解、环境配置、技能定义、多工具集成到简单工作流编排。我们从一个简单的对话程序开始,逐步升级为一个能自动判断、使用工具、甚至按流程执行任务的“信息助理”。这仅仅是起点,豆包Agent的潜力远不止于此。你可以尝试为其接入更多真实世界的API,设计更复杂的决策逻辑,甚至结合RAG(检索增强生成)技术,打造专属于你业务领域的超级助手。

Logo

北京人形旗下天工造物具身智能开源社区,聚焦具身天工与慧思开物两大平台

更多推荐