工具与集成:让 agent 长出手脚

30 秒导读: 大模型自己什么也做不了——它只会输出文字,顶多输出一段"我想调用 send_email(to=...)"的 JSON。本章讲 Julep 怎么把这段"我想调用"精确、可靠地变成一次真实的调用:发出真实的 HTTP 请求、真的去搜 Wikipedia、真的往数据库里写一行。核心就一件事——模型的意图,如何落到真实的工具执行上。

本章是 Julep 全景 里"手脚"那一支。它假设你已经知道核心概念(agent / task / tool)和任务执行引擎(Temporal 工作流、状态机)。检索类工具(文档搜索、RAG)的存储细节不在这里,留给记忆与混合检索。

---

1. 这是什么(零基础也能懂)

先建立一个直觉

大模型是个只会说话的大脑:你给它一段文字,它回你一段文字。它不能真的发邮件、不能真的查天气、不能真的往你的数据库写数据。

要让它"长出手脚",业界的做法叫 tool calling(工具调用):

1. 你在请求里告诉模型"你有这些工具可用"(每个工具带名字、描述、参数 schema)。
2. 模型不直接执行,而是回一段结构化的 JSON:"我想调用 get_weather,参数 {city: '北京'}"。
3. 你的系统负责真的去执行这次调用,把结果再喂回给模型。

第 3 步——"真的去执行"——就是本章的全部内容。这一步看着简单,其实是整个平台工程含量最高的一支:模型说的是一句话,真实世界的目标却五花八门(第三方 API、内部数据库、HTTP 端点……),要把这句话精确地送到对的地方,还要处理鉴权、参数合并、错误重试。

Julep 的四种工具类型

Julep 把一切"手脚"归成四类,判定依据是 Tool 对象上哪个字段非空(见 §3.1):

工具类型	落到哪里	典型例子
function	交还给客户端执行(Julep 只负责暂停等结果)	你自己的业务函数、需要人参与的动作
integration	走 integrations-service 微服务,由内置 provider 执行	Wikipedia 搜索、Brave 搜索、发邮件、浏览器自动化
api_call	Julep 直接发一个 HTTP 请求	调用任意第三方 REST API
system	调用 Julep 自己的内部能力	创建 session、搜文档、触发另一个 task

一句话记住四类: function 是"你来做",integration 是"我用现成 provider 做",api_call 是"我直接发 HTTP",system 是"我调我自己"。

用起来什么样

一个 task 里的工具定义(YAML)大概长这样——这就是一个 integration 工具:

# 示意:一个 integration 类型的工具定义
tools:
  - name: wikipedia_search        # 模型看到的工具名
    type: integration
    integration:
      provider: wikipedia         # 用哪个内置 provider
      method: search              # provider 的哪个方法
# 之后在工作流里:
- tool: wikipedia_search
  arguments:
    query: $ _.topic             # $ 开头是表达式,见第 03 章

模型(或工作流)说"调 wikipedia_search,query 是某某",Julep 就真的去 integrations-service 搜一次 Wikipedia,把 documents 返回。本章讲的就是这中间发生了什么。

---

2. 顶层全景(它大概怎么转)

一次工具调用的生命线

无论工具从哪来,Julep 都把它拍平成同一个中间产物——一个带 type 字段的"工具调用 dict",然后按 type 分发到四条落地路径。这是全章的主干:

                         ┌─────────────────────────────────┐
   模型/工作流产生意图    │  统一中间产物:tool_call dict     │
                         │  { type: "integration",          │
   ①prompt_step:         │    integration:{name,arguments}, │
      模型吐 tool_calls   │    id: "call_xxx" }              │
   ②tool_call_step:      └───────────────┬─────────────────┘
      工作流显式构造                      │  按 type 分发
                                          ▼
        ┌──────────────┬─────────────────┬──────────────────┐
        │              │                 │                  │
        ▼              ▼                 ▼                  ▼
   type=function   type=integration  type=api_call     type=system
        │              │                 │                  │
   交还客户端      execute_integration execute_api_call  execute_system
   (暂停工作流,   ─┐               │                  │
    等 client 回)  │ HTTP           │ httpx            │ 内部 handler
        │          ▼               ▼                  ▼
        │    integrations-service  第三方 REST API   Julep 自己的
        │    (providers.py +                          queries/routers
        │     各 provider 实现)
        ▼
   raise_complete_async

