今天我们来深入探讨一个技术话题：如何用30行代码实现AI Agent的核心功能。在当前AI技术快速发展的背景下，Agent（智能代理）已经成为连接大语言模型与现实世界应用的关键桥梁，但很多开发者对其底层实现机制仍感到神秘。

这个30行代码的实现方案，重点不是追求功能的完整性，而是通过最精简的代码揭示Agent的核心工作原理。我们将从消息处理、工具调用、结果返回等关键环节入手，让你在短时间内掌握Agent的基本架构。

1. Agent核心能力速览

能力项 说明
核心功能 消息处理、工具调用、结果返回
代码规模 约30行Python代码
依赖库 主要使用标准库，轻量级实现
学习价值 理解Agent底层消息流转机制
适用场景 教学演示、原型验证、核心逻辑理解

2. Agent技术背景与价值

AI Agent的核心价值在于能够将大语言模型的推理能力与实际工具操作相结合。从网络搜索材料中我们可以看到，Claude等主流AI平台已经将Agent作为重要功能模块，支持通过tool_use和tool_result块实现复杂的工具调用流程。

在实际应用中，Agent需要处理的消息包含文本、图像、tool_use和tool_result等多种类型的块。用户消息通常包含客户端内容和tool_result，而助手消息则包含tool_use指令。这种结构化的消息处理机制是Agent能够完成复杂任务的技术基础。

3. 环境准备与依赖检查

由于我们采用极简实现方案，环境要求非常宽松：

系统要求：

  • Python 3.7+
  • 无需GPU，纯CPU运行
  • 任意操作系统（Windows/macOS/Linux）

依赖库检查： 

# 检查Python版本
python --version

# 验证必要库是否存在（通常为标准库）
python -c "import json, re, random"

如果上述命令没有报错，说明环境已经就绪。我们的实现主要依赖Python标准库，避免了复杂的外部依赖。

4. Agent核心代码实现

下面是完整的30行Agent核心实现：

class SimpleAgent:
    def __init__(self):
        self.tools = {
            'calculate': self.calculate,
            'search': self.search,
            'format': self.format_text
        }
    
    def process_message(self, message):
        if 'tool_use' in message:
            tool_name = message['tool_use']['name']
            tool_args = message['tool_use']['arguments']
            return self.use_tool(tool_name, tool_args)
        return {'text': f"Received: {message}"}
    
    def use_tool(self, tool_name, arguments):
        if tool_name in self.tools:
            result = self.tools[tool_name](**arguments)
            return {'tool_result': {'name': tool_name, 'content': result}}
        return {'text': f"Tool {tool_name} not found"}
    
    def calculate(self, expression):
        return eval(expression)  # 简单实现，生产环境需要安全处理
    
    def search(self, query):
        return f"Search results for: {query}"
    
    def format_text(self, text, style="markdown"):
        return f"[{style.upper()}]: {text}"

# 使用示例
agent = SimpleAgent()
message = {'tool_use': {'name': 'calculate', 'arguments': {'expression': '2+3*4'}}}
result = agent.process_message(message)
print(result)

5. 代码结构深度解析

5.1 消息处理核心逻辑

process_message 方法是整个Agent的调度中心，负责识别消息类型并分发给相应的处理模块。关键判断逻辑基于消息中是否包含 tool_use 字段，这与主流AI平台的消息结构保持一致。

def process_message(self, message):
    # 识别工具调用请求
    if 'tool_use' in message:
        tool_name = message['tool_use']['name']
        tool_args = message['tool_use']['arguments']
        return self.use_tool(tool_name, tool_args)
    # 处理普通文本消息
    return {'text': f"Received: {message}"}

5.2 工具注册与调用机制

工具字典 self.tools 实现了灵活的插件架构，支持动态注册新工具。这种设计使得Agent可以轻松扩展功能。

def use_tool(self, tool_name, arguments):
    # 检查工具是否存在
    if tool_name in self.tools:
        # 执行工具调用
        result = self.tools[tool_name](**arguments)
        return {'tool_result': {'name': tool_name, 'content': result}}
    return {'text': f"Tool {tool_name} not found"}

5.3 工具函数实现示例

每个工具函数都遵循统一的接口规范，接收参数并返回处理结果。在实际应用中，这些工具可以替换为任何复杂的功能实现。

