arbis-llmwrap 0.3.6

Decorator to wrap LLM calls for production use with flexible prompt binding.

llmwrap

Usage guide for llmwrap with examples mirroring all scenarios covered in tests/testing_different_interfaces.py.

This document intentionally covers API usage only. Internal algorithm details are not disclosed.

Table of Contents

• Install
• Public APIs
• Required Config Fields
• Function Wrapper Cases (wrap_llm_call)
  • F1. OpenAI SDK direct call shape preserved
  • F2. OpenRouter via OpenAI SDK object flow
  • F3. OpenAI Responses API automatic shape preservation
  • F4. Multipart content preserved
  • F5. Tool calls passthrough then final wrapped answer
  • F6. Wrapped tools making internal wrapped calls
  • F7. Top agent with two wrapped tools and sub-agents
  • F8. Top-level real tools + nested wrapped llm tools
  • F9. Hierarchical manager with subagents and nested tools
  • F10. Custom extractor without merger returns text fallback
  • F11. Non-reconstructable model object falls back to tree
• Line Wrapper Cases (wrap_llm_line)
  • L1. OpenAI SDK direct call shape preserved
  • L2. OpenRouter via OpenAI SDK object flow
  • L3. OpenAI Responses API automatic shape preservation
  • L4. Multipart content preserved
  • L5. Tool calls passthrough then final wrapped answer
  • L6. Wrapped tools making internal wrapped calls
  • L7. Top agent with two wrapped tools and sub-agents
  • L8. Top-level real tools + nested wrapped llm tools
  • L9. Hierarchical manager with subagents and nested tools
  • L10. Custom extractor without merger returns text fallback
  • L11. Non-reconstructable model object falls back to tree
• Notes
• License

Install

pip install arbis-llmwrap

Public APIs

from llmwrap import wrap_llm_call, wrap_llm_line, openai_sdk_result_text

Required Config Fields

Shared fields (wrap_llm_call and wrap_llm_line)

• company_name: str
  Expects a non-empty company/organization name string.
• project_name: str
  Expects a non-empty project name string.
• agent_name: str
  Expects a non-empty agent/workflow name string.
• secret_key: str
  Expects your wrapper secret key string (store in environment variables).
• max_tries: int = 1
  Expects an integer >= 1 for maximum retry attempts.
• response_extractor: Callable[[Any], str] | None = None
  Expects an optional function that takes raw output and returns answer text (str).
• prompt_json_pointer: str | None = None
  Expects an optional RFC 6901 pointer string to the prompt field in JSON-like payloads (for example "/messages/0/content").
• passthrough_when: Callable[[Any], bool] | None = None
  Expects an optional predicate function; return True to pass raw model output through unchanged.
• return_merger: Callable[[Any, str], Any] | None = None
  Expects an optional merge function (base_output, answer_text) -> Any for custom final output shape.
• response_answer_json_pointer: str | None = None
  Expects an optional RFC 6901 pointer string to where answer text should be written in returned data.

wrap_llm_call specific fields

• prompt_arg: str = "prompt"
  Expects the function argument name that contains the prompt payload.
  Examples: "prompt", "question", "messages".

wrap_llm_line specific fields

• llm_call: Callable[[Any], Any]
  Expects a callable that receives wrapped payload and returns raw model output.
• prompt: Any
  Expects the prompt payload: usually str, or dict/JSON string when using prompt_json_pointer.

Function Wrapper Cases (wrap_llm_call)

F1. OpenAI SDK direct call shape preserved

@wrap_llm_call(
    company_name=CFG.company,
    project_name=CFG.project,
    agent_name="f1_openai_direct",
    secret_key=CFG.key,
    prompt_arg="messages",
    max_tries=1,
)
def one_turn(messages):
    return openai_client.chat.completions.create(model=CFG.model, messages=messages, temperature=0)

F2. OpenRouter via OpenAI SDK object flow