怎么读这张图: 左边两个"入口"制造出中间的 tool_call dict;中间按 type 一分四;右边四列是四条真实落地路径。看懂这张图,本章就懂了七成。

部件一句话职责

部件	干什么	在哪个文件
Tool.type 属性	看哪个字段非空,判定四类之一	autogen/openapi_model.py:95 type_property
construct_tool_call	把工具+参数拍成统一的 tool_call dict	activities/task_steps/tool_call_step.py
prompt_step	让模型产出 tool_calls(入口一)	activities/task_steps/prompt_step.py:31
_handle_ToolCallStep	工作流里按 type 分发(主分发点)	workflows/task_execution/__init__.py:572
run_tool_call	无 Temporal 上下文的分发点(chat/prompt 用)	common/utils/tool_runner.py:93
execute_integration	落地路径:调 integrations-service	activities/execute_integration.py:31
execute_api_call	落地路径:httpx 发 HTTP	activities/execute_api_call.py:27
execute_system	落地路径:调内部 handler	activities/execute_system.py:29
integrations-service	独立微服务,装着所有 provider	src/integrations-service/

两个"入口"、两套"分发"

这里有一个容易绕晕的点,先讲清楚:tool_call dict 有两个产生入口,落地时也有两套分发实现,但四条最终路径是同一批 execute_* 函数。

• 入口①(模型驱动): prompt_step 把工具告诉模型,模型自己决定要不要调、调哪个,吐出 tool_calls。

• 入口②(工作流显式): task 里写死一个 tool_call 步骤,由工作流用 construct_tool_call 直接构造调用(模型不参与)。

• 分发 A(Temporal 工作流): _handle_ToolCallStep 用 workflow.execute_activity(...) 把落地包成 Temporal 活动(有重试、有 heartbeat)。

• 分发 B(轻量上下文): run_tool_call 直接 await execute_*(...),给 chat 端点和"自动跑工具"的 prompt 用。

后面 §3 讲入口和统一产物,§4 讲四条落地路径,§5 讲两套分发的取舍和 feature flag 切换。

---

3. 核心原理之一:意图怎么变成统一的 tool_call

3.1 四种类型是怎么判定的

Julep 的 Tool 模型不是用一个 type 枚举硬填,而是四个可选字段谁非空就是谁。判定逻辑是一段被"打补丁"挂到模型上的计算属性:

# 真实源码 autogen/openapi_model.py:95  type_property
def type_property(self: BaseModel) -> str:
    return (
        "function"    if self.function
        else "integration" if self.integration
        else "system"      if self.system
        else "api_call"    if self.api_call
        else None
    )

Tool.type = computed_field(property(type_property))   # 打补丁挂上去

这段把 function / integration / system / api_call 四个字段按优先级翻译成一个字符串 type。同样的补丁也打在 TaskTool、UpdateToolRequest、PatchToolRequest 上(openapi_model.py:110-119)。

Tool 模型本身在 autogen/Tools.py:2967(class Tool),四个落地字段并列摆着:function、integration(一个巨大的 provider Union)、api_call、system。这些是 TypeSpec 生成的——真源在 src/typespec/tools/,不要手改 autogen/。

注意:Tool.type 的 Literal 里其实还有 computer_20241022 等 Anthropic computer-use 类型,但落地分发只认前四类,computer-use 走 prompt_step 里的特判(prompt_step.py:151),本章不展开。

3.2 统一产物:construct_tool_call

不管从哪个入口来,最终都要变成同一个形状的 dict。这活由 construct_tool_call 干:

# 真实源码 activities/task_steps/tool_call_step.py  construct_tool_call
def construct_tool_call(tool, arguments, call_id) -> dict:
    return {
        tool.type: {                    # 键就是 "integration"/"api_call"/"function"
            "arguments": arguments,
            "name": tool.name,
        }
        if tool.type != "system"        # system 特殊:展开 resource/operation/...
        else {
            "resource": tool.system and tool.system.resource,
            "operation": tool.system and tool.system.operation,
            "resource_id": tool.system and tool.system.resource_id,
            "subresource": tool.system and tool.system.subresource,
            "arguments": arguments,
        },
        "id": call_id,                  # 形如 call_xxxx,见下
        "type": tool.type,              # 分发就靠它
    }