def calculate(self, expression):
    # 注意：生产环境需要更安全表达式求值
    return eval(expression)

def search(self, query):
    # 模拟搜索功能，实际可接入真实搜索引擎
    return f"Search results for: {query}"

6. 功能测试与验证流程

6.1 基础消息处理测试

首先测试Agent对普通文本消息的处理能力：

# 测试普通消息处理
agent = SimpleAgent()
text_message = {"text": "Hello, Agent!"}
result = agent.process_message(text_message)
print(result)  # 输出: {'text': 'Received: Hello, Agent!'}

6.2 工具调用功能测试

验证工具调用的完整流程：

# 测试计算工具
calc_message = {
    'tool_use': {
        'name': 'calculate', 
        'arguments': {'expression': '2+3*4'}
    }
}
calc_result = agent.process_message(calc_message)
print(calc_result)  # 输出: {'tool_result': {'name': 'calculate', 'content': 14}}

# 测试搜索工具
search_message = {
    'tool_use': {
        'name': 'search',
        'arguments': {'query': 'AI Agent development'}
    }
}
search_result = agent.process_message(search_message)
print(search_result)

6.3 错误处理测试

验证Agent对异常情况的处理能力：

# 测试不存在的工具
invalid_tool_message = {
    'tool_use': {
        'name': 'unknown_tool',
        'arguments': {}
    }
}
error_result = agent.process_message(invalid_tool_message)
print(error_result)  # 输出: {'text': 'Tool unknown_tool not found'}

7. 扩展与进阶实现

7.1 支持多轮对话上下文

基础版本可以扩展为支持对话历史的完整Agent：

class AdvancedAgent(SimpleAgent):
    def __init__(self):
        super().__init__()
        self.conversation_history = []
    
    def process_message(self, message):
        self.conversation_history.append(message)
        
        # 基于对话历史进行决策
        if len(self.conversation_history) > 1:
            # 实现基于上下文的智能决策
            pass
        
        return super().process_message(message)

7.2 集成真实外部工具

将模拟工具替换为真实API调用：

def real_search(self, query, api_key=None):
    # 集成真实搜索引擎API
    # 注意：需要处理网络请求、错误重试、速率限制等
    import requests
    # 实际API调用代码
    return "Real search results"

7.3 添加安全控制机制

在生产环境中需要添加安全控制：

def safe_calculate(self, expression):
    # 限制可用的数学函数和操作符
    allowed_chars = set('0123456789+-*/.() ')
    if all(c in allowed_chars for c in expression):
        return eval(expression)
    else:
        return "Error: Invalid expression"

8. 与主流平台对比分析

8.1 消息结构兼容性

我们的实现与Claude等平台的消息结构保持兼容：

# Claude风格的消息结构
claude_style_message = {
    "messages": [
        {
            "role": "user",
            "content": "Calculate 2+3*4",
            "tool_use": {
                "name": "calculate",
                "arguments": {"expression": "2+3*4"}
            }
        }
    ]
}

8.2 工具调用流程对比

与专业AI平台相比，我们的简化实现保留了核心流程：

  1. 消息解析 ：识别工具调用意图
  2. 参数提取 ：获取工具名称和参数
  3. 工具执行 ：调用注册的工具函数
  4. 结果返回 ：格式化工具执行结果

9. 实际应用场景演示

9.1 自动化数据处理流程

将Agent集成到数据处理流水线中：

def data_processing_agent():
    agent = SimpleAgent()
    
    # 模拟数据处理流程
    processing_steps = [
        {'tool_use': {'name': 'format', 'arguments': {'text': '原始数据', 'style': 'json'}}},
        {'tool_use': {'name': 'calculate', 'arguments': {'expression': 'sum([1,2,3,4,5])'}}}
    ]
    
    for step in processing_steps:
        result = agent.process_message(step)
        print(f"Step result: {result}")

9.2 多工具协同工作

演示多个工具的组合使用：

def multi_tool_workflow():
    agent = SimpleAgent()
    
    # 复杂工作流：搜索→计算→格式化
    workflow = [
        {'tool_use': {'name': 'search', 'arguments': {'query': 'sales data'}}},
        {'tool_use': {'name': 'calculate', 'arguments': {'expression': '1000 * 0.15'}}},
        {'tool_use': {'name': 'format', 'arguments': {'text': '最终结果', 'style': 'markdown'}}}
    ]
    
    for step in workflow:
        result = agent.process_message(step)
        print(f"Workflow step: {result}")