@wrap_llm_call(
    company_name=CFG.company,
    project_name=CFG.project,
    agent_name="f2_openrouter_direct",
    secret_key=CFG.key,
    prompt_arg="messages",
    max_tries=1,
)
def one_turn(messages):
    return openrouter_client.chat.completions.create(model=CFG.openrouter_model, messages=messages, temperature=0)

F3. OpenAI Responses API automatic shape preservation

@wrap_llm_call(
    company_name=CFG.company,
    project_name=CFG.project,
    agent_name="f3_responses_api",
    secret_key=CFG.key,
    prompt_arg="question",
    max_tries=1,
)
def ask(question: str):
    return openai_client.responses.create(model=CFG.model, input=question)

F4. Multipart content preserved

@wrap_llm_call(
    company_name=CFG.company,
    project_name=CFG.project,
    agent_name="f4_multipart_content",
    secret_key=CFG.key,
    prompt_arg="messages",
    max_tries=1,
)
def one_turn(messages):
    return {
        "choices": [{
            "message": {
                "role": "assistant",
                "content": [
                    {"type": "text", "text": "Part A"},
                    {"type": "text", "text": "Part B"},
                ],
            }
        }]
    }

F5. Tool calls passthrough then final wrapped answer

def has_tool_calls(raw):
    try:
        return bool(raw.choices[0].message.tool_calls)
    except Exception:
        return False

@wrap_llm_call(
    company_name=CFG.company,
    project_name=CFG.project,
    agent_name="f5_tool_passthrough",
    secret_key=CFG.key,
    prompt_arg="messages",
    passthrough_when=has_tool_calls,
    max_tries=1,
)
def run_turn(messages):
    return openai_client.chat.completions.create(model=CFG.model, messages=messages, tools=TOOLS, tool_choice="auto")

F6. Wrapped tools making internal wrapped calls

@wrap_llm_call(company_name=CFG.company, project_name=CFG.project, agent_name="f6_tool_a", secret_key=CFG.key, prompt_arg="question", max_tries=1)
def tool_a(question): return openai_client.responses.create(model=CFG.model, input=question)

@wrap_llm_call(company_name=CFG.company, project_name=CFG.project, agent_name="f6_tool_b", secret_key=CFG.key, prompt_arg="question", max_tries=1)
def tool_b(question): return openai_client.responses.create(model=CFG.model, input=question)

@wrap_llm_call(company_name=CFG.company, project_name=CFG.project, agent_name="f6_top_agent", secret_key=CFG.key, prompt_arg="question", max_tries=1)
def top_agent(question): return f"A={tool_a(question)} | B={tool_b(question)}"

F7. Top agent with two wrapped tools and sub-agents

@wrap_llm_call(company_name=CFG.company, project_name=CFG.project, agent_name="f7_tool1_subagent", secret_key=CFG.key, prompt_arg="question", max_tries=1)
def tool1_subagent(question): return openai_client.responses.create(model=CFG.model, input=question)

@wrap_llm_call(company_name=CFG.company, project_name=CFG.project, agent_name="f7_tool2_subagent", secret_key=CFG.key, prompt_arg="question", max_tries=1)
def tool2_subagent(question): return openai_client.responses.create(model=CFG.model, input=question)

@wrap_llm_call(company_name=CFG.company, project_name=CFG.project, agent_name="f7_top_agent", secret_key=CFG.key, prompt_arg="question", max_tries=1)
def top_agent(question): return tool1_subagent(question) + "\n" + tool2_subagent(question)

F8. Top-level real tools + nested wrapped llm tools

def real_weather_tool(city): return {"city": city, "temp_c": 16}
def real_risk_tool(text): return {"risk": "medium", "summary": text}

@wrap_llm_call(company_name=CFG.company, project_name=CFG.project, agent_name="f8_weather_llm", secret_key=CFG.key, prompt_arg="question", max_tries=1)
def weather_llm_tool(question): return openai_client.responses.create(model=CFG.model, input=question)

@wrap_llm_call(company_name=CFG.company, project_name=CFG.project, agent_name="f8_risk_llm", secret_key=CFG.key, prompt_arg="question", max_tries=1)
def risk_llm_tool(question): return openai_client.responses.create(model=CFG.model, input=question)

