糖豆家

Gitee 仓库地址

什么是 Deep Agents？

Deep Agents 是一个基于 LangGraph 构建的先进智能体框架，提供以下功能：

规划能力 - 使用 write_todos 工具分解复杂任务
文件系统后端 - 通过文件操作保存和检索上下文
子智能体生成 - 将专业任务委托给专注型智能体处理
上下文管理 - 防止复杂任务中出现上下文窗口溢出

Deep Agent 工作原理

架构

用户问题
     ↓
Deep Agent（带规划功能）
     ├─ write_todos（规划方法）
     ├─ SQL 工具
     │  ├─ list_tables
     │  ├─ get_schema
     │  ├─ query_checker
     │  └─ execute_query
     ├─ 文件系统工具（可选）
     │  ├─ ls
     │  ├─ read_file
     │  ├─ write_file
     │  └─ edit_file
     └─ 子智能体生成（可选）
     ↓
SQLite 数据库（Chinook）
     ↓
格式化答案

配置

Deep Agents 使用基于渐进式披露的记忆文件和技能：

AGENTS.md（始终加载）包含：

智能体身份和角色
核心原则和安全规则
通用指南
沟通风格

skills/（按需加载）包含专业工作流：

query-writing - 如何编写和执行 SQL 查询（简单和复杂）
schema-exploration - 如何发现数据库结构和关系

智能体在其上下文中看到技能描述，但仅在确定当前任务需要哪个技能时，才加载完整的 SKILL.md 指令。这种渐进式披露模式既保持了上下文的效率，又在需要时提供了深入的专业知识。

查询示例

简单查询

"有多少客户来自加拿大？"

智能体将直接查询并返回数量。

带规划功能的复杂查询

"哪位员工在哪些国家创造了最多收入？"

智能体将：

使用 write_todos 来规划方法
识别所需的表（Employee、Invoice、Customer）
规划 JOIN 结构
执行查询
用分析结果格式化输出

推演过程

Deep Agent 会展示其推理过程：

问题：哪位员工在哪些国家创造了最多收入？

[规划步骤]
使用 write_todos：
- [ ] 列出数据库中的表
- [ ] 检查 Employee 和 Invoice 的结构
- [ ] 规划多表 JOIN 查询
- [ ] 按员工和国家进行聚合执行
- [ ] 格式化结果

[执行步骤]
1. 正在列出表...
2. 获取结构：Employee、Invoice、InvoiceLine、Customer
3. 生成 SQL 查询...
4. 执行查询...
5. 格式化结果...

[最终答案]
员工 Jane Peacock（ID: 3）创造了最多收入...
排名前列的国家：美国（$1000）、加拿大（$500）...

Deep Agent 关键点

1、create_deep_agent

python

D:\1sqyai\text-to-sql-agent\agent.py
from deepagents import create_deep_agent
# Create the Deep Agent with all parameters
agent = create_deep_agent(
	model=model, #Minimax or Ollama
	system_prompt=["./AGENTS.md"],  #人格规则设定
	tools=sql_tools,  # SQL database tools
	memory=["./memories/'%Y-%m-%d'.md"],  #当天的记忆
	skills=["./skills/"],  #存放技能的目录	
	backend=FilesystemBackend(root_dir=base_dir),  # Persistent file storage
)

.venv\Lib\site-packages\deepagents\graph.py
def create_deep_agent(  # noqa: C901, PLR0912, PLR0915  # Complex graph assembly logic with many conditional branches
    model: str | BaseChatModel | None = None,
    tools: Sequence[BaseTool | Callable | dict[str, Any]] | None = None,
    *,
    system_prompt: str | SystemMessage | None = None,
    middleware: Sequence[AgentMiddleware] = (),
    subagents: Sequence[SubAgent | CompiledSubAgent | AsyncSubAgent] | None = None,
    skills: list[str] | None = None,
    memory: list[str] | None = None,
    permissions: list[FilesystemPermission] | None = None,
    backend: BackendProtocol | BackendFactory | None = None,
    interrupt_on: dict[str, bool | InterruptOnConfig] | None = None,
    response_format: ResponseFormat[ResponseT] | type[ResponseT] | dict[str, Any] | None = None,
    state_schema: type[DeepAgentState] | None = None,
    context_schema: type[ContextT] | None = None,
    checkpointer: Checkpointer | None = None,
    store: BaseStore | None = None,
    debug: bool = False,
    name: str | None = None,
    cache: BaseCache | None = None,
)

2、tools

工具 1：sql_tools

python


D:\1sqyai\text-to-sql-agent\agent.py
toolkit = SQLDatabaseToolkit(db=db, llm=model)
sql_tools = toolkit.get_tools()

.venv\Lib\site-packages\langchain_community\agent_toolkits\sql\toolkit.py
toolkit -> SQLDatabaseToolkit()
from langchain_community.tools.sql_database.tool import (
    InfoSQLDatabaseTool, #sql_db_schema
    ListSQLDatabaseTool, #sql_db_list_tables
    QuerySQLCheckerTool, #sql_db_query_checker
    QuerySQLDatabaseTool, #sql_db_query
)

.venv\Lib\site-packages\langchain_community\tools\sql_database\tool.py
tool -> ListSQLDatabaseTool
class ListSQLDatabaseTool(BaseSQLDatabaseTool, BaseTool):
    """Tool for getting tables names."""
    name: str = "sql_db_list_tables"
    ……

#sql_db_list_tables → sql_db_schema → sql_db_query_checker → sql_db_query
#     列出表         →  获取结构       →  使用 LLM 检查查询    →  执行查询

工具 2：chart_tools

意图识别在 chart_tool.py ，下面是完整流程：

python

