👋 欢迎加入我们的 𝕏 (Twitter)、Discord 和 WeChat

安装

从源码安装：

git clone https://github.com/InternLM/lagent.git
cd lagent
pip install -e .

使用

Lagent 的设计理念受到 PyTorch 的启发。我们希望通过类比神经网络层的方式来使工作流程更加清晰直观，这样用户只需专注于以 Pythonic 的方式创建层，并定义它们之间的消息传递。以下是一个简单的教程，帮助你快速上手构建多智能体应用。

模型作为智能体

智能体使用 AgentMessage 进行通信。

from typing import Dict, List
from lagent.agents import Agent
from lagent.schema import AgentMessage
from lagent.llms import VllmModel, INTERNLM2_META

llm = VllmModel(
    path='Qwen/Qwen2-7B-Instruct',
    meta_template=INTERNLM2_META,
    tp=1,
    top_k=1,
    temperature=1.0,
    stop_words=['<|im_end|>'],
    max_new_tokens=1024,
)
system_prompt = '你的回答只能从“典”、“孝”、“急”三个字中选一个。'
agent = Agent(llm, system_prompt)

user_msg = AgentMessage(sender='user', content='今天天气情况')
bot_msg = agent(user_msg)
print(bot_msg)

content='急' sender='Agent' formatted=None extra_info=None type=None receiver=None stream_state=<AgentStatusCode.END: 0>

内存作为状态

每次前向传播时，输入和输出的消息都会被添加到 Agent 的内存中。这一操作是在 __call__ 方法中完成的，而不是在 forward 方法中。以下是伪代码示例：

    def __call__(self, *message):
        message = pre_hooks(message)
        add_memory(message)
        message = self.forward(*message)
        add_memory(message)
        message = post_hooks(message)
        return message

可以通过两种方式检查内存：

memory: List[AgentMessage] = agent.memory.get_memory()
print(memory)
print('-' * 120)
dumped_memory: Dict[str, List[dict]] = agent.state_dict()
print(dumped_memory['memory'])

[AgentMessage(content='今天天气情况', sender='user', formatted=None, extra_info=None, type=None, receiver=None, stream_state=<AgentStatusCode.END: 0>), AgentMessage(content='急', sender='Agent', formatted=None, extra_info=None, type=None, receiver=None, stream_state=<AgentStatusCode.END: 0>)]
------------------------------------------------------------------------------------------------------------------------
[{'content': '今天天气情况', 'sender': 'user', 'formatted': None, 'extra_info': None, 'type': None, 'receiver': None, 'stream_state': <AgentStatusCode.END: 0>}, {'content': '急', 'sender': 'Agent', 'formatted': None, 'extra_info': None, 'type': None, 'receiver': None, 'stream_state': <AgentStatusCode.END: 0>}]

清除当前会话的内存（默认会话 ID 为 0）：

agent.reset()

自定义消息聚合

后台会调用 DefaultAggregator 将 AgentMessage 组装并转换为 OpenAI 的消息格式。

    def forward(self, *message: AgentMessage, session_id=0, **kwargs) -> Union[AgentMessage, str]:
        formatted_messages = self.aggregator.aggregate(
            self.memory.get(session_id),
            self.name,
            self.output_format,
            self.template,
        )
        llm_response = self.llm.chat(formatted_messages, **kwargs)
        ...

实现一个可以接收少量示例的简单聚合器：

from typing import List, Union
from lagent.memory import Memory
from lagent.prompts import StrParser
from lagent.agents.aggregator import DefaultAggregator

class FewshotAggregator(DefaultAggregator):
    def __init__(self, few_shot: List[dict] = None):
        self.few_shot = few_shot or []

    def aggregate(self,
                  messages: Memory,
                  name: str,
                  parser: StrParser = None,
                  system_instruction: Union[str, dict, List[dict]] = None) -> List[dict]:
        _message = []
        if system_instruction:
            _message.extend(
                self.aggregate_system_intruction(system_instruction))
        _message.extend(self.few_shot)
        messages = messages.get_memory()
        for message in messages:
            if message.sender == name:
                _message.append(
                    dict(role='assistant', content=str(message.content)))
            else:
                user_message = message.content
                if len(_message) > 0 and _message[-1]['role'] == 'user':
                    _message[-1]['content'] += user_message
                else:
                    _message.append(dict(role='user', content=user_message))
        return _message

