(name:str, state: PipelineState, prompt_template: str, llm)
| 54 | condition: Annotated[bool, lambda x, y: y] |
| 55 | |
| 56 | def execute_step(name:str, state: PipelineState, prompt_template: str, llm) -> PipelineState: |
| 57 | logger(f"{name} is working...") |
| 58 | state["history"] = clip_history(state["history"]) |
| 59 | |
| 60 | generation = create_llm_chain(prompt_template, llm, state["history"]) |
| 61 | data = json.loads(generation) |
| 62 | |
| 63 | state["history"] += "\n" + json.dumps(data) |
| 64 | state["history"] = clip_history(state["history"]) |
| 65 | |
| 66 | logger(state["history"]) |
| 67 | return state |
| 68 | |
| 69 | def execute_tool(name: str, state: PipelineState, prompt_template: str, llm) -> PipelineState: |
| 70 |
no test coverage detected