10. 性能优化与生产级考虑

10.1 异步处理支持

对于需要长时间运行的工具，添加异步支持：

import asyncio

class AsyncAgent(SimpleAgent):
    async def process_message_async(self, message):
        # 异步处理消息
        if 'tool_use' in message:
            tool_name = message['tool_use']['name']
            tool_args = message['tool_use']['arguments']
            return await self.use_tool_async(tool_name, tool_args)
        return {'text': f"Received: {message}"}
    
    async def use_tool_async(self, tool_name, arguments):
        # 模拟异步工具调用
        await asyncio.sleep(0.1)  # 模拟IO操作
        return await super().use_tool(tool_name, arguments)

10.2 错误处理与重试机制

添加生产环境必需的健壮性功能：

def robust_tool_call(self, tool_name, arguments, max_retries=3):
    for attempt in range(max_retries):
        try:
            result = self.tools[tool_name](**arguments)
            return {'tool_result': {'name': tool_name, 'content': result}}
        except Exception as e:
            if attempt == max_retries - 1:
                return {'error': f"Tool {tool_name} failed after {max_retries} attempts: {str(e)}"}
            # 指数退避重试
            time.sleep(2 ** attempt)

11. 常见问题与解决方案

11.1 工具注册失败

问题现象 ：新工具无法正确注册到Agent中

解决方案 ：

# 确保工具函数签名正确
def custom_tool(self, param1, param2):
    # 工具函数必须接收明确的参数
    return f"Processed: {param1}, {param2}"

# 正确注册工具
agent.tools['custom_tool'] = agent.custom_tool

11.2 消息格式错误

问题现象 ：Agent无法正确解析输入消息

解决方案 ：

def validate_message_format(self, message):
    required_fields = ['tool_use']  # 根据消息类型调整
    if not any(field in message for field in required_fields):
        return {'error': 'Invalid message format'}
    return None

11.3 工具执行超时

问题现象 ：某些工具执行时间过长

解决方案 ：

import signal

def timeout_handler(signum, frame):
    raise TimeoutError("Tool execution timeout")

def execute_with_timeout(tool_func, args, timeout=30):
    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(timeout)
    try:
        result = tool_func(**args)
        signal.alarm(0)  # 取消定时器
        return result
    except TimeoutError:
        return "Tool execution timeout"

12. 进阶扩展方向

12.1 集成大语言模型

将Agent与LLM结合，实现智能工具选择：

class LLMEnhancedAgent(SimpleAgent):
    def __init__(self, llm_client):
        super().__init__()
        self.llm = llm_client
    
    def intelligent_tool_selection(self, user_query):
        # 使用LLM分析用户意图，选择合适的工具
        analysis_prompt = f"分析用户请求'{user_query}'，选择最合适的工具：calculate、search、format"
        tool_decision = self.llm.analyze(analysis_prompt)
        return self.map_tool_from_decision(tool_decision)

12.2 支持复杂工作流

实现多步骤的复杂任务处理：

class WorkflowAgent(SimpleAgent):
    def execute_workflow(self, workflow_definition):
        results = []
        for step in workflow_definition['steps']:
            step_result = self.process_message(step)
            results.append(step_result)
            
            # 根据上一步结果决定下一步行动
            if step.get('condition'):
                if not self.evaluate_condition(step['condition'], step_result):
                    break
        
        return {'workflow_results': results}

这个30行代码的Agent核心实现虽然简洁，但完整展示了智能代理的基本架构和消息处理流程。通过逐步扩展，你可以基于这个核心构建出功能完整的生产级AI Agent系统。

# AI Agent
Logo
码道开发者社区

码道开发者社区，聚焦华为云码道 CodeArts 代码智能体，沉淀 Agent、Skill、鸿蒙开发实战内容，供开发者查阅资料、交流技术、分享工程实践

更多推荐

cover

Claude API + Python实战：结构化输出后，如何拦住“看起来正确”的JSON

avatar 码道开发者社区
cover

GLM-5.3-Flash深度解析：3200亿参数只卖Opus四十分之一，国产算力首次跑通前沿模型

avatar 码道开发者社区
cover

AI Agent的应用场景

avatar 码道开发者社区