智能体核心技术七大模块之（一）

环境感知与状态表示——构建智能体的感知与内部世界模型

本文是七大模块的第一部分，深入探讨智能体如何“感知”外部环境（用户输入、工具反馈等）并将其转化为结构化的内部表示，同时管理自身的动态状态。我们将通过UML建模、详细设计和完整代码实现，增强基础智能体的感知能力和状态管理能力。

---

1. 核心概念

1.1 环境感知

环境感知是指智能体接收并解析来自外部环境的信息（如用户自然语言、系统反馈、传感器数据等），提取出对决策有用的要素，包括：

• 意图（Intent）：用户想要达成的目标（如查询天气、发送邮件）。
• 实体（Entity）/参数：实现意图所需的关键信息（如地点、时间、收件人）。
• 上下文线索：指代、省略等需要结合历史的信息。

1.2 状态表示

状态表示是智能体对当前内部情况的建模，包括：

• 短期记忆：当前对话轮次的历史消息。
• 长期记忆：用户偏好、知识库、学习到的经验。
• 任务状态：当前正在执行的计划、已完成步骤、待办步骤、中间结果。
• 环境快照：工具执行结果、外部数据等。

一个良好的状态表示应具备：

• 结构化：便于推理模块查询和更新。
• 可扩展：能容纳多种类型的信息。
• 一致性：确保状态在不同模块间正确传递。

---

2. 系统架构设计

2.1 新增核心类

在原有架构基础上，我们引入以下关键类：

• ParsedInput：感知模块的输出，包含意图、槽位、原始文本等。
• State：表示智能体的完整内部状态，聚合记忆、任务进度、环境信息。
• NLUEngine（可选）：基于规则或模型的意图/实体解析器，作为感知模块的组件。

2.2 类图

2.3 状态流转时序图

以下是一个典型任务中，感知与状态模块如何交互的时序图：

---

3. 详细设计

3.1 感知模块增强

原有的Perception基于简单规则，现在将其重构为可插拔的架构，支持多种NLU引擎：

• RuleBasedNLU：使用正则表达式或词典匹配，适合简单、固定的场景。
• LLMBasedNLU：利用LLM进行意图识别和槽位填充，适合复杂、开放域。

Perception类不再直接实现解析逻辑，而是持有NLUEngine实例，将解析任务委托给引擎。

3.2 状态模块设计

State类负责维护智能体的所有内部状态，并提供查询和更新方法。其属性包括：

• memory：Memory对象，管理对话历史。
• current_plan：当前待执行的计划步骤列表。
• completed_steps：已完成的步骤列表。
• intermediate_results：工具执行产生的中间结果，键为步骤名或自定义键。
• environment：环境信息，如当前时间、用户位置等（可动态获取）。

主要方法：

• update_after_tool(tool_name, result)：执行工具后，将结果存入intermediate_results，并可能根据结果更新计划。
• update_plan(new_plan)：替换或追加新计划。
• get_context()：返回供推理模块使用的状态摘要（文本形式）。
• add_user_message(text)和add_assistant_message(text)：向记忆中添加消息。

3.3 与原有模块的集成

• Agent类现在包含一个State实例，而非直接持有Memory。Memory作为State的一部分。
• Agent.run循环中，每次迭代前从State获取上下文，传递给推理模块；执行工具后，通过State.update_after_tool更新状态。

---

4. 项目结构更新

在原有项目基础上，新增或修改以下文件：

agent_core/
├── agent/
│   ├── core/
│   │   ├── __init__.py
│   │   ├── agent.py          # 修改：使用State
│   │   ├── perception.py      # 修改：支持NLU引擎
│   │   ├── state.py           # 新增：状态类
│   │   ├── memory.py          # 保留，但作为State内部使用
│   │   ├── models.py          # 新增ParsedInput等
│   │   └── ... (其他不变)
│   ├── nlu/                   # 新增：NLU引擎模块
│   │   ├── __init__.py
│   │   ├── base.py            # NLUEngine抽象基类
│   │   ├── rule_based.py      # 基于规则的实现
│   │   └── llm_based.py       # 基于LLM的实现
│   ├── tools/... (不变)
│   ├── llm/... (不变)
│   └── utils/... (不变)
├── examples/
│   └── enhanced_agent.py      # 新示例，展示增强感知和状态
└── ... (其余不变)