关键设计:用 tool.type 同时当"外层 key"和"分发字段"。于是产出的 dict 形如 {"integration": {...}, "id": ..., "type": "integration"}——下游只要看 type,再去取同名 key 里的负载即可。system 是唯一例外:它不带 name,而是展开成 resource/operation/subresource(呼应 §4.4 的内部调用寻址)。

那个 call_id 由 generate_call_id 生成——18 随机字节做 base64、去掉 padding、加 call_ 前缀(tool_call_step.py generate_call_id),这是对齐 OpenAI 工具调用 id 的格式。

源码里 construct_tool_call/generate_call_id 两个函数头上都挂着 FIXME: This shouldn't be here——作者自己也觉得该搬地方。属于"能用但欠打磨"的一块。

3.3 入口②:工作流显式构造(tool_call 步骤)

task 里写一个 tool_call 步骤,模型完全不参与——工作流按名字找到工具、求值参数、直接构造:

# 真实源码 workflows/task_execution/__init__.py:171  case ToolCallStep
case ToolCallStep(arguments=arguments):
    tools = await self.context.tools()             # 取到本步可见的工具集
    tool_name = self.context.current_step.tool
    tool = next((t for t in tools if t.name == tool_name), None)
    if tool is None:
        raise ApplicationError(f"Tool {tool_name} not found in the toolset")
    arguments = await base_evaluate_activity(arguments, self.context)  # 求值 $ 表达式
    call_id = generate_call_id()
    output = construct_tool_call(tool, arguments, call_id)   # → 统一产物

注意 arguments 先经过 base_evaluate_activity——task 里的 $ _.topic 这类 $ 表达式 在这里被求成真实值,再塞进 tool_call。产出的 output 就是那个统一 dict,交给下一节的分发。

3.4 入口①:模型驱动(prompt_step)

另一个入口是让模型自己决定。prompt_step 把工具格式化成 LLM 能懂的样子、调模型,拿回 tool_calls:

# 真实源码(节选) activities/task_steps/prompt_step.py:229  legacy 路径
for choice in response_as_dict["choices"]:
    if choice["finish_reason"] == "tool_calls":
        for call in choice["message"]["tool_calls"]:
            original_tool = tools_mapping.get(call["function"]["name"])
            if original_tool.type == "function":
                continue                       # function 保持原样交回客户端
            call.pop("function")               # 其余:把 openai 的 function 壳
            call["type"] = original_tool.type  # 换回 Julep 的内部 type
            call[original_tool.type] = {
                "name": call["function"]["name"],
                "arguments": call["function"]["arguments"],
            }

这里有个巧妙的"来回翻译":模型只认 OpenAI 的 {"type":"function", "function":{...}} 格式,所以送进去时所有工具(哪怕是 integration)都被 format_tool 伪装成 function(见 §5.1);拿回来时,再按工具原本的 type 翻译回内部形状,好让后续分发认得出它其实是个 integration。这一步靠 tools_mapping(工具名 → 原始 Tool)完成。

prompt_step 其实有新旧两条实现,由 feature flag 切换,细节见 §5.4。这里展示的是 legacy 路径的"翻译回内部格式"逻辑。

---

4. 核心原理之二:四条落地路径

统一产物有了,现在按 type 分发。工作流里的主分发点是 _handle_ToolCallStep(workflows/task_execution/__init__.py:572),一串 if tool_call["type"] == ...。逐条看。

4.1 function —— 交还给客户端

function 类型 Julep 自己不执行。它把工作流挂起,等外部客户端算完再回传结果:

# 真实源码(节选) workflows/task_execution/__init__.py:582
if tool_call["type"] == "function":
    tool_call_response = await workflow.execute_activity(
        task_steps.raise_complete_async,
        args=[self.context, tool_call],
        schedule_to_close_timeout=timedelta(days=31),   # 最长可等 31 天
        ...
    )
    return WorkflowResult(
        state=PartialTransition(output=tool_call_response, type="resume"),
    )