agent = Agent(
    llm,
    aggregator=FewshotAggregator(
        [
            {"role": "user", "content": "今天天气"},
            {"role": "assistant", "content": "【晴】"},
        ]
    )
)
user_msg = AgentMessage(sender='user', content='昨天天气')
bot_msg = agent(user_msg)
print(bot_msg)

content='【多云转晴，夜间有轻微降温】' sender='Agent' formatted=None extra_info=None type=None receiver=None stream_state=<AgentStatusCode.END: 0>

灵活的响应格式化

在 AgentMessage 中，formatted 被保留用于存储由 output_format 从模型输出中解析出的信息。

    def forward(self, *message: AgentMessage, session_id=0, **kwargs) -> Union[AgentMessage, str]:
        ...
        llm_response = self.llm.chat(formatted_messages, **kwargs)
        if self.output_format:
            formatted_messages = self.output_format.parse_response(llm_response)
            return AgentMessage(
                sender=self.name,
                content=llm_response,
                formatted=formatted_messages,
            )
        ...

可以使用工具解析器如下：

from lagent.prompts.parsers import ToolParser

system_prompt = "逐步分析并编写Python代码解决以下问题。"
parser = ToolParser(tool_type='code interpreter', begin='```python\n', end='\n```\n')
llm.gen_params['stop_words'].append('\n```\n')
agent = Agent(llm, system_prompt, output_format=parser)

user_msg = AgentMessage(
    sender='user',
    content='Marie is thinking of a multiple of 63, while Jay is thinking of a '
    'factor of 63. They happen to be thinking of the same number. There are '
    'two possibilities for the number that each of them is thinking of, one '
    'positive and one negative. Find the product of these two numbers.')
bot_msg = agent(user_msg)
print(bot_msg.model_dump_json(indent=4))

{
    "content": "首先，我们需要找出63的所有正因数和负因数。63的正因数可以通过分解63的质因数来找出，即\\(63 = 3^2 \\times 7\\)。因此，63的正因数包括1, 3, 7, 9, 21, 和 63。对于负因数，我们只需将上述正因数乘以-1。\n\n接下来，我们需要找出与63的正因数相乘的结果为63的数，以及与63的负因数相乘的结果为63的数。这可以通过将63除以每个正因数和负因数来实现。\n\n最后，我们将找到的两个数相乘得到最终答案。\n\n下面是Python代码实现：\n\n```python\ndef find_numbers():\n    # 正因数\n    positive_factors = [1, 3, 7, 9, 21, 63]\n    # 负因数\n    negative_factors = [-1, -3, -7, -9, -21, -63]\n    \n    # 找到与正因数相乘的结果为63的数\n    positive_numbers = [63 / factor for factor in positive_factors]\n    # 找到与负因数相乘的结果为63的数\n    negative_numbers = [-63 / factor for factor in negative_factors]\n    \n    # 计算两个数的乘积\n    product = positive_numbers[0] * negative_numbers[0]\n    \n    return product\n\nresult = find_numbers()\nprint(result)",
    "sender": "Agent",
    "formatted": {
        "tool_type": "code interpreter",
        "thought": "首先，我们需要找出63的所有正因数和负因数。63的正因数可以通过分解63的质因数来找出，即\\(63 = 3^2 \\times 7\\)。因此，63的正因数包括1, 3, 7, 9, 21, 和 63。对于负因数，我们只需将上述正因数乘以-1。\n\n接下来，我们需要找出与63的正因数相乘的结果为63的数，以及与63的负因数相乘的结果为63的数。这可以通过将63除以每个正因数和负因数来实现。\n\n最后，我们将找到的两个数相乘得到最终答案。\n\n下面是Python代码实现：\n\n",
        "action": "def find_numbers():\n    # 正因数\n    positive_factors = [1, 3, 7, 9, 21, 63]\n    # 负因数\n    negative_factors = [-1, -3, -7, -9, -21, -63]\n    \n    # 找到与正因数相乘的结果为63的数\n    positive_numbers = [63 / factor for factor in positive_factors]\n    # 找到与负因数相乘的结果为63的数\n    negative_numbers = [-63 / factor for factor in negative_factors]\n    \n    # 计算两个数的乘积\n    product = positive_numbers[0] * negative_numbers[0]\n    \n    return product\n\nresult = find_numbers()\nprint(result)",
        "status": 1
    },
    "extra_info": null,
    "type": null,
    "receiver": null,
    "stream_state": 0
}