---

5. 源代码完整实现

下面提供增强后的核心模块代码，重点关注感知和状态部分的实现。

5.1 数据模型 (agent/core/models.py)

from dataclasses import dataclass, field
from typing import Optional, Dict, Any, List

@dataclass
class ParsedInput:
    """感知模块的输出"""
    raw_text: str
    intent: str
    slots: Dict[str, Any] = field(default_factory=dict)
    confidence: float = 1.0

@dataclass
class Message:
    role: str  # 'user', 'assistant', 'system', 'tool'
    content: str
    metadata: Optional[Dict] = None

# 其他已有类（UserRequest可废弃，或兼容）
# 为简化，我们继续使用UserRequest，但建议逐步迁移到ParsedInput
@dataclass
class UserRequest:
    raw_input: str
    intent: str
    params: Dict[str, Any]

5.2 NLU引擎抽象 (agent/nlu/base.py)

from abc import ABC, abstractmethod
from agent.core.models import ParsedInput

class NLUEngine(ABC):
    @abstractmethod
    def understand(self, text: str) -> ParsedInput:
        """将文本解析为结构化输入"""
        pass

5.3 基于规则的NLU (agent/nlu/rule_based.py)

import re
from .base import NLUEngine
from agent.core.models import ParsedInput

class RuleBasedNLU(NLUEngine):
    """基于正则表达式的简单NLU"""
    def understand(self, text: str) -> ParsedInput:
        text_lower = text.lower()
        # 意图识别与槽位填充
        if "发送邮件" in text_lower or "send email" in text_lower:
            intent = "send_email"
            match_to = re.search(r"(给|to[:：]?\s*)([^，\s]+)", text)
            match_content = re.search(r"说(.*?)(?=$|[,，])", text)
            to = match_to.group(2) if match_to else "unknown"
            content = match_content.group(1) if match_content else ""
            slots = {"to": to, "content": content}
        elif "天气" in text_lower or "weather" in text_lower:
            intent = "query_weather"
            match_city = re.search(r"(天气|weather)(?:.*?)([^，\s]+)", text)
            city = match_city.group(2) if match_city else "北京"
            slots = {"city": city}
        elif "计算" in text_lower or "calculate" in text_lower:
            intent = "calculate"
            expr = text.replace("计算", "").replace("calculate", "").strip()
            slots = {"expression": expr}
        else:
            intent = "chat"
            slots = {}
        return ParsedInput(raw_text=text, intent=intent, slots=slots)

5.4 基于LLM的NLU (agent/nlu/llm_based.py)

import json
from .base import NLUEngine
from agent.core.models import ParsedInput
from agent.llm.base import BaseLLM

class LLMBasedNLU(NLUEngine):
    """使用LLM进行意图识别和槽位填充"""
    def __init__(self, llm: BaseLLM):
        self.llm = llm

    def understand(self, text: str) -> ParsedInput:
        prompt = f"""请解析用户输入，返回JSON格式的意图和槽位。
用户输入：{text}
可选的意图：send_email, query_weather, calculate, chat
对于send_email，槽位包括：to（收件人姓名或邮箱）、content（内容）
对于query_weather，槽位包括：city（城市）
对于calculate，槽位包括：expression（数学表达式）
对于chat，无需槽位。
请只返回JSON，例如：{{"intent": "query_weather", "slots": {{"city": "北京"}}}}
"""
        response = self.llm.generate(prompt)
        try:
            # 尝试从response中提取JSON
            # 这里简单假设response就是JSON字符串
            data = json.loads(response)
            intent = data.get("intent", "chat")
            slots = data.get("slots", {})
        except:
            # 解析失败，回退到chat
            intent = "chat"
            slots = {}
        return ParsedInput(raw_text=text, intent=intent, slots=slots)

5.5 感知模块 (agent/core/perception.py)

from agent.nlu.base import NLUEngine
from agent.core.models import ParsedInput, UserRequest  # 暂时兼容UserRequest

class Perception:
    """感知模块，使用指定的NLU引擎解析输入"""
    def __init__(self, nlu_engine: NLUEngine):
        self.nlu_engine = nlu_engine

    def parse(self, user_input: str) -> ParsedInput:
        return self.nlu_engine.understand(user_input)

    # 可选：保持向后兼容的UserRequest转换方法
    def to_user_request(self, parsed: ParsedInput) -> UserRequest:
        return UserRequest(
            raw_input=parsed.raw_text,
            intent=parsed.intent,
            params=parsed.slots
        )