F9. Hierarchical manager with subagents and nested tools

@wrap_llm_call(company_name=CFG.company, project_name=CFG.project, agent_name="f9_research_subagent", secret_key=CFG.key, prompt_arg="question", max_tries=1)
def research_subagent(question): return openai_client.responses.create(model=CFG.model, input=question)

@wrap_llm_call(company_name=CFG.company, project_name=CFG.project, agent_name="f9_writer_subagent", secret_key=CFG.key, prompt_arg="question", max_tries=1)
def writer_subagent(question): return openai_client.responses.create(model=CFG.model, input=question)

@wrap_llm_call(company_name=CFG.company, project_name=CFG.project, agent_name="f9_manager", secret_key=CFG.key, prompt_arg="question", max_tries=1)
def manager(question): return research_subagent(question) + "\n" + writer_subagent(question)

F10. Custom extractor without merger returns text fallback

@wrap_llm_call(
    company_name=CFG.company,
    project_name=CFG.project,
    agent_name="f10_custom_extractor_text",
    secret_key=CFG.key,
    prompt_arg="messages",
    response_extractor=lambda obj: obj["raw_text"],
    max_tries=1,
)
def one_turn(messages):
    return {"raw_text": "Model content here", "meta": {"id": "abc"}}

F11. Non-reconstructable model object falls back to tree

class NonCopyable:
    def model_dump(self): return {"choices": [{"message": {"role": "assistant", "content": "hello"}}]}
    def __deepcopy__(self, memo): raise RuntimeError("cannot deepcopy")

@wrap_llm_call(
    company_name=CFG.company,
    project_name=CFG.project,
    agent_name="f11_nonreconstructable",
    secret_key=CFG.key,
    prompt_arg="messages",
    max_tries=1,
)
def one_turn(messages): return NonCopyable()

Line Wrapper Cases (wrap_llm_line)

L1. OpenAI SDK direct call shape preserved

payload = {"model": CFG.model, "messages": [{"role": "user", "content": "one sentence"}]}
out = wrap_llm_line(
    llm_call=lambda p: openai_client.chat.completions.create(model=p["model"], messages=p["messages"], temperature=0),
    prompt=payload,
    prompt_json_pointer="/messages/0/content",
    company_name=CFG.company,
    project_name=CFG.project,
    agent_name="l1_openai_direct",
    secret_key=CFG.key,
    max_tries=1,
)

L2. OpenRouter via OpenAI SDK object flow

payload = {"model": CFG.openrouter_model, "messages": [{"role": "user", "content": "one sentence"}]}
out = wrap_llm_line(
    llm_call=lambda p: openrouter_client.chat.completions.create(model=p["model"], messages=p["messages"], temperature=0),
    prompt=payload,
    prompt_json_pointer="/messages/0/content",
    company_name=CFG.company,
    project_name=CFG.project,
    agent_name="l2_openrouter_direct",
    secret_key=CFG.key,
    max_tries=1,
)

L3. OpenAI Responses API automatic shape preservation

out = wrap_llm_line(
    llm_call=lambda prompt: openai_client.responses.create(model=CFG.model, input=prompt),
    prompt="Return one sentence.",
    company_name=CFG.company,
    project_name=CFG.project,
    agent_name="l3_responses_api",
    secret_key=CFG.key,
    max_tries=1,
)

L4. Multipart content preserved

payload = {"messages": [{"role": "user", "content": "multipart"}]}
out = wrap_llm_line(
    llm_call=lambda _p: {"choices": [{"message": {"role": "assistant", "content": [{"type": "text", "text": "Part A"}, {"type": "text", "text": "Part B"}]}}]},
    prompt=payload,
    prompt_json_pointer="/messages/0/content",
    company_name=CFG.company,
    project_name=CFG.project,
    agent_name="l4_multipart_content",
    secret_key=CFG.key,
    max_tries=1,
)

