博客

项目: deepset-ai/haystack

定位: 开源 AI 编排框架，用于构建生产级 LLM 应用（RAG、Agent、语义搜索、多模态应用）

版本: Haystack 3.0（2024年发布，重大重构版本）

许可证: Apache-2.0

分析时间: 2026-09

1. 核心架构理念：管道（Pipeline）驱动的组件编排

Haystack 的核心思想是将 AI 应用拆解为可组合的组件（Component），通过有向无环图（DAG）管道连接。每个组件有明确的输入/输出接口，管道负责数据流转、调度执行。

1.1 Pipeline 调度引擎

Pipeline 是 Haystack 的心脏，负责根据执行图（graph）调度组件运行。核心运行循环采用优先队列机制：

# haystack/core/pipeline/pipeline.py - Pipeline.run() 核心循环
while True:
    candidate = self._get_next_runnable_component(priority_queue, component_visits)

    # 如果没有可运行的组件，退出循环
    if candidate is None:
        break

    priority, component_name, component = candidate

    # 如果下一个组件被阻塞，检查管道是否可能被卡住
    if priority == ComponentPriority.BLOCKED:
        if self._is_pipeline_possibly_blocked(current_pipeline_outputs=pipeline_outputs):
            self._find_components_blocking_pipeline(
                priority_queue=priority_queue, component_visits=component_visits, inputs=inputs
            )
        # 总是退出循环，因为无法运行下一个组件
        break

    # 如果下一个组件已经调度，等待任务完成以取得进展
    if component_name in scheduled_components:
        async for partial_outputs in self._wait_for_tasks(
            running_tasks, scheduled_components, return_when=asyncio.FIRST_COMPLETED
        ):
            yield partial_outputs
        continue

    # HIGHEST 优先级组件必须单独运行
    if priority == ComponentPriority.HIGHEST:
        async for partial_outputs in self._run_component_in_isolation(...):
            yield partial_outputs
        continue

    # READY 优先级组件可以并发调度
    if priority == ComponentPriority.READY:
        self._schedule_component(...)
        # 尽可能调度更多 READY 任务
        while len(priority_queue) > 0 and not ready_sem.locked():
            peek_priority, peek_name = priority_queue.peek()
            if peek_priority != ComponentPriority.READY:
                break

1.2 组件执行模型

每个组件的运行被封装为独立的执行单元，支持同步和异步两种路径：

（更多…）

证件、账号、联系方式——把这一路学的所有东西，装进一个真正属于你的程序

一、你有多少”重要信息”散落在各处？

先别写代码，盘点一下：

• 身份证号、护照号、驾照号——平时用不到，用到时找不到
• 各种账号：邮箱、视频网站、购物网站、健身房卡号……
• 家人的身份证号（订机票要用）
• 银行卡号、社保号、保单号
• 重要日期：签证过期、证件到期、保修截止

这些信息有个共同点：不用时想不起，要用时找不到。翻聊天记录、翻抽屉、问家人——每次都要折腾半天。

手机备忘录？可以，但查找麻烦、没法分类、明文摆着还不安全。

（更多…）

基于 langgenius/dify 源码分析（154k+ Stars，24.4k Forks）

分析源码：api/core/agent/base_agent_runner.py、api/core/agent/cot_agent_runner.py、api/core/tools/tool_manager.py

1. 项目概览与技术栈

Dify 是一个开源 LLM 应用开发平台，提供 AI 工作流、RAG 管道、Agent 能力、模型管理和可观测性。后端使用 Python（Flask + SQLAlchemy），前端使用 Next.js，支持 Docker Compose 一键部署。

# 项目结构（从仓库目录可见）
# api/          — Python 后端核心（Agent、工具、模型运行时）
# web/          — Next.js 前端
# docker/       — Docker Compose 部署
# dify-agent/   — 独立 Agent 运行时
# dify-agent-runtime/ — Agent 运行时包
# sdks/         — 多语言 SDK
# packages/     — 共享包

Dify 的 Agent 系统基于 graphon 模型运行时抽象层，支持数百种 LLM 提供商，通过统一的 PromptMessage 体系与模型交互。

2. Agent 继承体系——三层架构

Dify 的 Agent 采用清晰的三层继承结构：AppRunner → BaseAgentRunner → CotAgentRunner。

（更多…）

仓库: ComposioHQ/composio (26.2k ⭐)

定位: 为 AI Agent 提供 1000+ 预认证工具集、会话管理、认证、触发器和沙箱执行环境