infer_display_type(question, columns, rows)
    │
    ├─ 1. extract_axis_from_question() - 解析用户指定的轴配置
    │
    ├─ 2. 根据 mode 判断用哪种识别方式
    │     ├─ 'llm' → infer_display_type_by_llm()
    │     └─ 'keyword' → infer_display_type_by_keyword()
    │
    ├─ 3. 根据 display_type 设置默认轴配置，用户指定了就用用户的
    │     ├─ bar/line → x_axis=文本列, y_axis=数值列
    │     └─ pie → category=文本列, value=数值列
    │
    └─ 返回 {display_type, x_axis, y_axis, category, value}

chart_tool.py 中的详细代码如下（第 497-545 行）：

python

    # 1. 先解析用户指定的轴配置
    user_axis = extract_axis_from_question(question, columns)
    
    # 2. 获取显示类型
    if mode == 'llm':
        display_type = infer_display_type_by_llm(question, columns, rows)
    else:
        display_type = infer_display_type_by_keyword(question, columns, rows)
    
    # 3. 如果用户已指定轴，使用用户的；否则根据显示类型推断默认值
    col_types = [_get_column_type(columns, rows, i) for i in range(len(columns))]
    numeric_cols = [columns[i] for i, t in enumerate(col_types) if t == 'numeric']
    text_cols = [columns[i] for i, t in enumerate(col_types) if t == 'text']
    
    # 默认轴配置
    default_x = text_cols[0] if text_cols else (columns[0] if columns else None)
    default_y = numeric_cols if numeric_cols else []
    default_category = text_cols[0] if text_cols else None
    default_value = numeric_cols[0] if numeric_cols else None
    
    return {
        'display_type': display_type,
        'x_axis_column': user_axis['x_axis_column'] or default_x,
        'y_axis_columns': user_axis['y_axis_columns'] or default_y,
        'category_column': user_axis['category_column'] or default_category,
        'value_column': user_axis['value_column'] or default_value,
    }

最终在 chat_service.py 中的使用（第 255-294 行）：

python

# 调用意图识别
intent_result = infer_display_type(message, pending_sql_columns, pending_sql_rows)
display_type = intent_result['display_type']

if display_type == 'table':
    # 只返回表格数据
    yield {...'chart_type': 'table', 'chart_option': None}
else:
    # 生成图表配置
    chart_config = GenerateEChartsConfigTool.build(
        message, columns, rows,
        chart_type=display_type,
        x_axis_column=intent_result.get('x_axis_column'),
        y_axis_columns=intent_result.get('y_axis_columns'),
        ...
    )
    yield {...'chart_type': chart_config['chart_type'], 'chart_option': chart_config['chart_option']}

python

关键点 ：
- display_type == 'table' → 不生成图表，前端用 <table> 显示
- display_type != 'table' → 调用 GenerateEChartsConfigTool.build() 生成 ECharts 配置，前端用图表渲染

3、skills

python

./skills
     |--- query-writing
               |--- SKILL.md
     |--- schema-exploration
               |--- SKILL.md

4、memory

backend\AGENTS.md
backend\ontology\schema.md
backend\ontology\rules.md

5、context

Step 1：加 checkpointer 单例

新建 / 修改 app/agent/agent.py:

python

from langgraph.checkpoint.memory import InMemorySaver

# 关键:模块级单例,所有 agent 共用
_CHECKPOINTER = InMemorySaver()


def get_checkpointer():
    """给 chat_service 调用的 accessor。"""
    return _CHECKPOINTER

⚠️ 不要在函数内部 InMemorySaver(),那每次都是新对象。

Step 2：`create_deep_agent` 接入

python

from deepagents import create_deep_agent

def create_deep_agent_instance(provider, base_dir, checkpointer=None):
    ...
    agent = create_deep_agent(
        model=...,
        system_prompt=...,
        tools=...,
        skills=...,
        memory=...,
        backend=...,
        checkpointer=checkpointer or _CHECKPOINTER,  # ← 关键
        debug=True,
    )
    return agent

Step 3：`stream_config` 必传 `thread_id`

python

thread_id = f"{user_id}_{conversation_id}"
stream_config = {
    "recursion_limit": 25,
    "configurable": {"thread_id": thread_id},
}

agent.stream(
    {"messages": [{"role": "user", "content": msg}]},
    config=stream_config,
    stream_mode="updates",
)

更详细的说明参见docs\tangdou\4-糖豆（tangdou）对话上下文.md

6、complex_query

复杂查询

python

任意选择报表信息表的3个分组字段，分组查询后生成1个折线图和2个柱状图

6、ontolory_rules

概念规则

python

通过定义本体模型的规则，提高查询的准确性和效率。

Gitee 仓库地址 ​

什么是 Deep Agents？ ​

Deep Agent 工作原理 ​

架构 ​

配置 ​

查询示例 ​

简单查询 ​

带规划功能的复杂查询 ​

推演过程 ​

Deep Agent 关键点 ​

1、create_deep_agent ​

2、tools ​

工具 1：sql_tools ​

工具 2：chart_tools ​

3、skills ​

4、memory ​

5、context ​

Step 1：加 checkpointer 单例 ​

Step 2：create_deep_agent 接入 ​

Step 3：stream_config 必传 thread_id ​

6、complex_query ​

6、ontolory_rules ​

Gitee 仓库地址

什么是 Deep Agents？

Deep Agent 工作原理

架构

配置

查询示例

简单查询

带规划功能的复杂查询

推演过程

Deep Agent 关键点

1、create_deep_agent

2、tools

工具 1：sql_tools

工具 2：chart_tools

3、skills

4、memory

5、context

Step 1：加 checkpointer 单例

Step 2：`create_deep_agent` 接入

Step 3：`stream_config` 必传 `thread_id`

6、complex_query

6、ontolory_rules