工具调用的一致性

ActionExecutor 使用与 Agent 相同的通信数据结构，但要求输入 AgentMessage 的内容是一个包含以下键的字典：

• name: 工具名称，例如 'IPythonInterpreter'、'WebBrowser.search'。
• parameters: 工具 API 的关键字参数，例如 {'command': 'import math;math.sqrt(2)'}、{'query': ['recent progress in AI']}。

你可以注册自定义钩子来进行消息转换。

from lagent.hooks import Hook
from lagent.schema import ActionReturn, ActionStatusCode, AgentMessage
from lagent.actions import ActionExecutor, IPythonInteractive

class CodeProcessor(Hook):
    def before_action(self, executor, message, session_id):
        message = message.copy(deep=True)
        message.content = dict(
            name='IPythonInteractive', parameters={'command': message.formatted['action']}
        )
        return message

    def after_action(self, executor, message, session_id):
        action_return = message.content
        if isinstance(action_return, ActionReturn):
            if action_return.state == ActionStatusCode.SUCCESS:
                response = action_return.format_result()
            else:
                response = action_return.errmsg
        else:
            response = action_return
        message.content = response
        return message

executor = ActionExecutor(actions=[IPythonInteractive()], hooks=[CodeProcessor()])
bot_msg = AgentMessage(
    sender='Agent',
    content='首先，我们需要...',
    formatted={
        'tool_type': 'code interpreter',
        'thought': '首先，我们需要...',
        'action': 'def find_numbers():\n    # 正因数\n    positive_factors = [1, 3, 7, 9, 21, 63]\n    # 负因数\n    negative_factors = [-1, -3, -7, -9, -21, -63]\n    \n    # 找到与正因数相乘的结果为63的数\n    positive_numbers = [63 / factor for factor in positive_factors]\n    # 找到与负因数相乘的结果为63的数\n    negative_numbers = [-63 / factor for factor in negative_factors]\n    \n    # 计算两个数的乘积\n    product = positive_numbers[0] * negative_numbers[0]\n    \n    return product\n\nresult = find_numbers()\nprint(result)',
        'status': 1
    })
executor_msg = executor(bot_msg)
print(executor_msg)

content='3969.0' sender='ActionExecutor' formatted=None extra_info=None type=None receiver=None stream_state=<AgentStatusCode.END: 0>

为了方便起见，Lagent 提供了 InternLMActionProcessor，它适用于上述提到的由 ToolParser 格式化的消息。

双重接口

Lagent 采用了双重接口设计，几乎每个组件（LLMs、actions、action executors 等）都有对应的异步变体，只需在其标识符前加上 Async 即可。建议在调试时使用同步代理，而在大规模推理时使用异步代理，以充分利用空闲的 CPU 和 GPU 资源。

然而，务必确保代理内部的一致性，即异步代理应配备异步 LLM 和驱动异步工具的异步行动执行器。

from lagent.llms import VllmModel, AsyncVllmModel, LMDeployPipeline, AsyncLMDeployPipeline
from lagent.actions import ActionExecutor, AsyncActionExecutor, WebBrowser, AsyncWebBrowser
from lagent.agents import Agent, AsyncAgent, AgentForInternLM, AsyncAgentForInternLM

---

实践

• 除非必要，尽量实现子类的 forward 方法而不是 __call__ 方法。
• 始终显式地包含 session_id 参数，该参数旨在在并发环境中隔离内存、LLM 请求和工具调用（例如维护多个独立的 IPython 环境）。

单一智能体

通过编程解决问题的数学智能体

from lagent.agents.aggregator import InternLMToolAggregator