5.6 状态模块 (agent/core/state.py)

from typing import List, Dict, Any, Optional
from .memory import Memory
from .models import Message

class State:
    """智能体内部状态"""
    def __init__(self):
        self.memory = Memory()
        self.current_plan: List[str] = []          # 待执行计划步骤
        self.completed_steps: List[str] = []       # 已完成的步骤
        self.intermediate_results: Dict[str, Any] = {}  # 步骤结果缓存
        self.environment: Dict[str, Any] = {}       # 环境信息（可动态更新）

    def add_user_message(self, content: str):
        self.memory.add_message(Message(role="user", content=content))

    def add_assistant_message(self, content: str):
        self.memory.add_message(Message(role="assistant", content=content))

    def add_system_message(self, content: str):
        self.memory.add_message(Message(role="system", content=content))

    def add_tool_message(self, tool_name: str, result: str):
        self.memory.add_message(Message(role="tool", content=f"{tool_name}: {result}"))

    def set_plan(self, plan: List[str]):
        self.current_plan = plan
        self.completed_steps = []

    def step_completed(self, step: str, result: Any = None):
        """标记一个步骤完成，并记录结果"""
        if step in self.current_plan:
            self.current_plan.remove(step)
            self.completed_steps.append(step)
            if result is not None:
                self.intermediate_results[step] = result

    def update_after_tool(self, tool_name: str, result: str, step: Optional[str] = None):
        """工具执行后的状态更新"""
        self.add_tool_message(tool_name, result)
        if step:
            self.step_completed(step, result)
        # 也可以根据result内容动态调整计划，此处简化

    def get_context(self) -> str:
        """生成供推理模块使用的上下文文本"""
        # 组合近期记忆、计划、已完成步骤、中间结果
        recent_msgs = self.memory.get_context()[-5:]  # 最近5条
        history = "\n".join([f"{m.role}: {m.content}" for m in recent_msgs])
        plan_status = f"当前计划: {self.current_plan}\n已完成: {self.completed_steps}"
        intermediates = f"中间结果: {self.intermediate_results}"
        return f"{history}\n\n{plan_status}\n{intermediates}"

    def get_short_term_memory(self) -> List[Message]:
        return self.memory.short_term

5.7 修改主控制器 Agent (agent/core/agent.py)

from typing import Dict
from .perception import Perception
from .planner import Planner
from .reasoning import Reasoning
from .executor import Executor
from .state import State
from .reflector import Reflector
from .models import Action, Message
from agent.core.tool import Tool
from agent.nlu.base import NLUEngine  # 新增

class Agent:
    def __init__(self, nlu_engine: NLUEngine, llm):
        # 感知模块现在需要NLU引擎
        self.perception = Perception(nlu_engine)
        self.planner = Planner()
        self.reasoning = Reasoning(llm)
        self.executor = Executor()
        self.state = State()          # 使用State代替直接Memory
        self.reflector = Reflector()
        self.tools: Dict[str, Tool] = {}
        self.max_iterations = 10

    def register_tool(self, tool: Tool):
        self.tools[tool.name] = tool

    def run(self, user_input: str) -> str:
        # 1. 感知
        parsed = self.perception.parse(user_input)
        self.state.add_user_message(user_input)

        # 2. 规划（可选）
        # 将ParsedInput转换为旧格式以兼容现有Planner，后续可升级Planner
        request = self.perception.to_user_request(parsed)
        plan = self.planner.create_plan(request, self.tools)
        if plan:
            self.state.set_plan(plan)
            self.state.add_system_message(f"计划步骤: {plan}")

        # 3. ReAct循环
        iteration = 0
        final_answer = None
        while iteration < self.max_iterations:
            iteration += 1
            # 从状态获取上下文
            context = self.state.get_context()
            action = self.reasoning.decide(context, self.state.get_short_term_memory(), self.tools)

            if action.type == 'final':
                final_answer = action.content
                self.state.add_assistant_message(final_answer)
                break
            elif action.type == 'tool':
                obs = self.executor.execute(action, self.tools)
                # 更新状态，传入当前步骤（如果有）
                # 简单假设action.tool对应计划中的一个步骤，实际需更精细映射
                self.state.update_after_tool(action.tool, obs.result if not obs.error else obs.error)
                if obs.error:
                    # 可以记录错误并尝试重规划
                    pass
            else:
                if action.content:
                    self.state.add_assistant_message(f"思考: {action.content}")

        if not final_answer:
            final_answer = "抱歉，无法完成您的请求。"

        # 4. 反思（可异步）
        reflection = self.reflector.reflect(self.state.get_short_term_memory())
        return final_answer