秘密在 raise_complete_async:它把当前 Temporal 活动的 task_token 存进一次 wait 转移,然后调用 activity.raise_complete_async() 让活动"异步完成":

# 真实源码(节选) activities/task_steps/raise_complete_async.py:13
captured_token = base64.b64encode(activity_info.task_token).decode("ascii")
transition_info = CreateTransitionRequest(
    current=context.cursor, type="wait", next=None, output=output,
    task_token=captured_token, ...)
await original_transition_step(context, transition_info)
activity.raise_complete_async()          # 活动挂起,不占 worker

直觉: 工作流走到这一步会冻结,状态机进入 wait(见执行引擎的状态机)。客户端在自己那边执行函数、拿到结果,再拿着这个 token 回调 Julep,工作流从 wait 恢复(type="resume")。这就是 Julep 支持"客户端侧工具 / 人在环"的机制——最长能等 31 天。

4.2 integration —— 走 integrations-service

integration 是最常用的一类。工作流侧先把 provider/method/setup 打包成 BaseIntegrationDef,连同一堆 id 交给 execute_integration 活动:

# 真实源码(节选) workflows/task_execution/__init__.py:595
if tool_call["type"] == "integration":
    integration_tool = next((t for t in tools if t.name == tool_name), None)
    integration = BaseIntegrationDef(
        provider=integration_tool.integration.provider,
        setup=..., method=integration_tool.integration.method,
        arguments=arguments)
    tool_call_response = await workflow.execute_activity(
        execute_integration,
        args=[developer_id, agent_id, task_id, session_id,
              tool_name, integration, arguments], ...)

execute_integration 活动(activities/execute_integration.py:31)做三件事:

1. 合并参数与鉴权:从数据库按元数据取出 agent/task/session 级别预存的 args 和 setup,和调用传入的合并(get_tool_args_from_metadata,execute_integration.py:50-79)。这让"API key 存一次、到处复用"成为可能。
2. 序列化:把所有 Pydantic 对象递归转成 JSON 可传的形状(serialize_pydantic_objects)。
3. 发出去:调 integrations.run_integration_service(...),除非 provider 是 dummy(直接回显参数,测试用,execute_integration.py:84)。

真正的 HTTP 在客户端里:

# 真实源码(节选) clients/integrations.py:11  run_integration_service
slug = f"{provider}/{method}" if method else provider
url = f"{integration_service_url}/execute/{slug}"
response = await client.post(url, json={"arguments": arguments, "setup": setup})

到这一步,请求离开了 agents-api,进了独立的 integrations-service 微服务。下一节专门讲它。

4.3 integrations-service 内部:provider 注册表 + 动态派发

integrations-service 是一个独立的 FastAPI 服务(src/integrations-service/),它的核心是一张 provider 注册表和一段动态派发。

注册表长这样——每个 provider 声明自己有哪些方法、每个方法吃什么参数、吐什么输出:

# 真实源码(节选) integrations-service/integrations/providers.py:77
wikipedia = BaseProvider(
    provider="wikipedia",
    setup=None,                         # 无需鉴权
    methods=[
        BaseProviderMethod(
            method="search",
            arguments=WikipediaSearchArguments,   # 入参 schema
            output=WikipediaSearchOutput,         # 出参 schema
        ),
    ],
    info=ProviderInfo(url=..., friendly_name="Wikipedia"),
)

BaseProvider / BaseProviderMethod 的结构在 models/base_models.py:19-27。所有 provider 汇进一个字典 available_providers(providers.py:469),截至本 commit 有 16 个:wikipedia、weather、spider、brave、email、browserbase、remote_browser、llama_parse、ffmpeg、cloudinary、arxiv、unstructured、algolia、mailgun、mcp、google_sheets。

派发发生在 HTTP 端点 POST /execute/{provider}/{method}(routers/execution/execute.py:24),它转调统一执行器:

# 真实源码(节选) integrations-service/integrations/utils/execute_integration.py:17
provider_obj = getattr(available_providers, provider, None)   # 按名取 provider
method = method or provider_obj.methods[0].method
method_config = next((m for m in provider_obj.methods if m.method == method), None)

provider_module = importlib.import_module(          # 动态导入 provider 实现模块
    f"integrations.utils.integrations.{provider_obj.provider}")