class Coder(Agent):
    def __init__(self, model_path, system_prompt, max_turn=3):
        super().__init__()
        llm = VllmModel(
            path=model_path,
            meta_template=INTERNLM2_META,
            tp=1,
            top_k=1,
            temperature=1.0,
            stop_words=['\n```\n', '<|im_end|>'],
            max_new_tokens=1024,
        )
        self.agent = Agent(
            llm,
            system_prompt,
            output_format=ToolParser(
                tool_type='code interpreter', begin='```python\n', end='\n```\n'
            ),
            # `InternLMToolAggregator` 是为聚合包含工具调用和执行结果的消息而适配的
            aggregator=InternLMToolAggregator(),
        )
        self.executor = ActionExecutor([IPythonInteractive()], hooks=[CodeProcessor()])
        self.max_turn = max_turn

    def forward(self, message: AgentMessage, session_id=0) -> AgentMessage:
        for _ in range(self.max_turn):
            message = self.agent(message, session_id=session_id)
            if message.formatted['tool_type'] is None:
                return message
            message = self.executor(message, session_id=session_id)
        return message

coder = Coder('Qwen/Qwen2-7B-Instruct', 'Solve the problem step by step with assistance of Python code')
query = AgentMessage(
    sender='user',
    content='Find the projection of $\\mathbf{a}$ onto $\\mathbf{b} = '
    '\\begin{pmatrix} 1 \\\\ -3 \\end{pmatrix}$ if $\\mathbf{a} \\cdot \\mathbf{b} = 2.$'
)
answer = coder(query)
print(answer.content)
print('-' * 120)
for msg in coder.state_dict()['agent.memory']:
    print('*' * 80)
    print(f'{msg["sender"]}:\n\n{msg["content"]}')

多智能体

通过自我改进来提升写作质量的异步博客撰写智能体（原始 AutoGen 示例）

import asyncio
import os
from lagent.llms import AsyncGPTAPI
from lagent.agents import AsyncAgent
os.environ['OPENAI_API_KEY'] = 'YOUR_API_KEY'

class PrefixedMessageHook(Hook):
    def __init__(self, prefix: str, senders: list = None):
        self.prefix = prefix
        self.senders = senders or []

    def before_agent(self, agent, messages, session_id):
        for message in messages:
            if message.sender in self.senders:
                message.content = self.prefix + message.content

class AsyncBlogger(AsyncAgent):
    def __init__(self, model_path, writer_prompt, critic_prompt, critic_prefix='', max_turn=3):
        super().__init__()
        llm = AsyncGPTAPI(model_type=model_path, retry=5, max_new_tokens=2048)
        self.writer = AsyncAgent(llm, writer_prompt, name='writer')
        self.critic = AsyncAgent(
            llm, critic_prompt, name='critic', hooks=[PrefixedMessageHook(critic_prefix, ['writer'])]
        )
        self.max_turn = max_turn

    async def forward(self, message: AgentMessage, session_id=0) -> AgentMessage:
        for _ in range(self.max_turn):
            message = await self.writer(message, session_id=session_id)
            message = await self.critic(message, session_id=session_id)
        return await self.writer(message, session_id=session_id)

blogger = AsyncBlogger(
    'gpt-4o-2024-05-13',
    writer_prompt="You are an writing assistant tasked to write engaging blogpost. You try to generate the best blogpost possible for the user's request. "
    "If the user provides critique, then respond with a revised version of your previous attempts",
    critic_prompt="Generate critique and recommendations on the writing. Provide detailed recommendations, including requests for length, depth, style, etc..",
    critic_prefix='Reflect and provide critique on the following writing. \n\n',
)
user_prompt = (
    "Write an engaging blogpost on the recent updates in {topic}. "
    "The blogpost should be engaging and understandable for general audience. "
    "Should have more than 3 paragraphes but no longer than 1000 words.")
bot_msgs = asyncio.get_event_loop().run_until_complete(
    asyncio.gather(
        *[
            blogger(AgentMessage(sender='user', content=user_prompt.format(topic=topic)), session_id=i)
            for i, topic in enumerate(['AI', 'Biotechnology', 'New Energy', 'Video Games', 'Pop Music'])
        ]
    )
)
print(bot_msgs[0].content)
print('-' * 120)
for msg in blogger.state_dict(session_id=0)['writer.memory']:
    print('*' * 80)
    print(f'{msg["sender"]}:\n\n{msg["content"]}')