5.8 规划模块微调 (agent/core/planner.py)

为了兼容ParsedInput，可稍作修改，但这里暂不改变，仍使用UserRequest。后续可以升级Planner直接接受ParsedInput。

5.9 运行示例 (examples/enhanced_agent.py)

创建新示例，使用基于规则的NLU和模拟LLM，展示增强感知和状态管理。

import sys
sys.path.append("..")

from agent.core.agent import Agent
from agent.nlu.rule_based import RuleBasedNLU
from agent.llm.mock import MockLLM
from agent.tools.calculator import CalculatorTool
from agent.tools.weather import WeatherTool
from agent.tools.email import EmailTool

def main():
    # 使用规则NLU和模拟LLM
    nlu = RuleBasedNLU()
    llm = MockLLM()
    agent = Agent(nlu, llm)
    agent.register_tool(CalculatorTool())
    agent.register_tool(WeatherTool())
    agent.register_tool(EmailTool())

    print("增强型智能体已启动（使用规则NLU）")
    while True:
        user_input = input("\n用户: ")
        if user_input.lower() == 'quit':
            break
        response = agent.run(user_input)
        print(f"助手: {response}")

if __name__ == "__main__":
    main()

运行效果应与之前类似，但内部状态管理更清晰。

5.10 可选：基于LLM的NLU示例

如果想尝试LLM-based NLU，只需将RuleBasedNLU替换为LLMBasedNLU(llm)，注意传入的llm需要是BaseLLM实例。但需要注意，模拟LLM可能无法返回有效的JSON，因此需要更强大的LLM（如OpenAI）。这里提供一个使用OpenAI LLM的示例片段：

from agent.llm.openai import OpenAILLM
from agent.nlu.llm_based import LLMBasedNLU

openai_llm = OpenAILLM(model="gpt-3.5-turbo")
nlu = LLMBasedNLU(openai_llm)
agent = Agent(nlu, openai_llm)  # 推理也使用同一个LLM

---

6. 总结与扩展

通过本文，我们深入探讨了智能体的环境感知与状态表示模块，并实现了以下增强：

• 模块化感知：引入NLUEngine抽象，支持规则和LLM两种解析方式。
• 结构化状态：新增State类，统一管理记忆、计划、中间结果，使智能体内部更清晰。
• 无缝集成：保持与原有模块的兼容，仅需少量修改即可升级。

这些改进为后续模块（规划、推理、反思）提供了更丰富、更一致的信息基础。你可以根据实际需求，进一步扩展State，例如增加长期记忆的持久化、环境感知的实时更新等。

掌握环境感知与状态表示，意味着智能体能够更准确地理解外部世界，并维护一个动态的内部模型，这是实现复杂任务执行的关键一步。

---

附录：项目文件清单

agent_core/
├── agent/
│   ├── __init__.py
│   ├── core/
│   │   ├── __init__.py
│   │   ├── agent.py          # 更新：使用State
│   │   ├── perception.py      # 更新：依赖NLU引擎
│   │   ├── state.py           # 新增
│   │   ├── memory.py          # 保留
│   │   ├── models.py          # 新增ParsedInput
│   │   ├── planner.py         # 保留（可后续优化）
│   │   ├── reasoning.py       # 保留
│   │   ├── executor.py        # 保留
│   │   ├── reflector.py       # 保留
│   │   └── tool.py            # 保留
│   ├── nlu/                   # 新增NLU模块
│   │   ├── __init__.py
│   │   ├── base.py
│   │   ├── rule_based.py
│   │   └── llm_based.py
│   ├── tools/... (不变)
│   ├── llm/... (不变)
│   └── utils/... (不变)
├── examples/
│   └── enhanced_agent.py      # 新示例
└── ...

现在，你已经拥有一个具备更强大感知和状态管理能力的智能体核心。继续学习后续模块，你将能构建出更智能、更自主的执行专家！