arguments = method_config.arguments(**arguments.model_dump())   # 用声明的 schema 校验入参
return await getattr(provider_module, method)(     # 按方法名取函数并调用
    **({"setup": setup} if setup else {}), arguments=arguments)

核心技巧: importlib.import_module + getattr(module, method)——用字符串动态找到 provider 模块和方法函数。加一个新集成,只要写一个 utils/integrations/<name>.py 模块、在 providers.py 注册,不用改任何分发代码。

每个 provider 实现就是一个带 schema 的 async 函数,统一配了重试:

# 真实源码(节选) integrations-service/integrations/utils/integrations/wikipedia.py
@beartype
@retry(wait=wait_exponential(multiplier=1, min=4, max=10),
       reraise=True, stop=stop_after_attempt(4))    # 指数退避,最多 4 次
async def search(arguments: WikipediaSearchArguments) -> WikipediaSearchOutput:
    loader = WikipediaLoader(query=arguments.query,
                             load_max_docs=arguments.load_max_docs)
    return WikipediaSearchOutput(documents=loader.load())

为什么单拆一个微服务? 集成往往带一堆重依赖(playwright、langchain、ffmpeg……),把它们隔离在独立服务里,agents-api 保持轻;而且集成可以独立伸缩、独立部署。

4.4 api_call —— 直接发 HTTP

api_call 不经过 integrations-service,agents-api 自己用 httpx 发请求。工作流侧把 ApiCallDef(method/url/headers/follow_redirects)和参数交给活动:

# 真实源码(节选) workflows/task_execution/__init__.py:653
if tool_call["type"] == "api_call":
    api_call = ApiCallDef(method=..., url=..., headers=..., follow_redirects=...)
    if "json_" in arguments:                 # json_ 是别名,回填成 json
        arguments["json"] = arguments["json_"]; del arguments["json_"]
    tool_call_response = await workflow.execute_activity(
        execute_api_call, args=[api_call, arguments], ...)

execute_api_call(activities/execute_api_call.py:27)就是一个规规矩矩的 httpx 客户端:

• 允许 arguments 覆盖 url / method / headers / follow_redirects(定义里是默认值,调用时可改)。
• headers 做合并:(arg_headers or {}) | (api_call.headers or {})。
• raise_for_status(),出错时把响应体前 500 字塞进日志。
• 响应打包成 {status_code, headers, json};需要时把原始 content 用 base64 塞进 content 字段(execute_api_call.py:83)。

直觉: 这是"万能逃生舱"——任何没有专用 provider 的第三方 REST API,都能用 api_call 直接打。

4.5 system —— 调 Julep 自己

system 类型让工具回头调用 Julep 平台自己的能力:创建 session、搜文档、列 agent、甚至触发另一个 task 执行。工作流侧:

# 真实源码(节选) workflows/task_execution/__init__.py:698
if tool_call["type"] == "system":
    system_call = SystemDef(**tool_call.get("system"))
    developer_id = self.context.execution_input.developer_id
    tool_call_response = await workflow.execute_activity(
        execute_system, args=[developer_id, system_call], ...)

SystemDef(autogen/Tools.py:2864)不是"名字+参数",而是三元寻址:resource(agent/user/task/session/doc/...)× operation(create/update/search/chat/list/...)× 可选 subresource(doc/execution/...)。这解释了 §3.2 为什么 system 要特殊展开。

execute_system(activities/execute_system.py:29)拿这个三元组去查一张巨大的 match 表,把它映射到真实的 query 函数或路由函数:

# 真实源码(节选) common/utils/evaluator.py:730  get_handler
match (system.resource, system.subresource, system.operation):
    case ("agent", "doc", "search"):  return search_agent_docs
    case ("agent", None,  "list"):    return list_agents_query
    case ("session", None, "chat"):   return chat
    case ("task", "execution", "create"): return create_task_execution
    ...

拿到 handler 后,execute_system 还要为不同操作做参数改形(比如 doc 操作要补 owner_type/owner_id;chat 操作要构造 ChatInput 和 BackgroundTasks 再手动跑,execute_system.py:97-115),再 await handler(**arguments)。