print('-' * 120)
for msg in blogger.state_dict(session_id=0)['critic.memory']:
    print('*' * 80)
    print(f'{msg["sender"]}:\n\n{msg["content"]}')

一个执行信息检索、数据收集和图表绘制的多智能体工作流（原始 LangGraph 示例）

import json
from lagent.actions import IPythonInterpreter, WebBrowser, ActionExecutor
from lagent.agents.stream import get_plugin_prompt
from lagent.llms import GPTAPI
from lagent.hooks import InternLMActionProcessor

TOOL_TEMPLATE = (
    "You are a helpful AI assistant, collaborating with other assistants. Use the provided tools to progress"
    " towards answering the question. If you are unable to fully answer, that's OK, another assistant with"
    " different tools will help where you left off. Execute what you can to make progress. If you or any of"
    " the other assistants have the final answer or deliverable, prefix your response with {finish_pattern}"
    " so the team knows to stop. You have access to the following tools:\n{tool_description}\nPlease provide"
    " your thought process when you need to use a tool, followed by the call statement in this format:"
    "\n{invocation_format}\\\\n**{system_prompt}**"
)

class 数据可视化器(代理):
    def __init__(self, 模型路径, 研究提示, 图表提示, 结束模式="最终答案", 最大轮数=10):
        super().__init__()
        llm = GPTAPI(模型路径, 密钥='YOUR_OPENAI_API_KEY', 重试次数=5, 最大生成长度=1024, 停止词=["```\n"])
        解释器, 浏览器 = IPython解释器(), Web浏览器("BingSearch", api_key="YOUR_BING_API_KEY")
        self.研究员 = 代理(
            llm,
            TOOL_TEMPLATE.format(
                结束模式=结束模式,
                工具描述=get_plugin_prompt(浏览器),
                调用格式='```json\n{"name": {{工具名称}}, "parameters": {{关键字参数}}}\n```\n',
                系统提示=研究提示,
            ),
            输出格式=ToolParser(
                "browser",
                开始="```json\n",
                结束="\n```\n",
                验证=lambda x: json.loads(x.rstrip('`')),
            ),
            聚合器=InternLMToolAggregator(),
            名称="researcher",
        )
        self.制图员 = 代理(
            llm,
            TOOL_TEMPLATE.format(
                结束模式=结束模式,
                工具描述=解释器.name,
                调用格式='```python\n{{代码}}\n```\n',
                系统提示=图表提示,
            ),
            输出格式=ToolParser(
                "interpreter",
                开始="```python\n",
                结束="\n```\n",
                验证=lambda x: x.rstrip('`'),
            ),
            聚合器=InternLMToolAggregator(),
            名称="charter",
        )
        self.执行器 = 行动执行器([解释器, 浏览器], 钩子=[InternLMActionProcessor()])
        self.结束模式 = 结束模式
        self.最大轮数 = 最大轮数

    def forward(self, 消息, 会话ID=0):
        for _ in range(self.最大轮数):
            消息 = 自己.研究员(消息, 会话ID=会话ID, 停止词=["```\n", "```python"]) # 覆盖llm的停止词
            while 消息.格式化["工具类型"]:
                消息 = 自己.执行器(消息, 会话ID=会话ID)
                消息 = 自己.研究员(消息, 会话ID=会话ID, 停止词=["```\n", "```python"])
            if 自己.结束模式 在 消息.内容 中:
                return 消息
            消息 = 自己.制图员(消息)
            while 消息.格式化["工具类型"]:
                消息 = 自己.执行器(消息, 会话ID=会话ID)
                消息 = 自己.制图员(消息, 会话ID=会话ID)
            if 自己.结束模式 在 消息.内容 中:
                return 消息
        return 消息