语言: Python SDK + TypeScript SDK（monorepo）

1. 整体架构：Provider 泛型 + Session 模型

Composio 采用 Provider 泛型模式，核心 Composio 类通过泛型参数 TTool 和 TToolCollection 实现框架无关的工具适配。默认使用 OpenAI 格式，但可以无缝切换到 Anthropic、LangChain 等任何框架。

# sdk.py - 核心 SDK 类定义
class Composio(t.Generic[TTool, TToolCollection], WithLogger):
    """
    Generic parameters:
        TTool: The individual tool type returned by the provider
        TToolCollection: The collection type returned by get_tools
    """

    tools: "Tools[TTool, TToolCollection]"

    @t.overload
    def __init__(
        self: "Composio[OpenAITool, OpenAIToolCollection]",
        provider: None = None,
        **kwargs: te.Unpack[SDKConfig],
    ) -> None:
        """Initialize with default OpenAI provider."""
        ...

    @t.overload
    def __init__(
        self,
        provider: BaseProvider[TTool, TToolCollection],
        **kwargs: te.Unpack[SDKConfig],
    ) -> None:
        """Initialize with an explicit provider. Types are inferred from the provider."""
        ...

设计亮点: 通过 @t.overload 实现两个初始化路径——无 provider 时默认 OpenAI 类型，有 provider 时自动推断泛型参数。这意味着 Composio(provider=AnthropicProvider()) 会自动得到 Composio[ToolParam, list[ToolParam]] 类型。

2. HTTP 客户端：Stainless 生成器 + 运行时环境检测

HTTP 客户端基于 Stainless 自动生成的 API 客户端包装，增加了运行时环境检测和请求拦截能力。

（更多…）

文件存账目、列表管记录、函数做统计——第十二篇价格工具的”理财升级版”

一、月底的灵魂拷问

每个月底总有那么一个瞬间：看看银行卡余额，倒吸一口凉气，然后问出那个经典问题——

“我钱呢？？”

明明没买什么大件，钱就是没了。吃饭、打车、奶茶、网购……每笔都不大，加起来吓人。

（更多…）

项目: letta-ai/letta | Stars: 24.6K | License: Apache 2.0

定位: 有状态 AI Agent 平台——具有高级记忆系统，能够随时间学习和自我改进

前身: MemGPT（UC Berkeley BAIR 实验室研究成果）

当前状态: V1 服务器代码归档于此仓库 archive 分支，新代码已迁移至 letta-ai/letta-code

1. 整体架构哲学：LLM 即操作系统

Letta 的核心理念源自 MemGPT 论文——将 LLM 视为操作系统的内核，Agent 就是运行在这个操作系统上的进程。系统采用分层记忆架构，灵感来自计算机体系结构中的虚拟内存管理：

• 核心记忆（Core Memory） → 寄存器/L1 缓存：始终在上下文窗口内
• 对话记忆（Recall Memory） → RAM：可搜索的历史对话
• 归档记忆（Archival Memory） → 磁盘：长期向量存储
• 文件系统（Filesystem） → 外部存储：Git 版本化的文件

class ContextWindowOverview(BaseModel):
    """Overview of the context window, including the number of messages and tokens."""
    context_window_size_max: int = Field(..., description="The maximum amount of tokens the context window can hold.")
    context_window_size_current: int = Field(..., description="The current number of tokens in the context window.")
    num_messages: int = Field(..., description="The number of messages in the context window.")
    num_archival_memory: int = Field(..., description="The number of messages in the archival memory.")
    num_recall_memory: int = Field(..., description="The number of messages in the recall memory.")
    num_tokens_external_memory_summary: int = Field(..., description="The number of tokens in the external memory summary.")
    num_tokens_system: int = Field(..., description="The number of tokens in the system prompt.")
    num_tokens_core_memory: int = Field(..., description="The number of tokens in the core memory.")
    num_tokens_memory_filesystem: int = Field(0, description="The number of tokens in the memory filesystem section.")
    num_tokens_summary_memory: int = Field(..., description="The number of tokens in the summary memory.")
    num_tokens_functions_definitions: int = Field(..., description="The number of tokens in the functions definitions.")

这个 ContextWindowOverview 精确追踪上下文窗口中每个组成部分的 token 消耗，类似于操作系统的内存使用报告。

2. Agent 类型体系：多形态 Agent 设计

Letta 定义了丰富的 Agent 类型枚举，每种类型对应不同的运行时行为和工具集：