直觉: system 工具让 agent 能操作 Julep 的资源本身——于是可以搭出"一个 agent 创建另一个 agent""在 task 里触发子 task"这类元能力。注意 handler 直接引到了 routers/queries(源码里挂着 FIXME: Do not use routes directly),是一处已知的耦合。

---

5. 核心原理之三:两套分发实现与 feature flag

前面 §4 讲的是 Temporal 工作流里的分发(_handle_ToolCallStep,每条落地都包成 execute_activity,带重试和 heartbeat)。但 Julep 还有第二套更轻量的分发,专给 sessions 的聊天和"自动跑工具"用。两套并存,靠 feature flag 切换——这一节讲清楚。

5.1 轻量分发:run_tool_call

common/utils/tool_runner.py 里的 run_tool_call 是一个不依赖 Temporal 上下文的分发点,直接 await 落地函数:

# 真实源码(节选) common/utils/tool_runner.py:93  run_tool_call
if tool.type == "integration" and tool.integration:
    output = await execute_integration(developer_id=..., tool_name=tool.name,
                                       integration=tool.integration, arguments=arguments, ...)
if tool.type == "system" and tool.system:
    output = await execute_system(developer_id=..., system=system_def, ...)
if tool.type == "api_call" and tool.api_call:
    output = await execute_api_call(tool.api_call, arguments)
# 注意:这里没有 function 分支 —— 无上下文场景不支持 function 工具

它复用的是同一批 execute_integration / execute_system / execute_api_call 函数(§4),只是不套 Temporal 活动。run_context_tool(tool_runner.py:145)是它的一层薄封装:从 StepContext 里掏出 developer/agent/task/session id 再转调。

配套还有两个格式转换器:

• format_tool(tool_runner.py:32):把内部 Tool 统一伪装成 OpenAI 的 function 形状喂给模型(integration 走 integrations-service 的 /tool 端点转,system 用 inspect.signature 反射出参数 schema)。
• convert_litellm_to_chosen_tool_call(tool_runner.py:171):模型回来后,把 litellm 的调用按工具原 type 翻译回内部 BaseChosenToolCall——和 §3.4 legacy prompt_step 里的翻译是一个意思,只是这里产出的是强类型对象而非裸 dict。

5.2 工具循环:run_llm_with_tools

轻量分发的驱动器是 run_llm_with_tools(tool_runner.py:200)——一个"调模型 → 跑工具 → 把结果喂回 → 再调模型"的循环:

# 真实源码(节选) common/utils/tool_runner.py:200  run_llm_with_tools
formatted_tools = [await format_tool(t) for t in tools]
while True:
    response = await litellm.acompletion(tools=formatted_tools, messages=messages_copy, ...)
    choice = response.choices[0]
    messages_copy.append(choice.message.model_dump())
    if choice.finish_reason != "tool_calls" or not choice.message.tool_calls:
        return messages_copy                     # 模型不再要工具了,结束
    for litellm_call in choice.message.tool_calls:
        tool = tool_map.get(litellm_call.function.name)
        internal_call = convert_litellm_to_chosen_tool_call(litellm_call, tool)
        result = await run_tool_call(tool, internal_call)     # ← 落地
        messages_copy.append(format_tool_results_for_llm(result))  # 结果喂回

直觉: 这就是标准的 agent tool loop。它在一次调用内部把工具跑完、结果回喂、直到模型给出最终文字答复——不像工作流那样每步一个 Temporal 活动。代价是没有 Temporal 的持久化/重试保护(它是普通 async 循环)。

5.3 sessions 的两套实现,靠 feature flag 切

sessions 的 /chat 端点是"两套并存"最明显的地方。chat.py 是个纯路由器,按 flag 选实现:

# 真实源码(节选) routers/sessions/chat.py:67
if get_feature_flag_value("auto_run_tools_chat", developer_id=str(developer.id)):
    from .auto_tools.chat import chat as chat_auto_tools     # 新:自动跑工具
    return await chat_auto_tools(...)
from .legacy.chat import chat as chat_legacy                 # 旧:只把 tool_calls 还给客户端
return await chat_legacy(...)

两者的行为差别:

	legacy(legacy/chat.py)	auto_tools(auto_tools/chat.py)