可视化器 = 数据可视化器(
    "gpt-4o-2024-05-13",
    研究提示="您应提供准确的数据供图表生成器使用。",
    图表提示="您展示的任何图表都将对用户可见。",
)
用户消息 = 代理消息(
    发送者='user',
    内容="获取中国过去5年的GDP数据，然后绘制折线图。编码完成后请结束。"
)
机器人消息 = 可视化器(用户消息)
print(机器人消息.内容)
json.dump(可视化器.状态字典(), open('visualizer.json', 'w'), ensure_ascii=False, indent=4)

引用

如果您在研究中发现本项目有用，请考虑引用：

@misc{lagent2023,
    title={{Lagent: InternLM} 是一个轻量级开源框架，允许用户高效构建基于大型语言模型(LLM)的智能体},
    author={Lagent开发者团队},
    howpublished = {\url{https://github.com/InternLM/lagent}},
    year={2023}
}

许可证

本项目采用Apache 2.0许可证发布。

🔼 返回顶部

Lagent 快速上手指南

Lagent 是一个轻量级的开源框架，灵感来源于 PyTorch 的设计理念。它将智能体（Agent）的工作流抽象为类似神经网络的“层”，开发者只需以 Pythonic 的方式创建组件并定义消息传递，即可快速构建多智能体应用。

环境准备

在开始之前，请确保您的开发环境满足以下要求：

• 操作系统: Linux 或 macOS (Windows 用户建议使用 WSL2)
• Python 版本: Python 3.8 及以上
• 前置依赖:
  • git: 用于克隆代码仓库
  • pip: Python 包管理工具
  • (可选) GPU 驱动及 CUDA 环境：如果您计划本地部署大模型并使用 vLLM 加速推理。

安装步骤

推荐通过源码安装，以便获取最新功能并进行二次开发。

# 1. 克隆仓库
git clone https://github.com/InternLM/lagent.git

# 2. 进入项目目录
cd lagent

# 3. 以可编辑模式安装
pip install -e .

提示：国内用户若遇到下载速度慢的问题，可配置 pip 使用国内镜像源（如清华源）： pip install -e . -i https://pypi.tuna.tsinghua.edu.cn/simple

基本使用

Lagent 的核心在于 Agent 类，它封装了大语言模型（LLM）、系统提示词（System Prompt）以及记忆模块。以下是构建一个简单智能体的最小示例。

1. 初始化模型与智能体

首先导入必要的模块，配置 LLM 参数（此处以 Qwen2-7B-Instruct 为例），并实例化 Agent。

from typing import Dict, List
from lagent.agents import Agent
from lagent.schema import AgentMessage
from lagent.llms import VllmModel, INTERNLM2_META

# 初始化大语言模型
llm = VllmModel(
    path='Qwen/Qwen2-7B-Instruct',
    meta_template=INTERNLM2_META,
    tp=1,
    top_k=1,
    temperature=1.0,
    stop_words=['<|im_end|>'],
    max_new_tokens=1024,
)

# 定义系统提示词
system_prompt = '你的回答只能从“典”、“孝”、“急”三个字中选一个。'

# 创建智能体
agent = Agent(llm, system_prompt)

2. 发送消息并获取回复

Lagent 使用 AgentMessage 对象进行通信。调用智能体实例即可处理消息并返回回复。

# 构造用户消息
user_msg = AgentMessage(sender='user', content='今天天气情况')

# 获取智能体回复
bot_msg = agent(user_msg)

# 打印回复内容
print(bot_msg.content)

预期输出：

急

3. 管理会话记忆

Lagent 会自动在每次交互中将输入和输出消息添加到内存中。您可以查看、导出或清空当前会话的记忆。

# 查看内存中的消息列表
memory: List[AgentMessage] = agent.memory.get_memory()
print(memory)

# 清空当前会话记忆 (默认 session_id=0)
agent.reset()

进阶特性简述

• 自定义消息聚合：通过继承 DefaultAggregator，您可以轻松实现 Few-shot（少样本学习）等功能，控制发送给 LLM 的消息格式。
• 灵活的工具调用：内置 ToolParser 可解析模型输出的代码块，配合 ActionExecutor 和 IPythonInteractive 等工具，可实现自动代码执行与结果反馈。
• 双接口支持：核心组件均提供同步和异步两种接口（异步组件名称通常带有特定前缀），方便集成到高并发场景中。