（更多…）

煮面计时5分钟，开会前10分钟提醒——这次用上全系列唯一的新东西：time模块

一、你缺的不是时间管理，是一个帮你盯时间的人

场景一：煮面条，说好煮5分钟。你去刷手机，一抬头——20分钟过去了，面糊了。

场景二：下午3点开会，你想提前10分钟准备。结果一忙起来抬头看表——3点05了。

场景三：敷面膜15分钟、煮鸡蛋8分钟、烤箱25分钟、午睡20分钟……

（更多…）

项目: agno-agi/agno

定位: Build, run, and manage agent platforms

Stars: 42,131 | Forks: 5,903 | License: Apache-2.0

语言: Python | 核心框架: Pydantic

一、项目概览与分层架构

Agno 不仅仅是一个 Agent 框架，而是一个完整的 Agent 平台运行时（AgentOS）。它由三层构成：

1. SDK 层 (libs/agno/agno/) — Agent 构建的核心 API
2. 运行时层 — 50+ REST/SSE/WebSocket 端点，提供生产级服务
3. 管理层 — Web UI 控制面，JWT RBAC，多租户隔离

源码目录结构清晰地反映了这种分层：

libs/agno/agno/
├── agent/          # Agent 核心（拆分为 _init, _run, _tools, _session 等子模块）
├── models/         # 模型抽象层（100+ 提供商适配）
├── tools/          # 工具系统（Toolkit + Function + @tool 装饰器）
├── memory/         # 记忆管理（MemoryManager + 优化策略）
├── session/        # 会话持久化
├── run/            # 运行时事件系统
├── knowledge/      # 知识库协议
├── db/             # 数据库抽象
└── ...

Agent 核心类采用 模块拆分 模式——agent.py 只是入口，实际逻辑分散在 _init.py、_run.py、_tools.py、_session.py 等子模块中，通过 __init__.py 的 re-export 组合在一起：

（更多…）

做饭要换算克和毫升，出差要换算时区——让字典和函数帮你记住所有公式

一、你一定经历过这种时刻

做饭看菜谱：”面粉200克，牛奶150毫升”。你家只有量杯，怎么量？

出差订机票：航班显示”当地时间14:30起飞”，北京时间是几点？

健身看教程：”做3组，每组15磅的哑铃”。磅是多少斤？

（更多…）

项目: browser-use/browser-use

定位: AI 浏览器自动化 Agent 框架，MIT 许可，86k+ Stars

核心理念: 让 AI 像人类一样操控浏览器——理解页面结构、规划操作序列、执行交互动作

架构总览

Browser-Use 采用三层分离架构：Agent 层（LLM 决策）→ Tools/Controller 层（动作注册与执行）→ BrowserSession 层（CDP 浏览器控制）。事件总线（EventBus）贯穿全栈，所有浏览器操作以事件驱动方式执行。以下是 10 个核心维度的深度分析。

1. Agent 核心循环（Step-Loop）

Agent 的执行是一个经典的感知-思考-行动循环。每个 step 由三个阶段组成：准备上下文、获取 LLM 输出、执行动作。

# agent/service.py - Agent.step()
@observe(name='agent.step', ignore_output=True, ignore_input=True)
@time_execution_async('--step')
async def step(self, step_info: AgentStepInfo | None = None) -> None:
    """Execute one step of the task"""
    self.step_start_time = time.time()
    browser_state_summary = None
    try:
        # Phase 0: CAPTCHA 等待
        if self.browser_session:
            captcha_wait = await self.browser_session.wait_if_captcha_solving()
            if captcha_wait and captcha_wait.waited:
                self.step_start_time = time.time()

        # Phase 1: 准备上下文（浏览器状态 + 消息构建）
        browser_state_summary = await self._prepare_context(step_info)
        self.state.last_model_output = None
        self.state.last_result = None

        # Phase 2: LLM 推理 + 动作执行
        await self._get_next_action(browser_state_summary)
        await self._execute_actions()

        # Phase 3: 后处理（下载追踪、循环检测、日志）
        await self._post_process()

    except Exception as e:
        await self._handle_step_error(e)
    finally:
        await self._finalize(browser_state_summary)

关键设计：每个 step 的超时由 step_timeout=180 控制，LLM 调用有独立的 llm_timeout（根据模型自动配置，如 Gemini 75s、Claude 90s）。CAPTCHA 解算时间不计入 step 耗时。

（更多…）