拿到模型的 tool_calls 后	原样返回给客户端,由客户端自己执行	内部用 run_llm_with_tools 自动跑完再返回最终答复
依赖的循环	无(单次 completion)	run_llm_with_tools(§5.2)
触发条件	flag 关	flag auto_run_tools_chat 开

auto_tools 版的关键就一行——把 chat 上下文的 run_tool_call 包成 partial 交给循环:

# 真实源码(节选) routers/sessions/auto_tools/chat.py:177
all_messages = await run_llm_with_tools(
    messages=messages, tools=tools_to_use,
    settings=completion_data, run_tool_call=run_tool_partial)   # 见 :112
final_response = all_messages[-1]

还有一个运行时开关 auto_run_tools(区别于 feature flag):它是请求/步骤上的布尔字段(ChatInput.auto_run_tools,autogen/Chat.py:600;PromptStep/ToolCallStep 也各有,autogen/Tasks.py:937,985)。当它为 False 时,代码干脆不把工具传给模型(tools_to_use = tools if chat_input.auto_run_tools else [],auto_tools/chat.py:81)——这样模型根本不会发起工具调用,简化了流程。

5.4 prompt_step 也是同一个套路

回到入口①。prompt_step 内部同样"新旧两条路",由另一个 flag auto_tool_calls_prompt_step 决定:

# 真实源码(节选) activities/task_steps/prompt_step.py:86
if get_feature_flag_value("auto_tool_calls_prompt_step", developer_id=...):
    tools_to_use = all_tools if context.current_step.auto_run_tools else []
    responses = await run_llm_with_tools(          # 新路径:自动跑工具循环
        messages=prompt, tools=tools_to_use, settings=completion_data,
        run_tool_call=functools.partial(run_context_tool, context))
    return StepOutcome(output=responses, next=None)
# ...否则走 legacy:直接 litellm.acompletion,再把 tool_calls 翻译回内部格式(§3.4)

归纳整个切换体系: Julep 正处在"从旧的手工分发迁到统一 run_llm_with_tools"的过程中。旧路径把工具调用当成一个需要下游/客户端再处理的产物;新路径在原地把工具跑完。哪条生效,由 per-developer 的 feature flag 控制——这是一种灰度迁移的姿势。

---

6. 巧妙之处(可借鉴的技术)

• 用"哪个字段非空"代替 type 枚举,再用计算属性反推 type(openapi_model.py:95)。好处:每类工具的负载天然只能填自己那一个字段,类型即数据,不会出现 "type 说是 integration 但填了 api_call" 的矛盾态。

• type 既当分发键、又当负载 key(construct_tool_call)。产出的 dict {"integration": {...}, "type": "integration"} 让下游一个字段既知道走哪条路、又知道去哪取参数,分发代码极简。

• integrations-service 的动态派发:importlib + getattr(utils/execute_integration.py:36-51)。加集成零改分发代码,只写模块 + 注册——插件式扩展的典范。

• function 工具 = 把 Temporal 活动"异步完成"(raise_complete_async)。借 task_token 把工作流冻在 wait,最长等 31 天,天然支持"客户端侧工具 / 人在环",且挂起期间不占 worker。

• setup/args 的多级合并(execute_integration.py:50-79):调用传入 < 工具定义 < 数据库预存元数据,层层 | 合并。让"API key 存一次、跨调用复用"成立,密钥不必写进每次调用。

• feature flag 做灰度迁移(auto_run_tools_chat / auto_tool_calls_prompt_step):新旧两套工具执行实现并存,按 developer 切,老用户不受影响。

---

7. 边界与局限(诚实)

• 两套实现并存 = 复杂度税。 工作流分发(_handle_ToolCallStep)和轻量分发(run_tool_call)是两处独立代码,四类工具在两边各写一遍;prompt_step / chat 又各有新旧两条路。读代码时极易看错自己在哪条路上。

• 轻量路径不支持 function 工具。 run_tool_call(tool_runner.py:93)没有 function 分支——"交还客户端"这套只在 Temporal 工作流里成立;chat/自动跑工具场景遇到 function 工具会落到末尾返回空 output(tool_runner.py:141)。