L5. Tool calls passthrough then final wrapped answer

def has_tool_calls(raw):
    try:
        return bool(raw.choices[0].message.tool_calls)
    except Exception:
        return False

out = wrap_llm_line(
    llm_call=lambda p: openai_client.chat.completions.create(model=p["model"], messages=p["messages"], tools=p["tools"], tool_choice="auto"),
    prompt={"model": CFG.model, "messages": MESSAGES, "tools": TOOLS},
    prompt_json_pointer="/messages/1/content",
    passthrough_when=has_tool_calls,
    company_name=CFG.company,
    project_name=CFG.project,
    agent_name="l5_tool_passthrough",
    secret_key=CFG.key,
    max_tries=1,
)

L6. Wrapped tools making internal wrapped calls

def tool_a(question):
    return wrap_llm_line(
        llm_call=lambda q: openai_client.responses.create(model=CFG.model, input=q),
        prompt=question, company_name=CFG.company, project_name=CFG.project,
        agent_name="l6_tool_a", secret_key=CFG.key, max_tries=1
    )

L7. Top agent with two wrapped tools and sub-agents

def top_agent(question):
    a = wrap_llm_line(llm_call=lambda q: openai_client.responses.create(model=CFG.model, input=q), prompt=question, company_name=CFG.company, project_name=CFG.project, agent_name="l7_tool1_subagent", secret_key=CFG.key, max_tries=1)
    b = wrap_llm_line(llm_call=lambda q: openai_client.responses.create(model=CFG.model, input=q), prompt=question, company_name=CFG.company, project_name=CFG.project, agent_name="l7_tool2_subagent", secret_key=CFG.key, max_tries=1)
    return f"{a}\n{b}"

L8. Top-level real tools + nested wrapped llm tools

def weather_llm_tool(question):
    return wrap_llm_line(
        llm_call=lambda q: openai_client.responses.create(model=CFG.model, input=q),
        prompt=question, company_name=CFG.company, project_name=CFG.project,
        agent_name="l8_weather_llm", secret_key=CFG.key, max_tries=1
    )

L9. Hierarchical manager with subagents and nested tools

def manager(question):
    research = wrap_llm_line(llm_call=lambda q: openai_client.responses.create(model=CFG.model, input=q), prompt=question, company_name=CFG.company, project_name=CFG.project, agent_name="l9_research_subagent", secret_key=CFG.key, max_tries=1)
    writing = wrap_llm_line(llm_call=lambda q: openai_client.responses.create(model=CFG.model, input=q), prompt=question, company_name=CFG.company, project_name=CFG.project, agent_name="l9_writer_subagent", secret_key=CFG.key, max_tries=1)
    return f"{research}\n{writing}"

L10. Custom extractor without merger returns text fallback

out = wrap_llm_line(
    llm_call=lambda _prompt: {"raw_text": "model answer", "meta": {"id": "abc"}},
    prompt="hello",
    response_extractor=lambda obj: obj["raw_text"],
    company_name=CFG.company,
    project_name=CFG.project,
    agent_name="l10_custom_extractor_text",
    secret_key=CFG.key,
    max_tries=1,
)

L11. Non-reconstructable model object falls back to tree

class NonCopyable:
    def model_dump(self): return {"choices": [{"message": {"role": "assistant", "content": "hello"}}]}
    def __deepcopy__(self, memo): raise RuntimeError("cannot deepcopy")

out = wrap_llm_line(
    llm_call=lambda _prompt: NonCopyable(),
    prompt="one sentence",
    company_name=CFG.company,
    project_name=CFG.project,
    agent_name="l11_nonreconstructable",
    secret_key=CFG.key,
    max_tries=1,
)

Notes

• Keep credentials in environment variables (.env) and never hardcode production keys.
• Use distinct agent_name values per workflow for clean tracking.
• For custom return shapes, pair response_extractor with response_answer_json_pointer or return_merger when needed.
• The full runnable integrations are in tests/testing_different_interfaces.py.

License

MIT. See LICENSE.