• 作者自认欠打磨的地方不少。 construct_tool_call/generate_call_id 挂着 FIXME: This shouldn't be here;execute_system 的 handler 直接引 routers(FIXME: Do not use routes directly);prompt_step 里为 Claude/Groq 的 formatted_tools 有一串 hack 和 AIDEV-TODO(prompt_step.py:146-187)。

• api_call 直连、无沙箱。 execute_api_call 用 httpx 直接打任意 URL(execute_api_call.py:33),url 可被 arguments 覆盖——SSRF 之类的边界由上层鉴权/配置负责,这一层不设防。

• integrations-service 是单点重依赖。 所有 integration 都得过这个服务;它挂了,integration 类工具全废(api_call/system 不受影响,因为它们不经过它)。

---

8. 横向对比

同 shelf 的其它 agent 框架在"工具落地"上取舍不同(逐条见总库 doc 的"工具执行"原理):

• Julep 的特色是"四类 + 微服务 + Temporal 活动":integration 抽成独立可伸缩服务、function 借工作流挂起支持人在环、每次落地都是有重试的 Temporal 活动。代价是架构重、两套实现并存。
• 轻量框架(如把工具当普通 Python 函数注册、进程内直接调)启动快、心智负担小,但没有持久化重试、没有跨调用的密钥合并、也难做"等 31 天"的人在环。

想深入 Julep 其它环节:工具产出的 tool_call 如何在状态机里流转见任务执行引擎;$ 参数怎么求值见工作流步骤与表达式;检索类工具的存储见记忆与混合检索;整体服务拓扑见平台架构。

---

9. 代码地图(导航索引)

主题	文件路径	符号名
四类工具的 type 判定	src/agents-api/agents_api/autogen/openapi_model.py:95	type_property
Tool 模型(四个落地字段)	src/agents-api/agents_api/autogen/Tools.py:2967	class Tool
SystemDef(resource×operation×subresource)	src/agents-api/agents_api/autogen/Tools.py:2864	class SystemDef
统一产物构造	src/agents-api/agents_api/activities/task_steps/tool_call_step.py	construct_tool_call / generate_call_id
入口②:工作流显式构造	src/agents-api/agents_api/workflows/task_execution/__init__.py:171	case ToolCallStep
主分发点(按 type 分四路)	src/agents-api/agents_api/workflows/task_execution/__init__.py:572	_handle_ToolCallStep
入口①:模型驱动 + 翻译回内部格式	src/agents-api/agents_api/activities/task_steps/prompt_step.py:31	prompt_step
function:挂起等客户端	src/agents-api/agents_api/activities/task_steps/raise_complete_async.py:13	raise_complete_async
integration 落地(合并/序列化/发出)	src/agents-api/agents_api/activities/execute_integration.py:31	execute_integration
调 integrations-service 的 HTTP 客户端	src/agents-api/agents_api/clients/integrations.py:11	run_integration_service
api_call 落地(httpx)	src/agents-api/agents_api/activities/execute_api_call.py:27	execute_api_call
system 落地	src/agents-api/agents_api/activities/execute_system.py:29	execute_system
system 的 handler 映射表	src/agents-api/agents_api/common/utils/evaluator.py:687	get_handler
轻量分发点(chat/prompt 用)	src/agents-api/agents_api/common/utils/tool_runner.py:93	run_tool_call
agent 工具循环	src/agents-api/agents_api/common/utils/tool_runner.py:200	run_llm_with_tools
工具伪装成 OpenAI function	src/agents-api/agents_api/common/utils/tool_runner.py:32	format_tool
chat 端点按 flag 选实现	src/agents-api/agents_api/routers/sessions/chat.py:67	chat
chat 自动跑工具实现	src/agents-api/agents_api/routers/sessions/auto_tools/chat.py:177	chat (auto_tools)
integrations-service:provider 注册表	src/integrations-service/integrations/providers.py:469	available_providers
integrations-service:动态派发	src/integrations-service/integrations/utils/execute_integration.py:17	execute_integration
integrations-service:执行端点	src/integrations-service/integrations/routers/execution/execute.py:24	execute
provider 结构定义	src/integrations-service/integrations/models/base_models.py:19	BaseProviderMethod / BaseProvider
provider 实现样例	src/integrations-service/integrations/utils/integrations/wikipedia.py	search